Consolidate class hierarchy around CassandraIndex

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

@@ -57,7 +57,7 @@
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.index.SecondaryIndexManager;
-import org.apache.cassandra.index.internal.ColumnIndexFunctions;
+import org.apache.cassandra.index.internal.CassandraIndex;
 import org.apache.cassandra.index.transactions.UpdateTransaction;
 import org.apache.cassandra.io.FSWriteError;
 import org.apache.cassandra.io.sstable.Component;
@@ -576,8 +576,7 @@ public static void scrubDataDirectories(CFMetaData metadata)
         for (IndexMetadata index : metadata.getIndexes())
             if (!index.isCustom())
             {
-                CFMetaData indexMetadata = ColumnIndexFunctions.getFunctions(metadata, index)
-                                                               .indexCfsMetadata(metadata, index);
+                CFMetaData indexMetadata = CassandraIndex.indexCfsMetadata(metadata, index);
                 scrubDataDirectories(indexMetadata);
             }
     }

src/java/org/apache/cassandra/index/SecondaryIndexManager.java

@@ -293,7 +293,7 @@ private Index createInstance(IndexMetadata indexDef)
         }
         else
         {
-            newIndex = new CassandraIndex(baseCfs, indexDef);
+            newIndex = CassandraIndex.newIndex(baseCfs, indexDef);
         }
         return newIndex;
     }

src/java/org/apache/cassandra/index/internal/CassandraIndexFunctions.java

@@ -0,0 +1,186 @@
+/*
+ * 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.index.internal;
+
+import java.util.List;
+
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.CollectionType;
+import org.apache.cassandra.db.marshal.CompositeType;
+import org.apache.cassandra.index.internal.composites.*;
+import org.apache.cassandra.index.internal.keys.KeysIndex;
+import org.apache.cassandra.schema.IndexMetadata;
+
+public interface CassandraIndexFunctions
+{
+    /**
+     *
+     * @param baseCfs
+     * @param indexMetadata
+     * @return
+     */
+    public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata);
+
+    /**
+     * Returns the type of the the values in the index. For most columns this is simply its type, but for collections
+     * it depends on whether the index is on the collection name/value element or on a frozen collection
+     * @param indexedColumn
+     * @return
+     */
+    default AbstractType<?> getIndexedValueType(ColumnDefinition indexedColumn)
+    {
+        return indexedColumn.type;
+    }
+
+    /**
+     * Add the clustering columns for a specific type of index table to the a CFMetaData.Builder (which is being
+     * used to construct the index table's CFMetadata. In the default implementation, the clustering columns of the
+     * index table hold the partition key & clustering columns of the base table. This is overridden in several cases:
+     * * When the indexed value is itself a clustering column, in which case, we only need store the base table's
+     *   *other* clustering values in the index - the indexed value being the index table's partition key
+     * * When the indexed value is a collection value, in which case we also need to capture the cell path from the base
+     *   table
+     * * In a KEYS index (for thrift/compact storage/static column indexes), where only the base partition key is
+     *   held in the index table.
+     *
+     * Called from indexCfsMetadata
+     * @param builder
+     * @param baseMetadata
+     * @param cfDef
+     * @return
+     */
+    default CFMetaData.Builder addIndexClusteringColumns(CFMetaData.Builder builder,
+                                                         CFMetaData baseMetadata,
+                                                         ColumnDefinition cfDef)
+    {
+        for (ColumnDefinition def : baseMetadata.clusteringColumns())
+            builder.addClusteringColumn(def.name, def.type);
+        return builder;
+    }
+
+    /*
+     * implementations providing specializations for the built in index types
+     */
+
+    static final CassandraIndexFunctions KEYS_INDEX_FUNCTIONS = new CassandraIndexFunctions()
+    {
+        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+        {
+            return new KeysIndex(baseCfs, indexMetadata);
+        }
+    };
+
+    static final CassandraIndexFunctions REGULAR_COLUMN_INDEX_FUNCTIONS = new CassandraIndexFunctions()
+    {
+        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+        {
+            return new RegularColumnIndex(baseCfs, indexMetadata);
+        }
+    };
+
+    static final CassandraIndexFunctions CLUSTERING_COLUMN_INDEX_FUNCTIONS = new CassandraIndexFunctions()
+    {
+        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+        {
+            return new ClusteringColumnIndex(baseCfs, indexMetadata);
+        }
+
+        public CFMetaData.Builder addIndexClusteringColumns(CFMetaData.Builder builder,
+                                                            CFMetaData baseMetadata,
+                                                            ColumnDefinition columnDef)
+        {
+            List<ColumnDefinition> cks = baseMetadata.clusteringColumns();
+            for (int i = 0; i < columnDef.position(); i++)
+            {
+                ColumnDefinition def = cks.get(i);
+                builder.addClusteringColumn(def.name, def.type);
+            }
+            for (int i = columnDef.position() + 1; i < cks.size(); i++)
+            {
+                ColumnDefinition def = cks.get(i);
+                builder.addClusteringColumn(def.name, def.type);
+            }
+            return builder;
+        }
+    };
+
+    static final CassandraIndexFunctions PARTITION_KEY_INDEX_FUNCTIONS = new CassandraIndexFunctions()
+    {
+        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+        {
+            return new PartitionKeyIndex(baseCfs, indexMetadata);
+        }
+    };
+
+    static final CassandraIndexFunctions COLLECTION_KEY_INDEX_FUNCTIONS = new CassandraIndexFunctions()
+    {
+        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+        {
+            return new CollectionKeyIndex(baseCfs, indexMetadata);
+        }
+
+        public AbstractType<?> getIndexedValueType(ColumnDefinition indexedColumn)
+        {
+            return ((CollectionType) indexedColumn.type).nameComparator();
+        }
+    };
+
+    static final CassandraIndexFunctions COLLECTION_VALUE_INDEX_FUNCTIONS = new CassandraIndexFunctions()
+    {
+
+        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+        {
+            return new CollectionValueIndex(baseCfs, indexMetadata);
+        }
+
+        public AbstractType<?> getIndexedValueType(ColumnDefinition indexedColumn)
+        {
+            return ((CollectionType)indexedColumn.type).valueComparator();
+        }
+
+        public CFMetaData.Builder addIndexClusteringColumns(CFMetaData.Builder builder,
+                                                            CFMetaData baseMetadata,
+                                                            ColumnDefinition columnDef)
+        {
+            for (ColumnDefinition def : baseMetadata.clusteringColumns())
+                builder.addClusteringColumn(def.name, def.type);
+
+            // collection key
+            builder.addClusteringColumn("cell_path", ((CollectionType)columnDef.type).nameComparator());
+            return builder;
+        }
+    };
+
+    static final CassandraIndexFunctions COLLECTION_ENTRY_INDEX_FUNCTIONS = new CassandraIndexFunctions()
+    {
+        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+        {
+            return new CollectionEntryIndex(baseCfs, indexMetadata);
+        }
+
+        public AbstractType<?> getIndexedValueType(ColumnDefinition indexedColumn)
+        {
+            CollectionType colType = (CollectionType)indexedColumn.type;
+            return CompositeType.getInstance(colType.nameComparator(), colType.valueComparator());
+        }
+    };
+}