content: /*
   *
   * 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;
  
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.Callable;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.function.Predicate;
  import java.util.function.Supplier;
  import java.util.stream.Collectors;
  
  import com.google.common.collect.ImmutableList;
  import com.google.common.collect.ImmutableSet;
  import com.google.common.collect.Maps;
  import com.google.common.collect.Sets;
  import org.junit.Assume;
  import org.junit.BeforeClass;
  import com.google.common.util.concurrent.Uninterruptibles;
  import org.junit.Assert;
  import org.junit.Test;
  
  import com.datastax.driver.core.exceptions.QueryValidationException;
  import org.apache.cassandra.Util;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.cql3.CQLTester;
  import org.apache.cassandra.cql3.ColumnIdentifier;
  import org.apache.cassandra.cql3.Operator;
  import org.apache.cassandra.cql3.restrictions.IndexRestrictions;
  import org.apache.cassandra.cql3.restrictions.StatementRestrictions;
  import org.apache.cassandra.cql3.statements.ModificationStatement;
  import org.apache.cassandra.db.ColumnFamilyStore.FlushReason;
  import org.apache.cassandra.db.filter.RowFilter;
  import org.apache.cassandra.cql3.statements.schema.IndexTarget;
  import org.apache.cassandra.db.CassandraWriteContext;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.db.DecoratedKey;
  import org.apache.cassandra.db.DeletionTime;
  import org.apache.cassandra.db.Keyspace;
  import org.apache.cassandra.db.RangeTombstone;
  import org.apache.cassandra.db.ReadCommand;
  import org.apache.cassandra.db.ReadExecutionController;
  import org.apache.cassandra.db.RegularAndStaticColumns;
  import org.apache.cassandra.db.WriteContext;
  import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
  import org.apache.cassandra.db.marshal.AbstractType;
  import org.apache.cassandra.db.marshal.Int32Type;
  import org.apache.cassandra.db.marshal.UTF8Type;
  import org.apache.cassandra.db.memtable.Memtable;
  import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
  import org.apache.cassandra.db.rows.Row;
  import org.apache.cassandra.db.rows.Unfiltered;
  import org.apache.cassandra.exceptions.InvalidRequestException;
  import org.apache.cassandra.index.internal.CassandraIndex;
  import org.apache.cassandra.index.transactions.IndexTransaction;
  import org.apache.cassandra.io.sstable.Component;
  import org.apache.cassandra.io.sstable.Descriptor;
  import org.apache.cassandra.io.sstable.SSTableFlushObserver;
  import org.apache.cassandra.schema.ColumnMetadata;
  import org.apache.cassandra.schema.IndexMetadata;
  import org.apache.cassandra.schema.Indexes;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.transport.ProtocolVersion;
  import org.apache.cassandra.utils.ByteBufferUtil;
  import org.apache.cassandra.utils.FBUtilities;
  import org.apache.cassandra.utils.concurrent.OpOrder;
  
  import static org.apache.cassandra.cql3.statements.schema.IndexTarget.CUSTOM_INDEX_OPTION_NAME;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertNotNull;
  import static org.junit.Assert.assertTrue;
  import static org.junit.Assert.fail;
  
  public class CustomIndexTest extends CQLTester
  {
      @BeforeClass
      public static void setUpClass() // overrides CQLTester.setUpClass()
      {
          // Accord breaks indexBuildingPagesLargePartitions because it introduces blocking OpOrder.Group
          // when it sees the schema change and forces a flush of the Accord keyspace topologies table
          // which creates a blocking OpOrder.Group.
          // The test is explicitly trying to assert none of the created groups are blocking and that is pretty
          // fragile as implemented since any background things could create mark a group blocking becuase Keyspace.writeOrder
          // is global
          CQLTester.daemonInitialization();
          DatabaseDescriptor.setAccordTransactionsEnabled(false);
          CQLTester.setUpClass();
      }
  
      @Test
      public void testInsertsOnCfsBackedIndex() throws Throwable
      {
          // test to ensure that we don't deadlock when flushing CFS backed custom indexers
          // see CASSANDRA-10181
          createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
          createIndex("CREATE CUSTOM INDEX ON %s(c) USING 'org.apache.cassandra.index.internal.CustomCassandraIndex'");
  
          execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 0, 0, 0, 2);
          execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 0, 1, 0, 1);
          execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 0, 2, 0, 0);
      }
  
      @Test
      public void testTruncateWithNonCfsCustomIndex() throws Throwable
      {
          // deadlocks and times out the test in the face of the synchronisation
          // issues described in the comments on CASSANDRA-9669
          createTable("CREATE TABLE %s (a int, b int, c int, PRIMARY KEY (a))");
          createIndex("CREATE CUSTOM INDEX ON %s(b) USING 'org.apache.cassandra.index.StubIndex'");
          execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 0, 1, 2);
          getCurrentColumnFamilyStore().truncateBlocking();
      }
  
      @Test
      public void indexControlsIfIncludedInBuildOnNewSSTables() throws Throwable
      {
          createTable("CREATE TABLE %s (a int, b int, PRIMARY KEY (a))");
          String toInclude = "include";
          String toExclude = "exclude";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(b) USING '%s'",
                                    toInclude, IndexIncludedInBuild.class.getName()));
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(b) USING '%s'",
                                    toExclude, IndexExcludedFromBuild.class.getName()));
  
          execute("INSERT INTO %s (a, b) VALUES (?, ?)", 0, 0);
          execute("INSERT INTO %s (a, b) VALUES (?, ?)", 1, 1);
          execute("INSERT INTO %s (a, b) VALUES (?, ?)", 2, 2);
          flush();
  
          SecondaryIndexManager indexManager = getCurrentColumnFamilyStore().indexManager;
          IndexIncludedInBuild included = (IndexIncludedInBuild)indexManager.getIndexByName(toInclude);
          included.reset();
          assertTrue(included.rowsInserted.isEmpty());
  
          IndexExcludedFromBuild excluded = (IndexExcludedFromBuild)indexManager.getIndexByName(toExclude);
          excluded.reset();
          assertTrue(excluded.rowsInserted.isEmpty());
  
          indexManager.rebuildIndexesBlocking(Sets.newHashSet(toInclude, toExclude));
  
          assertEquals(3, included.rowsInserted.size());
          assertTrue(excluded.rowsInserted.isEmpty());
      }
  
      @Test
      public void indexReceivesWriteTimeDeletionsCorrectly() throws Throwable
      {
          createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b, c))");
          String indexName = "test_index";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(d) USING '%s'",
                                    indexName, StubIndex.class.getName()));
  
          execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 0, 0, 0, 0);
          execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 0, 0, 1, 1);
          execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 0, 0, 2, 2);
          execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 0, 1, 3, 3);
  
          SecondaryIndexManager indexManager = getCurrentColumnFamilyStore().indexManager;
          StubIndex index = (StubIndex)indexManager.getIndexByName(indexName);
          assertEquals(4, index.rowsInserted.size());
          assertTrue(index.partitionDeletions.isEmpty());
          assertTrue(index.rangeTombstones.isEmpty());
  
          execute("DELETE FROM %s WHERE a=0 AND b=0");
          assertTrue(index.partitionDeletions.isEmpty());
          assertEquals(1, index.rangeTombstones.size());
  
          execute("DELETE FROM %s WHERE a=0");
          assertEquals(1, index.partitionDeletions.size());
          assertEquals(1, index.rangeTombstones.size());
      }
      @Test
      public void nonCustomIndexesRequireExactlyOneTargetColumn() throws Throwable
      {
          Util.assumeLegacySecondaryIndex();
  
          createTable("CREATE TABLE %s(k int, c int, v1 int, v2 int, PRIMARY KEY (k,c))");
  
          assertInvalidMessage("Only CUSTOM indexes support multiple columns", "CREATE INDEX multi_idx on %s(v1,v2)");
          assertInvalidMessage("Only CUSTOM indexes can be created without specifying a target column",
                             "CREATE INDEX no_targets on %s()");
  
          createIndex(String.format("CREATE CUSTOM INDEX multi_idx ON %%s(v1, v2) USING '%s'", StubIndex.class.getName()));
          assertIndexCreated("multi_idx", "v1", "v2");
      }
  
      @Test
      public void rejectDuplicateColumnsInTargetList() throws Throwable
      {
          createTable("CREATE TABLE %s(k int, c int, v1 int, v2 int, PRIMARY KEY (k,c))");
  
          assertInvalidMessage("Duplicate column 'v1' in index target list",
                               String.format("CREATE CUSTOM INDEX ON %%s(v1, v1) USING '%s'",
                                             StubIndex.class.getName()));
  
          assertInvalidMessage("Duplicate column 'v1' in index target list",
                               String.format("CREATE CUSTOM INDEX ON %%s(v1, v1, c, c) USING '%s'",
                                             StubIndex.class.getName()));
      }
  
      @Test
      public void requireFullQualifierForFrozenCollectionTargets() throws Throwable
      {
          // this is really just to prove that we require the full modifier on frozen collection
          // targets whether the index is multicolumn or not
          createTable("CREATE TABLE %s(" +
                      " k int," +
                      " c int," +
                      " fmap frozen<map<int, text>>," +
                      " flist frozen<list<int>>," +
                      " fset frozen<set<int>>," +
                      " PRIMARY KEY(k,c))");
  
          assertInvalidMessage("Cannot create keys() index on frozen column fmap. " +
                               "Frozen collections are immutable and must be fully indexed",
                               String.format("CREATE CUSTOM INDEX ON %%s(c, keys(fmap)) USING'%s'",
                                             StubIndex.class.getName()));
          assertInvalidMessage("Cannot create entries() index on frozen column fmap. " +
                               "Frozen collections are immutable and must be fully indexed",
                               String.format("CREATE CUSTOM INDEX ON %%s(c, entries(fmap)) USING'%s'",
                                             StubIndex.class.getName()));
          assertInvalidMessage("Cannot create values() index on frozen column fmap. " +
                               "Frozen collections are immutable and must be fully indexed",
                               String.format("CREATE CUSTOM INDEX ON %%s(c, fmap) USING'%s'", StubIndex.class.getName()));
  
          assertInvalidMessage("Cannot create keys() index on frozen column flist. " +
                               "Frozen collections are immutable and must be fully indexed",
                               String.format("CREATE CUSTOM INDEX ON %%s(c, keys(flist)) USING'%s'",
                                             StubIndex.class.getName()));
          assertInvalidMessage("Cannot create entries() index on frozen column flist. " +
                               "Frozen collections are immutable and must be fully indexed",
                               String.format("CREATE CUSTOM INDEX ON %%s(c, entries(flist)) USING'%s'",
                                             StubIndex.class.getName()));
          assertInvalidMessage("Cannot create values() index on frozen column flist. " +
                               "Frozen collections are immutable and must be fully indexed",
                               String.format("CREATE CUSTOM INDEX ON %%s(c, flist) USING'%s'", StubIndex.class.getName()));
  
          assertInvalidMessage("Cannot create keys() index on frozen column fset. " +
                               "Frozen collections are immutable and must be fully indexed",
                               String.format("CREATE CUSTOM INDEX ON %%s(c, keys(fset)) USING'%s'",
                                             StubIndex.class.getName()));
          assertInvalidMessage("Cannot create entries() index on frozen column fset. " +
                               "Frozen collections are immutable and must be fully indexed",
                               String.format("CREATE CUSTOM INDEX ON %%s(c, entries(fset)) USING'%s'",
                                             StubIndex.class.getName()));
          assertInvalidMessage("Cannot create values() index on frozen column fset. " +
                               "Frozen collections are immutable and must be fully indexed",
                               String.format("CREATE CUSTOM INDEX ON %%s(c, fset) USING'%s'", StubIndex.class.getName()));
  
          createIndex(String.format("CREATE CUSTOM INDEX ON %%s(c, full(fmap)) USING'%s'", StubIndex.class.getName()));
          createIndex(String.format("CREATE CUSTOM INDEX ON %%s(c, full(flist)) USING'%s'", StubIndex.class.getName()));
          createIndex(String.format("CREATE CUSTOM INDEX ON %%s(c, full(fset)) USING'%s'", StubIndex.class.getName()));
      }
  
      @Test
      public void defaultIndexNameContainsTargetColumns()
      {
          createTable("CREATE TABLE %s(k int, c int, v1 int, v2 int, PRIMARY KEY(k,c))");
  
          createIndex(String.format("CREATE CUSTOM INDEX ON %%s(v1, v2) USING '%s'", StubIndex.class.getName()));
          assertEquals(1, getCurrentColumnFamilyStore().metadata().indexes.size());
          assertIndexCreated(currentTable() + "_idx", "v1", "v2");
  
          createIndex(String.format("CREATE CUSTOM INDEX ON %%s(c, v1, v2) USING '%s'", StubIndex.class.getName()));
          assertEquals(2, getCurrentColumnFamilyStore().metadata().indexes.size());
          assertIndexCreated(currentTable() + "_idx_1", "c", "v1", "v2");
  
          createIndex(String.format("CREATE CUSTOM INDEX ON %%s(c, v2) USING '%s'", StubIndex.class.getName()));
          assertEquals(3, getCurrentColumnFamilyStore().metadata().indexes.size());
          assertIndexCreated(currentTable() + "_idx_2", "c", "v2");
  
          // duplicate the previous index with some additional options and check the name is generated as expected
          createIndex(String.format("CREATE CUSTOM INDEX ON %%s(c, v2) USING '%s' WITH OPTIONS = {'foo':'bar'}",
                                    StubIndex.class.getName()));
          assertEquals(4, getCurrentColumnFamilyStore().metadata().indexes.size());
          Map<String, String> options = new HashMap<>();
          options.put("foo", "bar");
          assertIndexCreated(currentTable() + "_idx_3", options, "c", "v2");
      }
  
      @Test
      public void createMultiColumnIndexes()
      {
          // smoke test for various permutations of multicolumn indexes
          createTable("CREATE TABLE %s (" +
                      " pk1 int," +
                      " pk2 int," +
                      " c1 int," +
                      " c2 int," +
                      " v1 int," +
                      " v2 int," +
                      " mval map<text, int>," +
                      " lval list<int>," +
                      " sval set<int>," +
                      " fmap frozen<map<text,int>>," +
                      " flist frozen<list<int>>," +
                      " fset frozen<set<int>>," +
                      " PRIMARY KEY ((pk1, pk2), c1, c2))");
  
          testCreateIndex("idx_1", "pk1", "pk2");
          testCreateIndex("idx_2", "pk1", "c1");
          testCreateIndex("idx_3", "pk1", "c2");
          testCreateIndex("idx_4", "c1", "c2");
          testCreateIndex("idx_5", "c2", "v1");
          testCreateIndex("idx_6", "v1", "v2");
          testCreateIndex("idx_7", "pk2", "c2", "v2");
  
          createIndex(String.format("CREATE CUSTOM INDEX idx_8 ON %%s(" +
                                    "  pk1, c1, v1, values(mval), values(sval), values(lval)" +
                                    ") USING '%s'",
                                    StubIndex.class.getName()));
          assertIndexCreated("idx_8",
                             new HashMap<>(),
                             ImmutableList.of(indexTarget("pk1", IndexTarget.Type.SIMPLE),
                                              indexTarget("c1", IndexTarget.Type.SIMPLE),
                                              indexTarget("v1", IndexTarget.Type.SIMPLE),
                                              indexTarget("mval", IndexTarget.Type.VALUES),
                                              indexTarget("sval", IndexTarget.Type.VALUES),
                                              indexTarget("lval", IndexTarget.Type.VALUES)));
  
          createIndex(String.format("CREATE CUSTOM INDEX inc_frozen ON %%s(" +
                                    "  pk2, c2, v2, full(fmap), full(fset), full(flist)" +
                                    ") USING '%s'",
                                    StubIndex.class.getName()));
          assertIndexCreated("inc_frozen",
                             new HashMap<>(),
                             ImmutableList.of(indexTarget("pk2", IndexTarget.Type.SIMPLE),
                                              indexTarget("c2", IndexTarget.Type.SIMPLE),
                                              indexTarget("v2", IndexTarget.Type.SIMPLE),
                                              indexTarget("fmap", IndexTarget.Type.FULL),
                                              indexTarget("fset", IndexTarget.Type.FULL),
                                              indexTarget("flist", IndexTarget.Type.FULL)));
  
          createIndex(String.format("CREATE CUSTOM INDEX all_teh_things ON %%s(" +
                                    "  pk1, pk2, c1, c2, v1, v2, keys(mval), lval, sval, full(fmap), full(fset), full(flist)" +
                                    ") USING '%s'",
                                    StubIndex.class.getName()));
          assertIndexCreated("all_teh_things",
                             new HashMap<>(),
                             ImmutableList.of(indexTarget("pk1", IndexTarget.Type.SIMPLE),
                                              indexTarget("pk2", IndexTarget.Type.SIMPLE),
                                              indexTarget("c1", IndexTarget.Type.SIMPLE),
                                              indexTarget("c2", IndexTarget.Type.SIMPLE),
                                              indexTarget("v1", IndexTarget.Type.SIMPLE),
                                              indexTarget("v2", IndexTarget.Type.SIMPLE),
                                              indexTarget("mval", IndexTarget.Type.KEYS),
                                              indexTarget("lval", IndexTarget.Type.VALUES),
                                              indexTarget("sval", IndexTarget.Type.VALUES),
                                              indexTarget("fmap", IndexTarget.Type.FULL),
                                              indexTarget("fset", IndexTarget.Type.FULL),
                                              indexTarget("flist", IndexTarget.Type.FULL)));
      }
  
      @Test
      public void createMultiColumnIndexIncludingUserTypeColumn()
      {
          String myType = KEYSPACE + '.' + createType("CREATE TYPE %s (a int, b int)");
          createTable("CREATE TABLE %s (k int PRIMARY KEY, v1 int, v2 frozen<" + myType + ">)");
          testCreateIndex("udt_idx", "v1", "v2");
      }
  
      @Test
      public void createIndexWithoutTargets() throws Throwable
      {
          Assume.assumeTrue("Test does not work with different default secondary index",
                            DatabaseDescriptor.getDefaultSecondaryIndex().equals(CassandraIndex.NAME));
          createTable("CREATE TABLE %s(k int, c int, v1 int, v2 int, PRIMARY KEY(k,c))");
          // only allowed for CUSTOM indexes
          assertInvalidMessage("Only CUSTOM indexes can be created without specifying a target column",
                               "CREATE INDEX ON %s()");
  
          // parentheses are mandatory
          assertInvalidSyntax("CREATE CUSTOM INDEX ON %%s USING '%s'", StubIndex.class.getName());
          createIndex(String.format("CREATE CUSTOM INDEX no_targets ON %%s() USING '%s'", StubIndex.class.getName()));
          assertIndexCreated("no_targets", new HashMap<>());
      }
  
      @Test
      public void testCustomIndexExpressionSyntax() throws Throwable
      {
          Object[] row = row(0, 0, 0, 0);
          createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
          String indexName = currentTable() + "_custom_index";
          execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", row);
  
  
          assertInvalidMessage(String.format(IndexRestrictions.INDEX_NOT_FOUND, indexName, currentTableMetadata().toString()),
                               String.format("SELECT * FROM %%s WHERE expr(%s, 'foo bar baz')", indexName));
  
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(c) USING '%s'", indexName, ColumnTargetedIndex.class.getName()));
  
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    String.format(IndexRestrictions.INDEX_NOT_FOUND, "no_such_index", currentTableMetadata().toString()),
                                    QueryValidationException.class,
                                    "SELECT * FROM %s WHERE expr(no_such_index, 'foo bar baz ')");
  
          // simple case
          assertRows(execute(String.format("SELECT * FROM %%s WHERE expr(%s, 'foo bar baz')", indexName)), row);
          assertRows(execute(String.format("SELECT * FROM %%s WHERE expr(\"%s\", 'foo bar baz')", indexName)), row);
          assertRows(execute(String.format("SELECT * FROM %%s WHERE expr(%s, $$foo \" ~~~ bar Baz$$)", indexName)), row);
  
          // multiple expressions on the same index
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    IndexRestrictions.MULTIPLE_EXPRESSIONS,
                                    QueryValidationException.class,
                                    String.format("SELECT * FROM %%s WHERE expr(%1$s, 'foo') AND expr(%1$s, 'bar')",
                                                  indexName));
  
          // multiple expressions on different indexes
          createIndex(String.format("CREATE CUSTOM INDEX other_custom_index ON %%s(d) USING '%s'", ColumnTargetedIndex.class.getName()));
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    IndexRestrictions.MULTIPLE_EXPRESSIONS,
                                    QueryValidationException.class,
                                    String.format("SELECT * FROM %%s WHERE expr(%s, 'foo') AND expr(other_custom_index, 'bar')",
                                                  indexName));
  
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    StatementRestrictions.REQUIRES_ALLOW_FILTERING_MESSAGE,
                                    QueryValidationException.class,
                                    String.format("SELECT * FROM %%s WHERE expr(%s, 'foo') AND d=0", indexName));
          assertRows(execute(String.format("SELECT * FROM %%s WHERE expr(%s, 'foo') AND d=0 ALLOW FILTERING", indexName)), row);
      }
  
      /**
       * A {@link StubIndex} that only supports expressions on its target column.
       */
      public static final class ColumnTargetedIndex extends StubIndex
      {
          private final ColumnMetadata indexedColumn;
  
          public ColumnTargetedIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
              indexedColumn = TargetParser.parse(baseCfs.metadata(), metadata).left;
          }
  
          @Override
          public boolean supportsExpression(ColumnMetadata column, Operator operator)
          {
              return column.equals(indexedColumn) && super.supportsExpression(column, operator);
          }
      }
  
      @Test
      public void customIndexDoesntSupportCustomExpressions() throws Throwable
      {
          createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
          String indexName = currentTable() + "_custom_index";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(c) USING '%s'",
                                    indexName,
                                    NoCustomExpressionsIndex.class.getName()));
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    String.format( IndexRestrictions.CUSTOM_EXPRESSION_NOT_SUPPORTED, indexName),
                                    QueryValidationException.class,
                                    String.format("SELECT * FROM %%s WHERE expr(%s, 'foo bar baz')", indexName));
      }
  
      @Test
      public void customIndexRejectsExpressionSyntax() throws Throwable
      {
          createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
          String indexName = currentTable() + "_custom_index";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(c) USING '%s'",
                                    indexName,
                                    AlwaysRejectIndex.class.getName()));
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    "None shall pass",
                                    QueryValidationException.class,
                                    String.format("SELECT * FROM %%s WHERE expr(%s, 'foo bar baz')", indexName));
      }
  
      @Test
      public void customExpressionsMustTargetCustomIndex() throws Throwable
      {
          Assume.assumeTrue("Test does not work with different default secondary index",
                            DatabaseDescriptor.getDefaultSecondaryIndex().equals(CassandraIndex.NAME));
          createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
          createIndex("CREATE INDEX non_custom_index ON %s(c)");
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    String.format(IndexRestrictions.NON_CUSTOM_INDEX_IN_EXPRESSION, "non_custom_index"),
                                    QueryValidationException.class,
                                    "SELECT * FROM %s WHERE expr(non_custom_index, 'c=0')");
      }
  
      @Test
      public void customExpressionsDisallowedInModifications() throws Throwable
      {
          createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
          String indexName = currentTable() + "_custom_index";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(c) USING '%s'",
                                    indexName, StubIndex.class.getName()));
  
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    ModificationStatement.CUSTOM_EXPRESSIONS_NOT_ALLOWED,
                                    QueryValidationException.class,
                                    String.format("DELETE FROM %%s WHERE expr(%s, 'foo bar baz ')", indexName));
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    ModificationStatement.CUSTOM_EXPRESSIONS_NOT_ALLOWED,
                                    QueryValidationException.class,
                                    String.format("UPDATE %%s SET d=0 WHERE expr(%s, 'foo bar baz ')", indexName));
      }
  
      @Test
      public void indexSelectionPrefersMostSelectiveIndex() throws Throwable
      {
          createTable("CREATE TABLE %s(a int, b int, c int, PRIMARY KEY (a))");
          createIndex(String.format("CREATE CUSTOM INDEX %s_more_selective ON %%s(b) USING '%s'",
                                    currentTable(),
                                    SettableSelectivityIndex.class.getName()));
          createIndex(String.format("CREATE CUSTOM INDEX %s_less_selective ON %%s(c) USING '%s'",
                                    currentTable(),
                                    SettableSelectivityIndex.class.getName()));
          SettableSelectivityIndex moreSelective =
              (SettableSelectivityIndex)getCurrentColumnFamilyStore().indexManager.getIndexByName(currentTable() + "_more_selective");
          SettableSelectivityIndex lessSelective =
              (SettableSelectivityIndex)getCurrentColumnFamilyStore().indexManager.getIndexByName(currentTable() + "_less_selective");
          assertEquals(0, moreSelective.searchersProvided);
          assertEquals(0, lessSelective.searchersProvided);
  
          // the more selective index should be chosen
          moreSelective.setEstimatedResultRows(1);
          lessSelective.setEstimatedResultRows(1000);
          execute("SELECT * FROM %s WHERE b=0 AND c=0 ALLOW FILTERING");
          assertEquals(1, moreSelective.searchersProvided);
          assertEquals(0, lessSelective.searchersProvided);
  
          // and adjusting the selectivity should have an observable effect
          moreSelective.setEstimatedResultRows(10000);
          execute("SELECT * FROM %s WHERE b=0 AND c=0 ALLOW FILTERING");
          assertEquals(1, moreSelective.searchersProvided);
          assertEquals(1, lessSelective.searchersProvided);
      }
  
      @Test
      public void customExpressionForcesIndexSelection() throws Throwable
      {
          createTable("CREATE TABLE %s(a int, b int, c int, PRIMARY KEY (a))");
          createIndex(String.format("CREATE CUSTOM INDEX %s_more_selective ON %%s(b) USING '%s'",
                                    currentTable(),
                                    SettableSelectivityIndex.class.getName()));
          createIndex(String.format("CREATE CUSTOM INDEX %s_less_selective ON %%s(c) USING '%s'",
                                    currentTable(),
                                    SettableSelectivityIndex.class.getName()));
          SettableSelectivityIndex moreSelective =
              (SettableSelectivityIndex)getCurrentColumnFamilyStore().indexManager.getIndexByName(currentTable() + "_more_selective");
          SettableSelectivityIndex lessSelective =
              (SettableSelectivityIndex)getCurrentColumnFamilyStore().indexManager.getIndexByName(currentTable() + "_less_selective");
          assertEquals(0, moreSelective.searchersProvided);
          assertEquals(0, lessSelective.searchersProvided);
  
          // without a custom expression, the more selective index should be chosen
          moreSelective.setEstimatedResultRows(1);
          lessSelective.setEstimatedResultRows(1000);
          execute("SELECT * FROM %s WHERE b=0 AND c=0 ALLOW FILTERING");
          assertEquals(1, moreSelective.searchersProvided);
          assertEquals(0, lessSelective.searchersProvided);
  
          // when a custom expression is present, its target index should be preferred
          execute(String.format("SELECT * FROM %%s WHERE b=0 AND expr(%s_less_selective, 'expression') ALLOW FILTERING", currentTable()));
          assertEquals(1, moreSelective.searchersProvided);
          assertEquals(1, lessSelective.searchersProvided);
      }
  
      @Test
      public void testCustomExpressionValueType() throws Throwable
      {
          // verify that the type of the expression value is determined by Index::customExpressionValueType
          createTable("CREATE TABLE %s (k int, v1 uuid, v2 blob, PRIMARY KEY(k))");
          createIndex(String.format("CREATE CUSTOM INDEX int_index ON %%s() USING '%s'",
                                    Int32ExpressionIndex.class.getName()));
          createIndex(String.format("CREATE CUSTOM INDEX text_index ON %%s() USING '%s'",
                                    UTF8ExpressionIndex.class.getName()));
  
          execute("SELECT * FROM %s WHERE expr(text_index, 'foo')");
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    "Invalid INTEGER constant (99) for \"custom index expression\" of type text",
                                    QueryValidationException.class,
                                    "SELECT * FROM %s WHERE expr(text_index, 99)");
  
          execute("SELECT * FROM %s WHERE expr(int_index, 99)");
          assertInvalidThrowMessage(Optional.of(ProtocolVersion.CURRENT),
                                    "Invalid STRING constant (foo) for \"custom index expression\" of type int",
                                    QueryValidationException.class,
                                    "SELECT * FROM %s WHERE expr(int_index, 'foo')");
      }
  
      @Test
      public void reloadIndexMetadataOnBaseCfsReload()
      {
          // verify that whenever the base table TableMetadata is reloaded, a reload of the index
          // metadata is performed
          createTable("CREATE TABLE %s (k int, v1 int, PRIMARY KEY(k))");
          createIndex(String.format("CREATE CUSTOM INDEX reload_counter ON %%s() USING '%s'",
                                    CountMetadataReloadsIndex.class.getName()));
          ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
          CountMetadataReloadsIndex index = (CountMetadataReloadsIndex)cfs.indexManager.getIndexByName("reload_counter");
          assertEquals(0, index.reloads.get());
  
          // reloading the CFS, even without any metadata changes invokes the index's metadata reload task
          cfs.reload();
          assertEquals(1, index.reloads.get());
      }
  
      @Test
      public void notifyIndexersOfPartitionAndRowRemovalDuringCleanup() throws Throwable
      {
          createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k,c))");
          createIndex(String.format("CREATE CUSTOM INDEX cleanup_index ON %%s() USING '%s'", StubIndex.class.getName()));
          ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
          StubIndex index  = (StubIndex)cfs.indexManager.getIndexByName("cleanup_index");
  
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 0, 0);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 1, 1);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 2, 2);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 3, 3, 3);
          assertEquals(4, index.rowsInserted.size());
          assertEquals(0, index.partitionDeletions.size());
  
          ReadCommand cmd = Util.cmd(cfs, 0).build();
          try (ReadExecutionController executionController = cmd.executionController();
               UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController))
          {
              assertTrue(iterator.hasNext());
              cfs.indexManager.deletePartition(iterator.next(), FBUtilities.nowInSeconds());
          }
  
          assertEquals(1, index.partitionDeletions.size());
          assertEquals(3, index.rowsDeleted.size());
          for (int i = 0; i < 3; i++)
              assertEquals(index.rowsDeleted.get(i).clustering(), index.rowsInserted.get(i).clustering());
      }
  
      @Test
      public void notifyIndexersOfExpiredRowsDuringCompaction() throws Throwable
      {
          createTable("CREATE TABLE %s (k int, c int, PRIMARY KEY (k,c))");
          createIndex(String.format("CREATE CUSTOM INDEX row_ttl_test_index ON %%s() USING '%s'", StubIndex.class.getName()));
          ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
          StubIndex index  = (StubIndex)cfs.indexManager.getIndexByName("row_ttl_test_index");
  
          execute("INSERT INTO %s (k, c) VALUES (?, ?) USING TTL 1", 0, 0);
          execute("INSERT INTO %s (k, c) VALUES (?, ?)", 0, 1);
          execute("INSERT INTO %s (k, c) VALUES (?, ?)", 0, 2);
          execute("INSERT INTO %s (k, c) VALUES (?, ?)", 3, 3);
          assertEquals(4, index.rowsInserted.size());
          // flush so that we end up with an expiring row in the first sstable
          flush();
  
          // let the row with the ttl expire, then force a compaction
          TimeUnit.SECONDS.sleep(2);
          compact();
  
          // the index should have been notified of the expired row
          assertEquals(1, index.rowsDeleted.size());
          Integer deletedClustering = Int32Type.instance.compose(index.rowsDeleted.get(0).clustering().bufferAt(0));
          assertEquals(0, deletedClustering.intValue());
      }
  
  
      // two stub indexes just to track number of row deletions
      // when we have fully-expired tables and indexes on different columns
      public static class StubIndex1 extends StubIndex
      {
          public StubIndex1(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
      }
  
      public static class StubIndex2 extends StubIndex
      {
  
          public StubIndex2(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
      }
  
      @Test
      public void notifyIndexesOfFullyExpiredSSTablesDuringCompaction()
      {
          createTable("CREATE TABLE %s (id int primary key, col1 int, col2 int) " +
                      "WITH compaction = {'class': 'TimeWindowCompactionStrategy', " +
                      "                   'compaction_window_size': 1," +
                      "                   'compaction_window_unit': 'MINUTES'," +
                      "                   'expired_sstable_check_frequency_seconds': 10} " +
                      "AND gc_grace_seconds = 0");
  
          createIndex(String.format("CREATE CUSTOM INDEX row_ttl_test_index_1 ON %%s(col1) USING '%s'", StubIndex1.class.getName()));
          createIndex(String.format("CREATE CUSTOM INDEX row_ttl_test_index_2 ON %%s(col2) USING '%s'", StubIndex2.class.getName()));
  
          ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
          StubIndex index1  = (StubIndex1)cfs.indexManager.getIndexByName("row_ttl_test_index_1");
          StubIndex index2  = (StubIndex2)cfs.indexManager.getIndexByName("row_ttl_test_index_2");
  
          execute("INSERT INTO %s (id, col1) VALUES (?, ?) USING TTL 20", 0, 0);
          execute("INSERT INTO %s (id, col1) VALUES (?, ?) USING TTL 20", 1, 1);
          execute("INSERT INTO %s (id, col1) VALUES (?, ?) USING TTL 20", 2, 2);
  
          execute("INSERT INTO %s (id, col2) VALUES (?, ?) USING TTL 20", 0, 0);
          execute("INSERT INTO %s (id, col2) VALUES (?, ?) USING TTL 20", 1, 1);
          execute("INSERT INTO %s (id, col2) VALUES (?, ?) USING TTL 20", 2, 2);
  
          flush();
  
          Uninterruptibles.sleepUninterruptibly(60, TimeUnit.SECONDS);
  
          compact();
  
          Assert.assertFalse(index1.rowsDeleted.isEmpty());
          Assert.assertEquals(3, index1.rowsDeleted.size());
  
          Assert.assertFalse(index2.rowsDeleted.isEmpty());
          Assert.assertEquals(3, index2.rowsDeleted.size());
      }
  
      @Test
      public void validateOptions()
      {
          createTable("CREATE TABLE %s(k int, c int, v1 int, v2 int, PRIMARY KEY(k,c))");
          createIndex(String.format("CREATE CUSTOM INDEX ON %%s(c, v2) USING '%s' WITH OPTIONS = {'foo':'bar'}",
                                    IndexWithValidateOptions.class.getName()));
          assertNotNull(IndexWithValidateOptions.options);
          assertEquals("bar", IndexWithValidateOptions.options.get("foo"));
      }
  
      @Test
      public void validateOptionsWithTableMetadata()
      {
          createTable("CREATE TABLE %s(k int, c int, v1 int, v2 int, PRIMARY KEY(k,c))");
          createIndex(String.format("CREATE CUSTOM INDEX ON %%s(c, v2) USING '%s' WITH OPTIONS = {'foo':'bar'}",
                                    IndexWithOverloadedValidateOptions.class.getName()));
          TableMetadata table = getCurrentColumnFamilyStore().metadata();
          assertEquals(table, IndexWithOverloadedValidateOptions.table);
          assertNotNull(IndexWithOverloadedValidateOptions.options);
          assertEquals("bar", IndexWithOverloadedValidateOptions.options.get("foo"));
      }
  
      @Test
      public void testFailing2iFlush() throws Throwable
      {
          createTable("CREATE TABLE %s (pk int PRIMARY KEY, value int)");
          createIndex("CREATE CUSTOM INDEX IF NOT EXISTS ON %s(value) USING 'org.apache.cassandra.index.CustomIndexTest$BrokenCustom2I'");
  
          for (int i = 0; i < 10; i++)
              execute("INSERT INTO %s (pk, value) VALUES (?, ?)", i, i);
  
          try
          {
              flush();
              fail("Exception should have been propagated");
          }
          catch (Throwable t)
          {
              assertTrue(t.getMessage().contains("Broken2I"));
          }
  
          // SSTables remain uncommitted.
          assertEquals(0, getCurrentColumnFamilyStore().getDirectories().getDirectoryForNewSSTables().tryList().length);
      }
  
      @Test
      public void indexBuildingPagesLargePartitions() throws Throwable
      {
          createTable("CREATE TABLE %s(k int, c int, v int, PRIMARY KEY(k,c))");
          ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
          SecondaryIndexManager indexManager = cfs.indexManager;
          int totalRows = SimulateConcurrentFlushingIndex.ROWS_IN_PARTITION;
          // Insert a single wide partition to be indexed
          for (int i = 0; i < totalRows; i++)
              execute("INSERT INTO %s (k, c, v) VALUES (0, ?, ?)", i, i);
          Util.flush(cfs);
  
          // Create the index, which won't automatically start building
          String indexName = "build_single_partition_idx";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v) USING '%s'",
                                    indexName, SimulateConcurrentFlushingIndex.class.getName()));
          SimulateConcurrentFlushingIndex index = (SimulateConcurrentFlushingIndex) indexManager.getIndexByName(indexName);
  
          // Index the partition with an Indexer which artificially simulates additional concurrent
          // flush activity by periodically issuing barriers on the read & write op groupings
          DecoratedKey targetKey = getCurrentColumnFamilyStore().decorateKey(ByteBufferUtil.bytes(0));
          indexManager.indexPartition(targetKey, Collections.singleton(index), totalRows / 10);
  
          // When indexing is done check that:
          // * The base table's read ordering at finish was > the one at the start (i.e. that
          //   we didn't hold a single read OpOrder.Group for the whole operation.
          // * That multiple write OpOrder.Groups were used to perform the writes to the index
          // * That all operations are complete, that none of the relevant OpOrder.Groups are
          //   marked as blocking progress and that all the barriers' ops are considered done.
          assertTrue(index.readOrderingAtFinish.compareTo(index.readOrderingAtStart) > 0);
          assertTrue(index.writeGroups.size() > 1);
          assertFalse(index.readOrderingAtFinish.isBlocking());
          index.writeGroups.forEach(group -> assertFalse(group.isBlocking()));
          index.readBarriers.forEach(b -> assertTrue(b.getSyncPoint().isFinished()));
          index.writeBarriers.forEach(b -> {
              b.await(); // Keyspace.writeOrder is global, so this might be temporally blocked by other tests
              assertTrue(b.getSyncPoint().isFinished());
          });
      }
  
      @Test
      public void partitionIndexTest() throws Throwable
      {
          createTable("CREATE TABLE %s(k int, c int, v int, s int static, PRIMARY KEY(k,c))");
          ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
  
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 1, 1, 1);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 1, 2, 2);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 1, 3, 3);
  
          execute("INSERT INTO %s (k, c) VALUES (?, ?)", 2, 2);
  
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 3, 1, 1);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 3, 2, 2);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 3, 3, 3);
          execute("DELETE FROM %s WHERE k = ? AND c >= ?", 3, 3);
          execute("DELETE FROM %s WHERE k = ? AND c <= ?", 3, 1);
  
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 4, 1, 1);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 4, 2, 2);
          execute("DELETE FROM %s WHERE k = ? AND c = ?", 4, 1);
  
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 5, 1, 1);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 5, 2, 2);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 5, 3, 3);
          execute("DELETE FROM %s WHERE k = ?", 5);
  
          Util.flush(cfs);
  
          String indexName = "partition_index_test_idx";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v) USING '%s'",
                                    indexName, StubIndex.class.getName()));
  
          SecondaryIndexManager indexManager = cfs.indexManager;
          StubIndex index = (StubIndex) indexManager.getIndexByName(indexName);
  
          DecoratedKey targetKey;
          for (int pageSize = 1; pageSize <= 5; pageSize++)
          {
              targetKey = getCurrentColumnFamilyStore().decorateKey(ByteBufferUtil.bytes(1));
              indexManager.indexPartition(targetKey, Collections.singleton(index), pageSize);
              assertEquals(3, index.rowsInserted.size());
              assertEquals(0, index.rangeTombstones.size());
              assertTrue(index.partitionDeletions.get(0).isLive());
              index.reset();
          }
  
          for (int pageSize = 1; pageSize <= 5; pageSize++)
          {
              targetKey = getCurrentColumnFamilyStore().decorateKey(ByteBufferUtil.bytes(2));
              indexManager.indexPartition(targetKey, Collections.singleton(index), pageSize);
              assertEquals(1, index.rowsInserted.size());
              assertEquals(0, index.rangeTombstones.size());
              assertTrue(index.partitionDeletions.get(0).isLive());
              index.reset();
          }
  
          for (int pageSize = 1; pageSize <= 5; pageSize++)
          {
              targetKey = getCurrentColumnFamilyStore().decorateKey(ByteBufferUtil.bytes(3));
              indexManager.indexPartition(targetKey, Collections.singleton(index), pageSize);
              assertEquals(1, index.rowsInserted.size());
              assertEquals(2, index.rangeTombstones.size());
              assertTrue(index.partitionDeletions.get(0).isLive());
              index.reset();
          }
  
          for (int pageSize = 1; pageSize <= 5; pageSize++)
          {
              targetKey = getCurrentColumnFamilyStore().decorateKey(ByteBufferUtil.bytes(5));
              indexManager.indexPartition(targetKey, Collections.singleton(index), pageSize);
              assertEquals(1, index.partitionDeletions.size());
              assertFalse(index.partitionDeletions.get(0).isLive());
              index.reset();
          }
      }
  
      @Test
      public void partitionIsNotOverIndexed() throws Throwable
      {
          createTable("CREATE TABLE %s(k int, c int, v int, PRIMARY KEY(k,c))");
          ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
          SecondaryIndexManager indexManager = cfs.indexManager;
  
          int totalRows = 1;
  
          // Insert a single row partition to be indexed
          for (int i = 0; i < totalRows; i++)
              execute("INSERT INTO %s (k, c, v) VALUES (0, ?, ?)", i, i);
          Util.flush(cfs);
  
          // Create the index, which won't automatically start building
          String indexName = "partition_overindex_test_idx";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v) USING '%s'",
                                    indexName, StubIndex.class.getName()));
          StubIndex index = (StubIndex) indexManager.getIndexByName(indexName);
  
          // Index the partition
          DecoratedKey targetKey = getCurrentColumnFamilyStore().decorateKey(ByteBufferUtil.bytes(0));
          indexManager.indexPartition(targetKey, Collections.singleton(index), totalRows);
  
          // Assert only one partition is counted
          assertEquals(1, index.beginCalls);
          assertEquals(1, index.finishCalls);
      }
  
      @Test
      public void rangeTombstoneTest() throws Throwable
      {
          createTable("CREATE TABLE %s(k int, c int, v int, v2 int, PRIMARY KEY(k,c))");
          ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
          SecondaryIndexManager indexManager = cfs.indexManager;
  
          // Insert a single range tombstone
          execute("DELETE FROM %s WHERE k=1 and c > 2");
          Util.flush(cfs);
  
          // Create the index, which won't automatically start building
          String indexName = "range_tombstone_idx";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v) USING '%s'",
                                    indexName, StubIndex.class.getName()));
          String indexName2 = "range_tombstone_idx2";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v2) USING '%s'",
                                    indexName2, StubIndex.class.getName()));
  
          StubIndex index = (StubIndex) indexManager.getIndexByName(indexName);
          StubIndex index2 = (StubIndex) indexManager.getIndexByName(indexName2);
  
          // Index the partition
          DecoratedKey targetKey = getCurrentColumnFamilyStore().decorateKey(ByteBufferUtil.bytes(1));
          indexManager.indexPartition(targetKey, Sets.newHashSet(index, index2), 1);
  
          // and both indexes should have the same range tombstone
          assertEquals(index.rangeTombstones, index2.rangeTombstones);
      }
  
  
      // Used for index creation above
      @SuppressWarnings("unused")
      public static class BrokenCustom2I extends StubIndex
      {
          public BrokenCustom2I(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          public Callable<?> getBlockingFlushTask()
          {
              throw new RuntimeException("Broken2I");
          }
      }
  
      private void testCreateIndex(String indexName, String... targetColumnNames)
      {
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(%s) USING '%s'",
                                    indexName, String.join(",", targetColumnNames), StubIndex.class.getName()));
          assertIndexCreated(indexName, targetColumnNames);
      }
  
      private void assertIndexCreated(String name, String... targetColumnNames)
      {
          assertIndexCreated(name, new HashMap<>(), targetColumnNames);
      }
  
      private void assertIndexCreated(String name, Map<String, String> options, String... targetColumnNames)
      {
          List<IndexTarget> targets = Arrays.stream(targetColumnNames)
                                            .map(s -> new IndexTarget(ColumnIdentifier.getInterned(s, true), IndexTarget.Type.SIMPLE))
                                            .collect(Collectors.toList());
          assertIndexCreated(name, options, targets);
      }
  
      private void assertIndexCreated(String name, Map<String, String> options, List<IndexTarget> targets)
      {
          // all tests here use StubIndex as the custom index class,
          // so add that to the map of options
          options.put(CUSTOM_INDEX_OPTION_NAME, StubIndex.class.getName());
          IndexMetadata expected = IndexMetadata.fromIndexTargets(targets, name, IndexMetadata.Kind.CUSTOM, options);
          Indexes indexes = getCurrentColumnFamilyStore().metadata().indexes;
          for (IndexMetadata actual : indexes)
              if (actual.equals(expected))
                  return;
  
          fail(String.format("Index %s not found", expected));
      }
  
      private static IndexTarget indexTarget(String name, IndexTarget.Type type)
      {
          return new IndexTarget(ColumnIdentifier.getInterned(name, true), type);
      }
  
      public static final class CountMetadataReloadsIndex extends StubIndex
      {
          private final AtomicInteger reloads = new AtomicInteger(0);
  
          public CountMetadataReloadsIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          public void reset()
          {
              super.reset();
              reloads.set(0);
          }
  
          public Callable<?> getMetadataReloadTask(IndexMetadata indexMetadata)
          {
              return reloads::incrementAndGet;
          }
      }
  
      public static final class IndexIncludedInBuild extends StubIndex
      {
          public IndexIncludedInBuild(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          public boolean shouldBuildBlocking()
          {
              return true;
          }
      }
  
      public static final class UTF8ExpressionIndex extends StubIndex
      {
          public UTF8ExpressionIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          public AbstractType<?> customExpressionValueType()
          {
              return UTF8Type.instance;
          }
      }
  
      public static final class Int32ExpressionIndex extends StubIndex
      {
          public Int32ExpressionIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          public AbstractType<?> customExpressionValueType()
          {
              return Int32Type.instance;
          }
      }
  
      public static final class SettableSelectivityIndex extends StubIndex
      {
          private int searchersProvided = 0;
          private long estimatedResultRows = 0;
  
          public SettableSelectivityIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          public void setEstimatedResultRows(long estimate)
          {
              estimatedResultRows = estimate;
          }
  
          public long getEstimatedResultRows()
          {
              return estimatedResultRows;
          }
  
          public Searcher searcherFor(ReadCommand command)
          {
                  searchersProvided++;
                  return super.searcherFor(command);
          }
      }
  
      public static final class IndexExcludedFromBuild extends StubIndex
      {
          public IndexExcludedFromBuild(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          public boolean shouldBuildBlocking()
          {
              return false;
          }
      }
  
      public static final class NoCustomExpressionsIndex extends StubIndex
      {
          public NoCustomExpressionsIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          public AbstractType<?> customExpressionValueType()
          {
              return null;
          }
      }
  
      public static final class AlwaysRejectIndex extends StubIndex
      {
          public AlwaysRejectIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          public void validate(ReadCommand command) throws InvalidRequestException
          {
              throw new InvalidRequestException("None shall pass");
          }
  
          public Searcher searcherFor(ReadCommand command)
          {
              throw new InvalidRequestException("None shall pass (though I'd have expected to fail faster)");
          }
      }
  
      public static final class IndexWithValidateOptions extends StubIndex
      {
          public static Map<String, String> options;
  
          public IndexWithValidateOptions(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          @SuppressWarnings("unused")
          public static Map<String, String> validateOptions(Map<String, String> options)
          {
              IndexWithValidateOptions.options = options;
              return new HashMap<>();
          }
      }
  
      public static final class IndexWithOverloadedValidateOptions extends StubIndex
      {
          public static TableMetadata table;
          public static Map<String, String> options;
  
          public IndexWithOverloadedValidateOptions(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          @SuppressWarnings("unused")
          public static Map<String, String> validateOptions(Map<String, String> options, TableMetadata table)
          {
              IndexWithOverloadedValidateOptions.options = options;
              IndexWithOverloadedValidateOptions.table = table;
              return new HashMap<>();
          }
      }
  
      public static final class SimulateConcurrentFlushingIndex extends StubIndex
      {
          ColumnFamilyStore baseCfs;
          AtomicInteger indexedRowCount = new AtomicInteger(0);
  
          OpOrder.Group readOrderingAtStart = null;
          OpOrder.Group readOrderingAtFinish = null;
          Set<OpOrder.Group> writeGroups = new HashSet<>();
          List<OpOrder.Barrier> readBarriers = new ArrayList<>();
          List<OpOrder.Barrier> writeBarriers = new ArrayList<>();
  
          static final int ROWS_IN_PARTITION = 1000;
  
          public SimulateConcurrentFlushingIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
              this.baseCfs = baseCfs;
          }
  
          // When indexing an entire partition 2 potential problems can be caused by
          // whilst holding a single read & a single write OpOrder.Group.
          // * By holding a write group too long, flushes are blocked
          // * Holding a read group for too long prevents the memory from flushed memtables
          //   from being reclaimed.
          // See CASSANDRA-12796 for details.
          // To test that the index builder pages through a large partition, using
          // finer grained OpOrder.Groups we write a "large" partition to disk, then
          // kick off an index build on it, using this indexer.
          // To simulate concurrent flush activity, we periodically issue barriers on
          // the current read and write groups.
          // When we're done indexing the partition, the test checks the states of the
          // various OpOrder.Groups, which it can obtain from this index.
  
          @Override
          public Indexer indexerFor(final DecoratedKey key,
                                    RegularAndStaticColumns columns,
                                    long nowInSec,
                                    WriteContext ctx,
                                    IndexTransaction.Type transactionType,
                                    Memtable memtable)
          {
              CassandraWriteContext cassandraWriteContext = (CassandraWriteContext) ctx;
              if (readOrderingAtStart == null)
                  readOrderingAtStart = baseCfs.readOrdering.getCurrent();
  
              writeGroups.add(cassandraWriteContext.getGroup());
  
              return new Indexer()
              {
                  public void begin()
                  {
                      // to simulate other activity on base table during indexing, issue
                      // barriers on the read and write orderings. This is analogous to
                      // what happens when other flushes are being processed during the
                      // indexing of a partition
                      OpOrder.Barrier readBarrier = baseCfs.readOrdering.newBarrier();
                      readBarrier.issue();
                      readBarriers.add(readBarrier);
                      OpOrder.Barrier writeBarrier = Keyspace.writeOrder.newBarrier();
                      writeBarrier.issue();
                      writeBarriers.add(writeBarrier);
                  }
  
                  public void insertRow(Row row)
                  {
                      indexedRowCount.incrementAndGet();
                  }
  
                  public void finish()
                  {
                      // we've indexed all rows in the target partition,
                      // grab the read OpOrder.Group for the base CFS so
                      // we can compare it with the starting group
                      if (indexedRowCount.get() < ROWS_IN_PARTITION)
                          readOrderingAtFinish = baseCfs.readOrdering.getCurrent();
                  }
  
                  public void partitionDelete(DeletionTime deletionTime) { }
  
                  public void rangeTombstone(RangeTombstone tombstone) { }
  
                  public void updateRow(Row oldRowData, Row newRowData) { }
  
                  public void removeRow(Row row) { }
  
              };
          }
      }
  
  
      @Test
      public void testFlushObserver()
      {
          createTable("CREATE TABLE %s (k int, c int, s int static, v int, PRIMARY KEY (k, c))");
          String indexName = "test_index_with_flush_observer";
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v) USING '%s'",
                                    indexName, IndexWithFlushObserver.class.getName()));
  
          execute("INSERT INTO %s (k, c, s, v) VALUES (?, ?, ?, ?)", 0, 0, 0, 0);
          execute("INSERT INTO %s (k, c, s, v) VALUES (?, ?, ?, ?)", 0, 1, 1, 1);
          execute("INSERT INTO %s (k, c, s, v) VALUES (?, ?, ?, ?)", 1, 0, 2, 2);
          execute("INSERT INTO %s (k, c, s, v) VALUES (?, ?, ?, ?)", 1, 1, 3, 3);
  
          ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
          SecondaryIndexManager indexManager = cfs.indexManager;
          IndexWithFlushObserver index = (IndexWithFlushObserver) indexManager.getIndexByName(indexName);
  
          assertEquals(0, index.beginFlushCalls.get());
          assertEquals(0, index.flushedPartitions.get());
          assertEquals(0, index.flushedStaticRows.get());
          assertEquals(0, index.flushedUnfiltereds.get());
          assertEquals(0, index.completeFlushCalls.get());
  
          cfs.forceBlockingFlush(FlushReason.UNIT_TESTS);
  
          assertEquals(1, index.beginFlushCalls.get());
          assertEquals(2, index.flushedPartitions.get());
          assertEquals(2, index.flushedStaticRows.get());
          assertEquals(4, index.flushedUnfiltereds.get());
          assertEquals(1, index.completeFlushCalls.get());
  
          execute("DELETE FROM %s WHERE k=?", 0);
          execute("DELETE FROM %s WHERE k=? AND c>=?", 1, 1);
          index.reset();
          cfs.forceBlockingFlush(FlushReason.UNIT_TESTS);
  
          assertEquals(1, index.beginFlushCalls.get());
          assertEquals(2, index.flushedPartitions.get());
          assertEquals(0, index.flushedStaticRows.get()); // flushed data has no static values..
          assertEquals(2, index.flushedUnfiltereds.get());
          assertEquals(1, index.completeFlushCalls.get());
      }
  
      /**
       * A {@link StubIndex} using a {@link SSTableFlushObserver} that just keeps count of operations.
       */
      public static final class IndexWithFlushObserver extends StubIndex
      {
  
          AtomicInteger beginFlushCalls = new AtomicInteger();
          AtomicInteger flushedPartitions = new AtomicInteger();
          AtomicInteger flushedStaticRows = new AtomicInteger();
          AtomicInteger flushedUnfiltereds = new AtomicInteger();
          AtomicInteger completeFlushCalls = new AtomicInteger();
  
          public IndexWithFlushObserver(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          @Override
          public void reset()
          {
              super.reset();
              beginFlushCalls.set(0);
              flushedPartitions.set(0);
              flushedStaticRows.set(0);
              flushedUnfiltereds.set(0);
              completeFlushCalls.set(0);
          }
  
          @Override
          public SSTableFlushObserver getFlushObserver(Descriptor descriptor, ILifecycleTransaction txn)
          {
              return new SSTableFlushObserver() {
  
                  @Override
                  public void begin()
                  {
                      beginFlushCalls.incrementAndGet();
                  }
  
                  @Override
                  public void startPartition(DecoratedKey key, long keyPosition, long keyPositionForSASI)
                  {
                      flushedPartitions.incrementAndGet();
                  }
  
                  @Override
                  public void staticRow(Row staticRow)
                  {
                      flushedStaticRows.incrementAndGet();
                  }
  
                  @Override
                  public void nextUnfilteredCluster(Unfiltered unfiltered)
                  {
                      flushedUnfiltereds.incrementAndGet();
                  }
  
                  @Override
                  public void complete()
                  {
                      completeFlushCalls.incrementAndGet();
                  }
              };
          }
      }
  
      /**
       * Verify that writes for indexes in the same {@link Index.Group} are grouped.
       */
      @Test
      public void testGroupedWrites() throws Throwable
      {
          // create the schema with two indexes in the same group
          String indexClassName = IndexWithSharedGroup.class.getName();
          createTable("CREATE TABLE %s (k int, c int, s int static, v int, PRIMARY KEY (k,c))");
          createIndex(String.format("CREATE CUSTOM INDEX grouped_index_c ON %%s(c) USING '%s'", indexClassName));
          createIndex(String.format("CREATE CUSTOM INDEX grouped_index_v ON %%s(v) USING '%s'", indexClassName));
  
          // retrieve the indexes and their shared group
          ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
          SecondaryIndexManager indexManager = cfs.indexManager;
          StubIndex index1 = (IndexWithSharedGroup) indexManager.getIndexByName("grouped_index_c");
          StubIndex index2 = (IndexWithSharedGroup) indexManager.getIndexByName("grouped_index_v");
          IndexWithSharedGroup.Group group = indexManager.listIndexGroups()
                                                         .stream()
                                                         .filter(g -> g instanceof IndexWithSharedGroup.Group)
                                                         .map(g -> (IndexWithSharedGroup.Group) g)
                                                         .findAny()
                                                         .orElseThrow(AssertionError::new);
  
          // verify that row insertions get to the index group and they are propagated to their members
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 0, 0);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 1, 1);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 1, 0, 3);
          assertEquals(3, group.rowsInserted.get());
          assertEquals(3, index1.rowsInserted.size());
          assertEquals(3, index2.rowsInserted.size());
  
          // verify that row updates get to the index group and they are propagated to their members
          execute("UPDATE %s SET v=? WHERE k=? AND c=?", 10, 0, 0);
          execute("UPDATE %s SET v=? WHERE k=? AND c=?", 10, 1, 0);
          assertEquals(2, group.rowsUpdated.get());
          assertEquals(2, index1.rowsUpdated.size());
          assertEquals(2, index2.rowsUpdated.size());
  
          // verify that partition deletions get to the index group and its members
          ReadCommand cmd = Util.cmd(cfs, 0).build();
          try (ReadExecutionController executionController = cmd.executionController();
               UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController))
          {
              assertTrue(iterator.hasNext());
              cfs.indexManager.deletePartition(iterator.next(), FBUtilities.nowInSeconds());
          }
          assertEquals(1, group.partitionDeletions.get());
          assertEquals(1, index1.partitionDeletions.size());
          assertEquals(1, index2.partitionDeletions.size());
  
          // verify that the row deletions produced by the previous partition deletion get to the group and its members
          assertEquals(2, group.rowsDeleted.get());
          assertEquals(2, index1.rowsDeleted.size());
          assertEquals(2, index2.rowsDeleted.size());
  
          // verify that range tombstones get to the index group and its members
          execute("DELETE FROM %s WHERE k=? AND c>?", 0, 0);
          execute("DELETE FROM %s WHERE k=? AND c>?", 1, 1);
          assertEquals(2, group.rangeTombstones.get());
          assertEquals(2, index1.rangeTombstones.size());
          assertEquals(2, index2.rangeTombstones.size());
  
          // verify the total number of begin calls
          assertEquals(10, group.beginCalls.get());
          assertEquals(10, index1.beginCalls);
          assertEquals(10, index2.beginCalls);
  
          // verify the total number of finish calls
          assertEquals(10, group.finishCalls.get());
          assertEquals(10, index1.finishCalls);
          assertEquals(10, index2.finishCalls);
  
          // flush the previous data to get rid of it, reset the group counters and flush a new memtable
          cfs.forceBlockingFlush(FlushReason.UNIT_TESTS);
          group.reset();
          execute("INSERT INTO %s (k, s) VALUES (?, ?)", 1, 0);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 1, 0, 0);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 1, 1, 0);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 2, 0, 0);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 2, 1, 0);
          execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 2, 2, 0);
          execute("DELETE FROM %s WHERE k=? AND c=?", 2, 3);
          execute("DELETE FROM %s WHERE k=?", 3);
          cfs.forceBlockingFlush(FlushReason.UNIT_TESTS);
  
          // verify that the flush observer calls get only once to the group
          assertEquals(1, group.beginFlushCalls.get());
          assertEquals(3, group.flushedPartitions.get());
          assertEquals(3, group.flushedStaticRows.get());
          assertEquals(6, group.flushedUnfiltereds.get());
          assertEquals(1, group.completeFlushCalls.get());
  
          // verify that the index rebuilds can be directed only to the first index
          group.reset();
          indexManager.rebuildIndexesBlocking(Collections.singleton(index1.getIndexMetadata().name));
          assertEquals(8, group.rowsInserted.get());
          assertEquals(8, index1.rowsInserted.size());
          assertEquals(0, index2.rowsInserted.size());
  
          // verify that the index rebuilds can be directed only to the second index
          group.reset();
          indexManager.rebuildIndexesBlocking(Collections.singleton(index2.getIndexMetadata().name));
          assertEquals(8, group.rowsInserted.get());
          assertEquals(0, index1.rowsInserted.size());
          assertEquals(8, index2.rowsInserted.size());
      }
  
      @Test
      public void testIndexGroupsInstancesManagement() throws Throwable
      {
          Assume.assumeTrue("Test does not work with different default secondary index",
                            DatabaseDescriptor.getDefaultSecondaryIndex().equals(CassandraIndex.NAME));
          String indexClassName = IndexWithSharedGroup.class.getName();
          createTable("CREATE TABLE %s (k int PRIMARY KEY, v1 int, v2 int, v3 int, v4 int, v5 int)");
          SecondaryIndexManager indexManager = getCurrentColumnFamilyStore().indexManager;
  
          // create two indexes belonging to the same group and verify that only one group is added to the manager
          String idx1 = createIndex(String.format("CREATE CUSTOM INDEX ON %%s(v1) USING '%s'", indexClassName));
          String idx2 = createIndex(String.format("CREATE CUSTOM INDEX ON %%s(v2) USING '%s'", indexClassName));
          Supplier<IndexWithSharedGroup.Group> groupSupplier =
                  () -> indexManager.listIndexGroups().stream()
                                                      .filter(g -> g instanceof IndexWithSharedGroup.Group)
                                                      .map(g -> (IndexWithSharedGroup.Group) g)
                                                      .findAny()
                                                      .orElse(null);
          IndexWithSharedGroup.Group group = groupSupplier.get();
          // verify that only one group has been added to the manager
          assertEquals(2, indexManager.listIndexes().size());
          assertEquals(1, indexManager.listIndexGroups().size());
          assertEquals(2, group.indexes.size());
  
          // create two indexes belonging to their own singleton group and verify that two groups are added to the manager
          String idx3 = createIndex("CREATE INDEX ON %s(v3)");
          String idx4 = createIndex("CREATE INDEX ON %s(v4)");
          assertEquals(4, indexManager.listIndexes().size());
          assertEquals(3, indexManager.listIndexGroups().size());
  
          // create another index to the shared group and verify that they are added to the existing group instance
          String idx5 = createIndex(String.format("CREATE CUSTOM INDEX ON %%s(v5) USING '%s'", indexClassName));
          assertEquals(5, indexManager.listIndexes().size());
          assertEquals(3, indexManager.listIndexGroups().size());
          assertEquals(3, group.indexes.size());
  
          // drop one of the shared group members and verify that the manager still has the same group count
          dropIndex("DROP INDEX %s." + idx1);
          assertEquals(4, indexManager.listIndexes().size());
          assertEquals(3, indexManager.listIndexGroups().size());
          assertEquals(2, group.indexes.size());
  
          // drop the standalone indexes and verify that their singleton groups are removed from the manager
          dropIndex("DROP INDEX %s." + idx3);
          dropIndex("DROP INDEX %s." + idx4);
          assertEquals(2, indexManager.listIndexes().size());
          assertEquals(1, indexManager.listIndexGroups().size());
  
          // drop the remaining members of the shared group and verify that it no longer exists in the manager
          dropIndex("DROP INDEX %s." + idx2);
          dropIndex("DROP INDEX %s." + idx5);
          assertEquals(0, indexManager.listIndexes().size());
          assertEquals(0, indexManager.listIndexGroups().size());
          assertEquals(0, group.indexes.size());
  
          // create the sharing group members again and verify that they are added to a new group instance
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v1) USING '%s'", idx1, indexClassName));
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v2) USING '%s'", idx2, indexClassName));
          createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(v3) USING '%s'", idx3, indexClassName));
          IndexWithSharedGroup.Group newGroup = indexManager.listIndexGroups()
                                                            .stream()
                                                            .filter(g -> g instanceof IndexWithSharedGroup.Group)
                                                            .map(g -> (IndexWithSharedGroup.Group) g)
                                                            .findAny()
                                                            .orElseThrow(AssertionError::new);
          assertEquals(3, indexManager.listIndexes().size());
          assertEquals(1, indexManager.listIndexGroups().size());
          assertEquals(3, newGroup.indexes.size());
      }
  
      /**
       * {@link StubIndex} implementation that uses the same {@link Index.Group} for all its instances.
       * That group keeps count of the calls and passes them to its members.
       */
      public static class IndexWithSharedGroup extends StubIndex
      {
          public IndexWithSharedGroup(ColumnFamilyStore baseCfs, IndexMetadata metadata)
          {
              super(baseCfs, metadata);
          }
  
          @Override
          public boolean shouldBuildBlocking()
          {
              return true;
          }
  
          @Override
          public void register(IndexRegistry registry)
          {
              registry.registerIndex(this, new Group.Key(Group.class), Group::new);
          }
  
          @Override
          public void unregister(IndexRegistry registry)
          {
              registry.unregisterIndex(this, new Group.Key(Group.class));
          }
  
          private static class Group implements Index.Group
          {
              Map<String, IndexWithSharedGroup> indexes = Maps.newConcurrentMap();
  
              AtomicInteger beginCalls = new AtomicInteger();
              AtomicInteger finishCalls = new AtomicInteger();
              AtomicInteger partitionDeletions = new AtomicInteger();
              AtomicInteger rangeTombstones = new AtomicInteger();
              AtomicInteger rowsInserted = new AtomicInteger();
              AtomicInteger rowsDeleted = new AtomicInteger();
              AtomicInteger rowsUpdated = new AtomicInteger();
  
              AtomicInteger beginFlushCalls = new AtomicInteger();
              AtomicInteger flushedPartitions = new AtomicInteger();
              AtomicInteger flushedStaticRows = new AtomicInteger();
              AtomicInteger flushedUnfiltereds = new AtomicInteger();
              AtomicInteger completeFlushCalls = new AtomicInteger();
  
              public void reset()
              {
                  beginCalls.set(0);
                  finishCalls.set(0);
                  partitionDeletions.set(0);
                  rangeTombstones.set(0);
                  rowsInserted.set(0);
                  rowsDeleted.set(0);
                  rowsUpdated.set(0);
                  beginFlushCalls.set(0);
                  flushedPartitions.set(0);
                  flushedStaticRows.set(0);
                  flushedUnfiltereds.set(0);
                  completeFlushCalls.set(0);
                  indexes.values().forEach(IndexWithSharedGroup::reset);
              }
  
              @Override
              public Set<Index> getIndexes()
              {
                  return ImmutableSet.copyOf(indexes.values());
              }
  
              @Override
              public void addIndex(Index index)
              {
                  indexes.put(index.getIndexMetadata().name, (IndexWithSharedGroup) index);
              }
  
              @Override
              public void removeIndex(Index index)
              {
                  indexes.remove(index.getIndexMetadata().name);
              }
  
              @Override
              public boolean containsIndex(Index index)
              {
                  return indexes.containsKey(index.getIndexMetadata().name);
              }
  
              @Override
              public boolean isSingleton()
              {
                  return false;
              }
  
              @Override
              public Index.Indexer indexerFor(Predicate<Index> indexSelector,
                                              DecoratedKey key,
                                              RegularAndStaticColumns columns,
                                              long nowInSec,
                                              WriteContext context,
                                              IndexTransaction.Type transactionType,
                                              Memtable memtable)
              {
                  Set<Index.Indexer> indexers = indexes.values()
                                                       .stream()
                                                       .filter(indexSelector)
                                                       .map(i -> i.indexerFor(key, columns, nowInSec, context, transactionType, memtable))
                                                       .filter(Objects::nonNull)
                                                       .collect(Collectors.toSet());
  
                  return indexers.isEmpty() ? null : new Index.Indexer() {
  
                      @Override
                      public void begin()
                      {
                          beginCalls.incrementAndGet();
                          indexers.forEach(Indexer::begin);
                      }
  
                      @Override
                      public void partitionDelete(DeletionTime deletionTime)
                      {
                          partitionDeletions.incrementAndGet();
                          indexers.forEach(indexer -> indexer.partitionDelete(deletionTime));
                      }
  
                      @Override
                      public void rangeTombstone(RangeTombstone tombstone)
                      {
                          rangeTombstones.incrementAndGet();
                          indexers.forEach(indexer -> indexer.rangeTombstone(tombstone));
                      }
  
                      @Override
                      public void insertRow(Row row)
                      {
                          rowsInserted.incrementAndGet();
                          indexers.forEach(indexer -> indexer.insertRow(row));
                      }
  
                      @Override
                      public void removeRow(Row row)
                      {
                          rowsDeleted.incrementAndGet();
                          indexers.forEach(indexer -> indexer.removeRow(row));
                      }
  
                      @Override
                      public void updateRow(Row oldRow, Row newRow)
                      {
                          rowsUpdated.incrementAndGet();
                          indexers.forEach(indexer -> indexer.updateRow(oldRow, newRow));
                      }
  
                      @Override
                      public void finish()
                      {
                          finishCalls.incrementAndGet();
                          indexers.forEach(Indexer::finish);
                      }
                  };
              }
  
              @Override
              public IndexWithSharedGroupQueryPlan queryPlanFor(RowFilter rowFilter)
              {
                  for (RowFilter.Expression e : rowFilter.getExpressions())
                  {
                      for (Index index : indexes.values())
                      {
                          if (index.supportsExpression(e))
                              return new IndexWithSharedGroupQueryPlan(index, index.getPostIndexQueryFilter(rowFilter), indexes.values());
                      }
                  }
  
                  return null;
              }
  
              @Override
              public SSTableFlushObserver getFlushObserver(Descriptor descriptor, ILifecycleTransaction txn, TableMetadata tableMetadata)
              {
                  Set<SSTableFlushObserver> observers = indexes.values()
                                                               .stream()
                                                               .map(i -> i.getFlushObserver(descriptor, txn))
                                                               .filter(Objects::nonNull)
                                                               .collect(Collectors.toSet());
  
                  return new SSTableFlushObserver() {
  
                      @Override
                      public void begin()
                      {
                          beginFlushCalls.incrementAndGet();
                          observers.forEach(SSTableFlushObserver::begin);
                      }
  
                      @Override
                      public void startPartition(DecoratedKey key, long position, long keyPositionForSASI)
                      {
                          flushedPartitions.incrementAndGet();
                          observers.forEach(o -> o.startPartition(key, position, keyPositionForSASI));
                      }
  
                      @Override
                      public void staticRow(Row staticRow)
                      {
                          flushedStaticRows.incrementAndGet();
                          observers.forEach(o -> o.staticRow(staticRow));
                      }
  
                      @Override
                      public void nextUnfilteredCluster(Unfiltered unfiltered)
                      {
                          flushedUnfiltereds.incrementAndGet();
                          observers.forEach(o -> o.nextUnfilteredCluster(unfiltered));
                      }
  
                      @Override
                      public void complete()
                      {
                          completeFlushCalls.incrementAndGet();
                          observers.forEach(SSTableFlushObserver::complete);
                      }
                  };
              }
  
              @Override
              public Set<Component> getComponents()
              {
                  return Collections.emptySet();
              }
          }
      }
  
      private static class IndexWithSharedGroupQueryPlan extends SingletonIndexQueryPlan
      {
          private final Set<Index> indexes;
  
          public <T extends Index> IndexWithSharedGroupQueryPlan(Index index, RowFilter postIndexFilter, Collection<T> indexes)
          {
              super(index, postIndexFilter);
              this.indexes = ImmutableSet.copyOf(indexes);
          }
  
          @Override
          public Set<Index> getIndexes()
          {
              return indexes;
          }
      }
  }
  