Literal on-disk index and index write path (apache#9)

This commit contains the following additions
 to SAI:
 - The index write path and index building
   based around StorageAttachedIndexBuilder
   and StorageAttachedIndexWriter
 - The on-disk index versioning using the
   SSTable Descriptor analog IndexDescriptor
   with Version and OnDiskFormat
 - The literal on-disk index using the
   LiteralIndexWriter

patch by Mike Adamson; reviewed by Caleb Rackliffe and Andres de la Peña for CASSANDRA-18062

Co-authored-by: Mike Adamson <mikeatdot@gmail.com>
Co-authored-by: Caleb Rackliffe <calebrackliffe@gmail.com>
Co-authored-by: Andres de la Peña <a.penya.garcia@gmail.com>
Co-authored-by: Piotr Kołaczkowski <pkolaczk@gmail.com>
Co-authored-by: Jason Rutherglen <jason.rutherglen@gmail.com>

Author: 5 people

conf/cassandra.yaml

@@ -1526,6 +1526,12 @@ transparent_data_encryption_options:
           store_type: JCEKS
           key_password: cassandra
 
+# Storage Attached Indexing options.
+# sai_options:
+  ## Total permitted memory allowed for writing SAI index segments. This memory
+  ## is split between all SAI indexes being built so more indexes will mean smaller
+  ## segment sizes.
+  # segment_write_buffer_size: 1024MiB
 
 #####################
 # SAFETY THRESHOLDS #

src/java/org/apache/cassandra/config/CassandraRelevantProperties.java

@@ -418,6 +418,8 @@ public enum CassandraRelevantProperties
     // SAI specific properties
     /** Controls the maximum number of index query intersections that will take part in a query */
     SAI_INTERSECTION_CLAUSE_LIMIT("cassandra.sai.intersection.clause.limit", "2"),
+    /** Latest version to be used for SAI index writing */
+    SAI_LATEST_VERSION("cassandra.sai.latest_version", "aa"),
 
     /** Defines how often schema definitions are pulled from the other nodes */
     SCHEMA_PULL_INTERVAL_MS("cassandra.schema_pull_interval_ms", "60000"),

src/java/org/apache/cassandra/config/Config.java

@@ -777,6 +777,8 @@ public static class SSTableConfig
      */
     public volatile double range_tombstone_list_growth_factor = 1.5;
 
+    public StorageAttachedIndexOptions sai_options = new StorageAttachedIndexOptions();
+
     /**
      * @deprecated migrate to {@link DatabaseDescriptor#isClientInitialized()}
      */

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -967,6 +967,8 @@ else if (conf.max_value_size.toMebibytes() >= 2048)
 
         if (conf.dump_heap_on_uncaught_exception && DatabaseDescriptor.getHeapDumpPath() == null)
             throw new ConfigurationException(String.format("Invalid configuration. Heap dump is enabled but cannot create heap dump output path: %s.", conf.heap_dump_path != null ? conf.heap_dump_path : "null"));
+
+        conf.sai_options.validate();
     }
 
     @VisibleForTesting
@@ -4751,4 +4753,9 @@ public static StorageCompatibilityMode getStorageCompatibilityMode()
         else
             return conf.storage_compatibility_mode;
     }
+
+    public static DataStorageSpec.IntMebibytesBound getSAISegmentWriteBufferSpace()
+    {
+        return conf.sai_options.segment_write_buffer_size;
+    }
 }

src/java/org/apache/cassandra/config/StorageAttachedIndexOptions.java

@@ -0,0 +1,44 @@
+/*
+ * 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.config;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.cassandra.exceptions.ConfigurationException;
+
+public class StorageAttachedIndexOptions
+{
+    public static final int DEFAULT_SEGMENT_BUFFER_MB = 1024;
+
+    @VisibleForTesting
+    public static final int MAXIMUM_SEGMENT_BUFFER_MB = 32768;
+
+    @VisibleForTesting
+    public static final String INVALID_BUFFER_SIZE_ERROR = "Invalid value for segment_write_buffer_size. " +
+                                                           "Value must be a positive integer less than " + MAXIMUM_SEGMENT_BUFFER_MB + "MiB";
+
+    public DataStorageSpec.IntMebibytesBound segment_write_buffer_size = new DataStorageSpec.IntMebibytesBound(DEFAULT_SEGMENT_BUFFER_MB);
+
+    public void validate()
+    {
+        if (segment_write_buffer_size.toMebibytes() > MAXIMUM_SEGMENT_BUFFER_MB)
+        {
+            throw new ConfigurationException(INVALID_BUFFER_SIZE_ERROR);
+        }
+    }
+}

src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java

@@ -48,8 +48,35 @@
 
 public final class CreateIndexStatement extends AlterSchemaStatement
 {
+    public static final String SASI_INDEX_DISABLED = "SASI indexes are disabled. Enable in cassandra.yaml to use.";
+    public static final String KEYSPACE_DOES_NOT_EXIST = "Keyspace '%s' doesn't exist";
+    public static final String TABLE_DOES_NOT_EXIST = "Table '%s' doesn't exist";
+    public static final String COUNTER_TABLES_NOT_SUPPORTED = "Secondary indexes on counter tables aren't supported";
+    public static final String MATERIALIZED_VIEWS_NOT_SUPPORTED = "Secondary indexes on materialized views aren't supported";
+    public static final String TRANSIENTLY_REPLICATED_KEYSPACE_NOT_SUPPORTED = "Secondary indexes are not supported on transiently replicated keyspaces";
+    public static final String CUSTOM_CREATE_WITHOUT_COLUMN = "Only CUSTOM indexes can be created without specifying a target column";
+    public static final String CUSTOM_MULTIPLE_COLUMNS = "Only CUSTOM indexes support multiple columns";
+    public static final String DUPLICATE_TARGET_COLUMN = "Duplicate column '%s' in index target list";
+    public static final String COLUMN_DOES_NOT_EXIST = "Column '%s' doesn't exist";
     public static final String INVALID_CUSTOM_INDEX_TARGET = "Column '%s' is longer than the permissible name length of %d characters or" +
                                                              " contains non-alphanumeric-underscore characters";
+    public static final String COLLECTIONS_WITH_DURATIONS_NOT_SUPPORTED = "Secondary indexes are not supported on collections containing durations";
+    public static final String TUPLES_WITH_DURATIONS_NOT_SUPPORTED = "Secondary indexes are not supported on tuples containing durations";
+    public static final String DURATIONS_NOT_SUPPORTED = "Secondary indexes are not supported on duration columns";
+    public static final String UDTS_WITH_DURATIONS_NOT_SUPPORTED = "Secondary indexes are not supported on UDTs containing durations";
+    public static final String PRIMARY_KEY_IN_COMPACT_STORAGE = "Secondary indexes are not supported on PRIMARY KEY columns in COMPACT STORAGE tables";
+    public static final String COMPACT_COLUMN_IN_COMPACT_STORAGE = "Secondary indexes are not supported on compact value column of COMPACT STORAGE tables";
+    public static final String ONLY_PARTITION_KEY = "Cannot create secondary index on the only partition key column %s";
+    public static final String CREATE_ON_FROZEN_COLUMN = "Cannot create %s() index on frozen column %s. Frozen collections are immutable and must be fully " +
+                                                         "indexed by using the 'full(%s)' modifier";
+    public static final String FULL_ON_FROZEN_COLLECTIONS = "full() indexes can only be created on frozen collections";
+    public static final String NON_COLLECTION_SIMPLE_INDEX = "Cannot create %s() index on %s. Non-collection columns only support simple indexes";
+    public static final String CREATE_WITH_NON_MAP_TYPE = "Cannot create index on %s of column %s with non-map type";
+    public static final String CREATE_ON_NON_FROZEN_UDT = "Cannot create index on non-frozen UDT column %s";
+    public static final String INDEX_ALREADY_EXISTS = "Index '%s' already exists";
+    public static final String INDEX_DUPLICATE_OF_EXISTING = "Index %s is a duplicate of existing index %s";
+    public static final String KEYSPACE_DOES_NOT_MATCH_TABLE = "Keyspace name '%s' doesn't match table name '%s'";
+    public static final String KEYSPACE_DOES_NOT_MATCH_INDEX = "Keyspace name '%s' doesn't match index name '%s'";
 
     private final String indexName;
     private final String tableName;
@@ -90,32 +117,32 @@ public Keyspaces apply(Keyspaces schema)
         Guardrails.createSecondaryIndexesEnabled.ensureEnabled("Creating secondary indexes", state);
 
         if (attrs.isCustom && attrs.customClass.equals(SASIIndex.class.getName()) && !DatabaseDescriptor.getSASIIndexesEnabled())
-            throw new InvalidRequestException("SASI indexes are disabled. Enable in cassandra.yaml to use.");
+            throw new InvalidRequestException(SASI_INDEX_DISABLED);
 
         KeyspaceMetadata keyspace = schema.getNullable(keyspaceName);
         if (null == keyspace)
-            throw ire("Keyspace '%s' doesn't exist", keyspaceName);
+            throw ire(KEYSPACE_DOES_NOT_EXIST, keyspaceName);
 
         TableMetadata table = keyspace.getTableOrViewNullable(tableName);
         if (null == table)
-            throw ire("Table '%s' doesn't exist", tableName);
+            throw ire(TABLE_DOES_NOT_EXIST, tableName);
 
         if (null != indexName && keyspace.hasIndex(indexName))
         {
             if (ifNotExists)
                 return schema;
 
-            throw ire("Index '%s' already exists", indexName);
+            throw ire(INDEX_ALREADY_EXISTS, indexName);
         }
 
         if (table.isCounter())
-            throw ire("Secondary indexes on counter tables aren't supported");
+            throw ire(COUNTER_TABLES_NOT_SUPPORTED);
 
         if (table.isView())
-            throw ire("Secondary indexes on materialized views aren't supported");
+            throw ire(MATERIALIZED_VIEWS_NOT_SUPPORTED);
 
         if (Keyspace.open(table.keyspace).getReplicationStrategy().hasTransientReplicas())
-            throw new InvalidRequestException("Secondary indexes are not supported on transiently replicated keyspaces");
+            throw new InvalidRequestException(TRANSIENTLY_REPLICATED_KEYSPACE_NOT_SUPPORTED);
 
         // guardrails to limit number of secondary indexes per table.
         Guardrails.secondaryIndexesPerTable.guard(table.indexes.size() + 1,
@@ -128,17 +155,17 @@ public Keyspaces apply(Keyspaces schema)
         List<IndexTarget> indexTargets = Lists.newArrayList(transform(rawIndexTargets, t -> t.prepare(table)));
 
         if (indexTargets.isEmpty() && !attrs.isCustom)
-            throw ire("Only CUSTOM indexes can be created without specifying a target column");
+            throw ire(CUSTOM_CREATE_WITHOUT_COLUMN);
 
         if (indexTargets.size() > 1)
         {
             if (!attrs.isCustom)
-                throw ire("Only CUSTOM indexes support multiple columns");
+                throw ire(CUSTOM_MULTIPLE_COLUMNS);
 
             Set<ColumnIdentifier> columns = new HashSet<>();
             for (IndexTarget target : indexTargets)
                 if (!columns.add(target.column))
-                    throw ire("Duplicate column '%s' in index target list", target.column);
+                    throw ire(DUPLICATE_TARGET_COLUMN, target.column);
         }
 
         IndexMetadata.Kind kind = attrs.isCustom ? IndexMetadata.Kind.CUSTOM : IndexMetadata.Kind.COMPOSITES;
@@ -158,7 +185,7 @@ public Keyspaces apply(Keyspaces schema)
             if (ifNotExists)
                 return schema;
 
-            throw ire("Index %s is a duplicate of existing index %s", index.name, equalIndex.name);
+            throw ire(INDEX_DUPLICATE_OF_EXISTING, index.name, equalIndex.name);
         }
 
         TableMetadata newTable = table.withSwapped(table.indexes.with(index));
@@ -181,52 +208,51 @@ private void validateIndexTarget(TableMetadata table, IndexMetadata.Kind kind, I
         ColumnMetadata column = table.getColumn(target.column);
 
         if (null == column)
-            throw ire("Column '%s' doesn't exist", target.column);
+            throw ire(COLUMN_DOES_NOT_EXIST, target.column);
 
         if ((kind == IndexMetadata.Kind.CUSTOM) && !SchemaConstants.isValidName(target.column.toString()))
             throw ire(INVALID_CUSTOM_INDEX_TARGET, target.column, SchemaConstants.NAME_LENGTH);
 
         if (column.type.referencesDuration())
         {
             if (column.type.isCollection())
-                throw ire("Secondary indexes are not supported on collections containing durations");
+                throw ire(COLLECTIONS_WITH_DURATIONS_NOT_SUPPORTED);
 
             if (column.type.isTuple())
-                throw ire("Secondary indexes are not supported on tuples containing durations");
+                throw ire(TUPLES_WITH_DURATIONS_NOT_SUPPORTED);
 
             if (column.type.isUDT())
-                throw  ire("Secondary indexes are not supported on UDTs containing durations");
+                throw  ire(UDTS_WITH_DURATIONS_NOT_SUPPORTED);
 
-            throw ire("Secondary indexes are not supported on duration columns");
+            throw ire(DURATIONS_NOT_SUPPORTED);
         }
 
         if (table.isCompactTable())
         {
             TableMetadata.CompactTableMetadata compactTable = (TableMetadata.CompactTableMetadata) table;
             if (column.isPrimaryKeyColumn())
-                throw new InvalidRequestException("Secondary indexes are not supported on PRIMARY KEY columns in COMPACT STORAGE tables");
+                throw new InvalidRequestException(PRIMARY_KEY_IN_COMPACT_STORAGE);
             if (compactTable.compactValueColumn.equals(column))
-                throw new InvalidRequestException("Secondary indexes are not supported on compact value column of COMPACT STORAGE tables");
+                throw new InvalidRequestException(COMPACT_COLUMN_IN_COMPACT_STORAGE);
         }
 
         if (column.isPartitionKey() && table.partitionKeyColumns().size() == 1)
-            throw ire("Cannot create secondary index on the only partition key column %s", column);
+            throw ire(ONLY_PARTITION_KEY, column);
 
         if (column.type.isFrozenCollection() && target.type != Type.FULL)
-            throw ire("Cannot create %s() index on frozen column %s. Frozen collections are immutable and must be fully " +
-                      "indexed by using the 'full(%s)' modifier", target.type, column, column);
+            throw ire(CREATE_ON_FROZEN_COLUMN, target.type, column, column);
 
         if (!column.type.isFrozenCollection() && target.type == Type.FULL)
-            throw ire("full() indexes can only be created on frozen collections");
+            throw ire(FULL_ON_FROZEN_COLLECTIONS);
 
         if (!column.type.isCollection() && target.type != Type.SIMPLE)
-            throw ire("Cannot create %s() index on %s. Non-collection columns only support simple indexes", target.type, column);
+            throw ire(NON_COLLECTION_SIMPLE_INDEX, target.type, column);
 
         if (!(column.type instanceof MapType && column.type.isMultiCell()) && (target.type == Type.KEYS || target.type == Type.KEYS_AND_VALUES))
-            throw ire("Cannot create index on %s of column %s with non-map type", target.type, column);
+            throw ire(CREATE_WITH_NON_MAP_TYPE, target.type, column);
 
         if (column.type.isUDT() && column.type.isMultiCell())
-            throw ire("Cannot create index on non-frozen UDT column %s", column);
+            throw ire(CREATE_ON_NON_FROZEN_UDT, column);
     }
 
     private String generateIndexName(KeyspaceMetadata keyspace, List<IndexTarget> targets)
@@ -286,10 +312,10 @@ public CreateIndexStatement prepare(ClientState state)
                                 : indexName.hasKeyspace() ? indexName.getKeyspace() : state.getKeyspace();
 
             if (tableName.hasKeyspace() && !keyspaceName.equals(tableName.getKeyspace()))
-                throw ire("Keyspace name '%s' doesn't match table name '%s'", keyspaceName, tableName);
+                throw ire(KEYSPACE_DOES_NOT_MATCH_TABLE, keyspaceName, tableName);
 
             if (indexName.hasKeyspace() && !keyspaceName.equals(indexName.getKeyspace()))
-                throw ire("Keyspace name '%s' doesn't match index name '%s'", keyspaceName, tableName);
+                throw ire(KEYSPACE_DOES_NOT_MATCH_INDEX, keyspaceName, tableName);
 
             return new CreateIndexStatement(keyspaceName, tableName.getName(), indexName.getName(), rawIndexTargets, attrs, ifNotExists);
         }

src/java/org/apache/cassandra/db/ReadCommand.java

@@ -411,7 +411,6 @@ public UnfilteredPartitionIterator executeLocally(ReadExecutionController execut
             {
                 cfs.indexManager.checkQueryability(indexQueryPlan);
 
-                Index index = indexQueryPlan.getFirst();
                 searcher = indexQueryPlan.searcherFor(this);
                 Tracing.trace("Executing read on {}.{} using index{} {}",
                               cfs.metadata.keyspace,

src/java/org/apache/cassandra/db/lifecycle/Tracker.java

@@ -217,6 +217,12 @@ Throwable updateSizeTracking(Iterable<SSTableReader> oldSSTables, Iterable<SSTab
         return accumulate;
     }
 
+    public void updateLiveDiskSpaceUsed(long adjustment)
+    {
+        cfstore.metric.liveDiskSpaceUsed.inc(adjustment);
+        cfstore.metric.totalDiskSpaceUsed.inc(adjustment);
+    }
+
     // SETUP / CLEANUP
 
     public void addInitialSSTables(Iterable<SSTableReader> sstables)

src/java/org/apache/cassandra/db/memtable/AbstractMemtable.java

@@ -30,6 +30,7 @@
 
 import org.apache.cassandra.db.RegularAndStaticColumns;
 import org.apache.cassandra.db.commitlog.CommitLogPosition;
+import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
 import org.apache.cassandra.db.partitions.Partition;
 import org.apache.cassandra.db.rows.EncodingStats;
 import org.apache.cassandra.schema.ColumnMetadata;
@@ -39,6 +40,7 @@
 
 public abstract class AbstractMemtable implements Memtable
 {
+    private final AtomicReference<LifecycleTransaction> flushTransaction = new AtomicReference<>(null);
     protected final AtomicLong currentOperations = new AtomicLong(0);
     protected final ColumnsCollector columnsCollector;
     protected final StatsCollector statsCollector = new StatsCollector();
@@ -66,21 +68,25 @@ public AbstractMemtable(TableMetadataRef metadataRef, long minTimestamp)
         this.minTimestamp = new AtomicLong(minTimestamp);
     }
 
+    @Override
     public TableMetadata metadata()
     {
         return metadata.get();
     }
 
+    @Override
     public long operationCount()
     {
         return currentOperations.get();
     }
 
+    @Override
     public long getMinTimestamp()
     {
         return minTimestamp.get() != EncodingStats.NO_STATS.minTimestamp ? minTimestamp.get() : NO_MIN_TIMESTAMP;
     }
 
+    @Override
     public long getMinLocalDeletionTime()
     {
         return minLocalDeletionTime.get();
@@ -120,6 +126,18 @@ EncodingStats encodingStats()
         return statsCollector.get();
     }
 
+    @Override
+    public LifecycleTransaction getFlushTransaction()
+    {
+        return flushTransaction.get();
+    }
+
+    @Override
+    public LifecycleTransaction setFlushTransaction(LifecycleTransaction flushTransaction)
+    {
+        return this.flushTransaction.getAndSet(flushTransaction);
+    }
+
     protected static class ColumnsCollector
     {
         private final HashMap<ColumnMetadata, AtomicBoolean> predefined = new HashMap<>();
@@ -205,26 +223,31 @@ public EncodingStats get()
 
     protected abstract class AbstractFlushablePartitionSet<P extends Partition> implements FlushablePartitionSet<P>
     {
+        @Override
         public long dataSize()
         {
             return getLiveDataSize();
         }
 
+        @Override
         public CommitLogPosition commitLogLowerBound()
         {
             return AbstractMemtable.this.getCommitLogLowerBound();
         }
 
+        @Override
         public LastCommitLogPosition commitLogUpperBound()
         {
             return AbstractMemtable.this.getFinalCommitLogUpperBound();
         }
 
+        @Override
         public EncodingStats encodingStats()
         {
             return AbstractMemtable.this.encodingStats();
         }
 
+        @Override
         public RegularAndStaticColumns columns()
         {
             return AbstractMemtable.this.columns();