test/distributed/org/apache/cassandra/distributed/test/IndexDroppingTest.java

/*
 * 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.distributed.test;

import java.io.IOException;
import java.time.Instant;
import java.util.concurrent.TimeUnit;

import com.google.common.util.concurrent.Uninterruptibles;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;

import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;

import static java.time.temporal.ChronoUnit.MINUTES;
import static org.apache.cassandra.distributed.api.ConsistencyLevel.ONE;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.junit.Assert.assertEquals;

/**
 * Tests CASSANDRA-18105.
 * <p>
 * Truncation will insert an entry to system.local into truncated_at map for given table. Because id of an index
 * is same as id of the base table, when dropping an index, it in fact drops a table with same id as the base table
 * which will remove an entry for truncated_at column. Hence, after restart of the node, querying the base table
 * will resurrect the data because commitlog was fully replayed, not taking into consideration (now removed) entry
 * in truncated_at column.
 * <p>
 * The fix consists of not removing an entry in truncated_at column in system.local
 * if the table being removed is an index.
 */
public class IndexDroppingTest extends TestBaseImpl
{

    private static Cluster CLUSTER;

    @BeforeClass
    public static void init() throws IOException
    {
        CLUSTER = Cluster.build(1).withConfig(conf -> conf.with(NETWORK).set("materialized_views_enabled", "true")).start();
        CLUSTER.get(1).runOnInstance((IIsolatedExecutor.SerializableRunnable) () -> CompactionManager.instance.disableAutoCompaction());
    }

    @AfterClass
    public static void shutdown()
    {
        if (CLUSTER != null)
            CLUSTER.close();
    }

    @Test
    public void testIndexDropping() throws Throwable
    {
        CLUSTER.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE + " WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }");
        CLUSTER.schemaChange("CREATE TABLE " + KEYSPACE + ".tb1 (c3 TEXT, c4 TEXT, c2 INT, c1 TEXT, PRIMARY KEY (c1, c2, c3 ))");
        CLUSTER.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tb1 (c3, c1, c2) VALUES ('val1','val2',1)", ONE);
        CLUSTER.schemaChange("CREATE INDEX tb ON " + KEYSPACE + ".tb1 (c3)");

        waitForIndex(KEYSPACE, "tb1", "tb");

        CLUSTER.schemaChange("TRUNCATE TABLE " + KEYSPACE + " .tb1");
        CLUSTER.schemaChange("DROP INDEX " + KEYSPACE + ".tb");

        assertEmptyTable("tb1");
        restart();
        assertEmptyTable("tb1");
    }

    private void restart() throws Throwable
    {
        CLUSTER.get(1).shutdown(true).get();
        CLUSTER.get(1).startup();
        CLUSTER.get(1).runOnInstance((IIsolatedExecutor.SerializableRunnable) () ->
                                                                              CompactionManager.instance.disableAutoCompaction());
    }

    private void assertEmptyTable(String table)
    {
        assertEquals(0, CLUSTER.coordinator(1).execute("SELECT * FROM " + KEYSPACE + " ." + table, ONE).length);
    }

    private static void waitForIndex(String keyspace, String table, String indexName)
    {
        Instant startTime = Instant.now();
        Instant maxTime = startTime.plus(10, MINUTES);

        boolean builtIndex = false;
        while (!builtIndex)
        {
            if (maxTime.isBefore(Instant.now()))
                return;

            Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS);
            builtIndex = CLUSTER.get(1).applyOnInstance((IIsolatedExecutor.SerializableTriFunction<String, String, String, Boolean>) (ks, tbl, idx) -> {
                ColumnFamilyStore cf = ColumnFamilyStore.getIfExists(ks, tbl);
                if (cf == null)
                    return false;
                return cf.indexManager.getBuiltIndexNames().contains(idx);
            }, keyspace, table, indexName);
        }
    }
}