<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.cassandra.db;

import java.io.DataInputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.util.*;
import java.util.concurrent.ExecutionException;

import com.google.common.collect.HashMultimap;
import com.google.common.collect.SetMultimap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.Constants;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.CounterId;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;

import static org.apache.cassandra.cql3.QueryProcessor.processInternal;

public class SystemTable
{
    private static final Logger logger = LoggerFactory.getLogger(SystemTable.class);

    // see CFMetaData for schema definitions
    public static final String PEERS_CF = "peers";
    public static final String PEER_EVENTS_CF = "peer_events";
    public static final String LOCAL_CF = "local";
    public static final String INDEX_CF = "IndexInfo";
    public static final String COUNTER_ID_CF = "NodeIdInfo";
    public static final String HINTS_CF = "hints";
    public static final String RANGE_XFERS_CF = "range_xfers";
    public static final String BATCHLOG_CF = "batchlog";
    // see layout description in the DefsTable class header
    public static final String SCHEMA_KEYSPACES_CF = "schema_keyspaces";
    public static final String SCHEMA_COLUMNFAMILIES_CF = "schema_columnfamilies";
    public static final String SCHEMA_COLUMNS_CF = "schema_columns";

    @Deprecated
    public static final String OLD_STATUS_CF = "LocationInfo";
    @Deprecated
    public static final String OLD_HINTS_CF = "HintsColumnFamily";

    private static final String LOCAL_KEY = "local";
    private static final ByteBuffer ALL_LOCAL_NODE_ID_KEY = ByteBufferUtil.bytes("Local");

    public enum BootstrapState
    {
        NEEDS_BOOTSTRAP,
        COMPLETED,
        IN_PROGRESS
    }

    private static DecoratedKey decorate(ByteBuffer key)
    {
        return StorageService.getPartitioner().decorateKey(key);
    }

    public static void finishStartup()
    {
        DefsTable.fixSchemaNanoTimestamps();
        setupVersion();
        try
        {
            upgradeSystemData();
        }
        catch (ExecutionException e)
        {
            throw new RuntimeException(e);
        }
        catch (InterruptedException e)
        {
            throw new RuntimeException(e);
        }

        // add entries to system schema columnfamilies for the hardcoded system definitions
        for (String ksname : Schema.systemKeyspaceNames)
        {
            KSMetaData ksmd = Schema.instance.getKSMetaData(ksname);

            // delete old, possibly obsolete entries in schema columnfamilies
            for (String cfname : Arrays.asList(SystemTable.SCHEMA_KEYSPACES_CF, SystemTable.SCHEMA_COLUMNFAMILIES_CF, SystemTable.SCHEMA_COLUMNS_CF))
            {
                String req = String.format("DELETE FROM system.%s WHERE keyspace_name = '%s'", cfname, ksmd.name);
                processInternal(req);
            }

            // (+1 to timestamp to make sure we don't get shadowed by the tombstones we just added)
            ksmd.toSchema(FBUtilities.timestampMicros() + 1).apply();
        }
    }

    private static void setupVersion()
    {
        String req = "INSERT INTO system.%s (key, release_version, cql_version, thrift_version, data_center, rack, partitioner) VALUES ('%s', '%s', '%s', '%s', '%s', '%s', '%s')";
        IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
        processInternal(String.format(req, LOCAL_CF,
                                         LOCAL_KEY,
                                         FBUtilities.getReleaseVersionString(),
                                         QueryProcessor.CQL_VERSION.toString(),
                                         Constants.VERSION,
                                         snitch.getDatacenter(FBUtilities.getBroadcastAddress()),
                                         snitch.getRack(FBUtilities.getBroadcastAddress()),
                                         DatabaseDescriptor.getPartitioner().getClass().getName()));
    }

    /** if system data becomes incompatible across versions of cassandra, that logic (and associated purging) is managed here */
    private static void upgradeSystemData() throws ExecutionException, InterruptedException
    {
        Table table = Table.open(Table.SYSTEM_KS);
        ColumnFamilyStore oldStatusCfs = table.getColumnFamilyStore(OLD_STATUS_CF);
        if (oldStatusCfs.getSSTables().size() > 0)
        {
            SortedSet<ByteBuffer> cols = new TreeSet<ByteBuffer>(BytesType.instance);
            cols.add(ByteBufferUtil.bytes("ClusterName"));
            cols.add(ByteBufferUtil.bytes("Token"));
            QueryFilter filter = QueryFilter.getNamesFilter(decorate(ByteBufferUtil.bytes("L")), new QueryPath(OLD_STATUS_CF), cols);
            ColumnFamily oldCf = oldStatusCfs.getColumnFamily(filter);
            Iterator<IColumn> oldColumns = oldCf.columns.iterator();

            String clusterName = null;
            try
            {
                clusterName = ByteBufferUtil.string(oldColumns.next().value());
            }
            catch (CharacterCodingException e)
            {
                throw new RuntimeException(e);
            }
            // serialize the old token as a collection of (one )tokens.
            Token token = StorageService.getPartitioner().getTokenFactory().fromByteArray(oldColumns.next().value());
            String tokenBytes = tokensAsSet(Collections.singleton(token));
            // (assume that any node getting upgraded was bootstrapped, since that was stored in a separate row for no particular reason)
            String req = "INSERT INTO system.%s (key, cluster_name, tokens, bootstrapped) VALUES ('%s', '%s', %s, '%s')";
            processInternal(String.format(req, LOCAL_CF, LOCAL_KEY, clusterName, tokenBytes, BootstrapState.COMPLETED.name()));

            oldStatusCfs.truncate();
        }

        ColumnFamilyStore oldHintsCfs = table.getColumnFamilyStore(OLD_HINTS_CF);
        if (oldHintsCfs.getSSTables().size() > 0)
        {
            logger.info("Possible old-format hints found. Truncating");
            oldHintsCfs.truncate();
        }
    }

    public static void saveTruncationRecord(ColumnFamilyStore cfs, long truncatedAt, ReplayPosition position)
    {
        String req = "UPDATE system.%s SET truncated_at = truncated_at + %s WHERE key = '%s'";
        processInternal(String.format(req, LOCAL_CF, truncationAsMapEntry(cfs, truncatedAt, position), LOCAL_KEY));
        forceBlockingFlush(LOCAL_CF);
    }

    /**
     * This method is used to remove information about truncation time for specified column family
     */
    public static void removeTruncationRecord(UUID cfId)
    {
        String req = "DELETE truncated_at['%s'] from system.%s WHERE key = '%s'";
        processInternal(String.format(req, cfId, LOCAL_CF, LOCAL_KEY));
        forceBlockingFlush(LOCAL_CF);
    }

    private static String truncationAsMapEntry(ColumnFamilyStore cfs, long truncatedAt, ReplayPosition position)
    {
        DataOutputBuffer out = new DataOutputBuffer();
        try
        {
            ReplayPosition.serializer.serialize(position, out);
            out.writeLong(truncatedAt);
        }
        catch (IOException e)
        {
            throw new RuntimeException(e);
        }
        return String.format("{%s: 0x%s}",
                             cfs.metadata.cfId,
                             ByteBufferUtil.bytesToHex(ByteBuffer.wrap(out.getData(), 0, out.getLength())));
    }

    public static Map<UUID, Pair<ReplayPosition, Long>> getTruncationRecords()
    {
        String req = "SELECT truncated_at FROM system.%s WHERE key = '%s'";
        UntypedResultSet rows = processInternal(String.format(req, LOCAL_CF, LOCAL_KEY));<commit_msg>unquote uuid in cql string<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.cassandra.db;

import java.io.DataInputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.util.*;
import java.util.concurrent.ExecutionException;

import com.google.common.collect.HashMultimap;
import com.google.common.collect.SetMultimap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.Constants;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.CounterId;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;

import static org.apache.cassandra.cql3.QueryProcessor.processInternal;

public class SystemTable
{
    private static final Logger logger = LoggerFactory.getLogger(SystemTable.class);

    // see CFMetaData for schema definitions
    public static final String PEERS_CF = "peers";
    public static final String PEER_EVENTS_CF = "peer_events";
    public static final String LOCAL_CF = "local";
    public static final String INDEX_CF = "IndexInfo";
    public static final String COUNTER_ID_CF = "NodeIdInfo";
    public static final String HINTS_CF = "hints";
    public static final String RANGE_XFERS_CF = "range_xfers";
    public static final String BATCHLOG_CF = "batchlog";
    // see layout description in the DefsTable class header
    public static final String SCHEMA_KEYSPACES_CF = "schema_keyspaces";
    public static final String SCHEMA_COLUMNFAMILIES_CF = "schema_columnfamilies";
    public static final String SCHEMA_COLUMNS_CF = "schema_columns";

    @Deprecated
    public static final String OLD_STATUS_CF = "LocationInfo";
    @Deprecated
    public static final String OLD_HINTS_CF = "HintsColumnFamily";

    private static final String LOCAL_KEY = "local";
    private static final ByteBuffer ALL_LOCAL_NODE_ID_KEY = ByteBufferUtil.bytes("Local");

    public enum BootstrapState
    {
        NEEDS_BOOTSTRAP,
        COMPLETED,
        IN_PROGRESS
    }

    private static DecoratedKey decorate(ByteBuffer key)
    {
        return StorageService.getPartitioner().decorateKey(key);
    }

    public static void finishStartup()
    {
        DefsTable.fixSchemaNanoTimestamps();
        setupVersion();
        try
        {
            upgradeSystemData();
        }
        catch (ExecutionException e)
        {
            throw new RuntimeException(e);
        }
        catch (InterruptedException e)
        {
            throw new RuntimeException(e);
        }

        // add entries to system schema columnfamilies for the hardcoded system definitions
        for (String ksname : Schema.systemKeyspaceNames)
        {
            KSMetaData ksmd = Schema.instance.getKSMetaData(ksname);

            // delete old, possibly obsolete entries in schema columnfamilies
            for (String cfname : Arrays.asList(SystemTable.SCHEMA_KEYSPACES_CF, SystemTable.SCHEMA_COLUMNFAMILIES_CF, SystemTable.SCHEMA_COLUMNS_CF))
            {
                String req = String.format("DELETE FROM system.%s WHERE keyspace_name = '%s'", cfname, ksmd.name);
                processInternal(req);
            }

            // (+1 to timestamp to make sure we don't get shadowed by the tombstones we just added)
            ksmd.toSchema(FBUtilities.timestampMicros() + 1).apply();
        }
    }

    private static void setupVersion()
    {
        String req = "INSERT INTO system.%s (key, release_version, cql_version, thrift_version, data_center, rack, partitioner) VALUES ('%s', '%s', '%s', '%s', '%s', '%s', '%s')";
        IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
        processInternal(String.format(req, LOCAL_CF,
                                         LOCAL_KEY,
                                         FBUtilities.getReleaseVersionString(),
                                         QueryProcessor.CQL_VERSION.toString(),
                                         Constants.VERSION,
                                         snitch.getDatacenter(FBUtilities.getBroadcastAddress()),
                                         snitch.getRack(FBUtilities.getBroadcastAddress()),
                                         DatabaseDescriptor.getPartitioner().getClass().getName()));
    }

    /** if system data becomes incompatible across versions of cassandra, that logic (and associated purging) is managed here */
    private static void upgradeSystemData() throws ExecutionException, InterruptedException
    {
        Table table = Table.open(Table.SYSTEM_KS);
        ColumnFamilyStore oldStatusCfs = table.getColumnFamilyStore(OLD_STATUS_CF);
        if (oldStatusCfs.getSSTables().size() > 0)
        {
            SortedSet<ByteBuffer> cols = new TreeSet<ByteBuffer>(BytesType.instance);
            cols.add(ByteBufferUtil.bytes("ClusterName"));
            cols.add(ByteBufferUtil.bytes("Token"));
            QueryFilter filter = QueryFilter.getNamesFilter(decorate(ByteBufferUtil.bytes("L")), new QueryPath(OLD_STATUS_CF), cols);
            ColumnFamily oldCf = oldStatusCfs.getColumnFamily(filter);
            Iterator<IColumn> oldColumns = oldCf.columns.iterator();

            String clusterName = null;
            try
            {
                clusterName = ByteBufferUtil.string(oldColumns.next().value());
            }
            catch (CharacterCodingException e)
            {
                throw new RuntimeException(e);
            }
            // serialize the old token as a collection of (one )tokens.
            Token token = StorageService.getPartitioner().getTokenFactory().fromByteArray(oldColumns.next().value());
            String tokenBytes = tokensAsSet(Collections.singleton(token));
            // (assume that any node getting upgraded was bootstrapped, since that was stored in a separate row for no particular reason)
            String req = "INSERT INTO system.%s (key, cluster_name, tokens, bootstrapped) VALUES ('%s', '%s', %s, '%s')";
            processInternal(String.format(req, LOCAL_CF, LOCAL_KEY, clusterName, tokenBytes, BootstrapState.COMPLETED.name()));

            oldStatusCfs.truncate();
        }

        ColumnFamilyStore oldHintsCfs = table.getColumnFamilyStore(OLD_HINTS_CF);
        if (oldHintsCfs.getSSTables().size() > 0)
        {
            logger.info("Possible old-format hints found. Truncating");
            oldHintsCfs.truncate();
        }
    }

    public static void saveTruncationRecord(ColumnFamilyStore cfs, long truncatedAt, ReplayPosition position)
    {
        String req = "UPDATE system.%s SET truncated_at = truncated_at + %s WHERE key = '%s'";
        processInternal(String.format(req, LOCAL_CF, truncationAsMapEntry(cfs, truncatedAt, position), LOCAL_KEY));
        forceBlockingFlush(LOCAL_CF);
    }

    /**
     * This method is used to remove information about truncation time for specified column family
     */
    public static void removeTruncationRecord(UUID cfId)
    {
        String req = "DELETE truncated_at[%s] from system.%s WHERE key = '%s'";
        processInternal(String.format(req, cfId, LOCAL_CF, LOCAL_KEY));
        forceBlockingFlush(LOCAL_CF);
    }

    private static String truncationAsMapEntry(ColumnFamilyStore cfs, long truncatedAt, ReplayPosition position)
    {
        DataOutputBuffer out = new DataOutputBuffer();
        try
        {
            ReplayPosition.serializer.serialize(position, out);
            out.writeLong(truncatedAt);
        }
        catch (IOException e)
        {
            throw new RuntimeException(e);
        }
        return String.format("{%s: 0x%s}",
                             cfs.metadata.cfId,
                             ByteBufferUtil.bytesToHex(ByteBuffer.wrap(out.getData(), 0, out.getLength())));
    }

    public static Map<UUID, Pair<ReplayPosition, Long>> getTruncationRecords()
    {
        String req = "SELECT truncated_at FROM system.%s WHERE key = '%s'";
        UntypedResultSet rows = processInternal(String.format(req, LOCAL_CF, LOCAL_KEY));
{'fraction_non_alphanumeric': 0.08972443986608293, 'fraction_numerical': 0.001545197012619109, 'mean_word_length': 2.9074260414570334, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7385058', 'n_tokens_mistral': 5369, 'n_tokens_neox': 5235, 'n_words': 1339}