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.distributed.test;
  
  import java.io.IOException;
  import java.nio.file.Files;
  import java.util.Arrays;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Set;
  import java.util.function.Function;
  import java.util.stream.Collectors;
  
  import com.google.common.collect.ImmutableSet;
  import org.apache.commons.io.FileUtils;
  import org.junit.AfterClass;
  import org.junit.Before;
  import org.junit.BeforeClass;
  import org.junit.Test;
  
  import org.apache.cassandra.cql3.UntypedResultSet;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.db.Keyspace;
  import org.apache.cassandra.db.SystemKeyspace;
  import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
  import org.apache.cassandra.db.compaction.LeveledCompactionStrategy;
  import org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy;
  import org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy;
  import org.apache.cassandra.distributed.Cluster;
  import org.apache.cassandra.distributed.api.Feature;
  import org.apache.cassandra.distributed.api.IInstance;
  import org.apache.cassandra.distributed.api.IInvokableInstance;
  import org.apache.cassandra.distributed.api.SimpleQueryResult;
  import org.apache.cassandra.distributed.shared.ClusterUtils;
  import org.apache.cassandra.io.sstable.Descriptor;
  import org.apache.cassandra.io.sstable.SequenceBasedSSTableId;
  import org.apache.cassandra.io.sstable.UUIDBasedSSTableId;
  import org.apache.cassandra.io.util.File;
  import org.apache.cassandra.metrics.RestorableMeter;
  import org.apache.cassandra.service.snapshot.SnapshotManager;
  import org.apache.cassandra.service.snapshot.SnapshotOptions;
  import org.apache.cassandra.service.snapshot.TableSnapshot;
  import org.apache.cassandra.tools.SystemExitException;
  import org.apache.cassandra.utils.TimeUUID;
  import org.assertj.core.api.Assertions;
  import org.assertj.core.data.Offset;
  
  import static java.lang.String.format;
  import static org.apache.cassandra.Util.bulkLoadSSTables;
  import static org.apache.cassandra.Util.getBackups;
  import static org.apache.cassandra.Util.getSSTables;
  import static org.apache.cassandra.Util.relativizePath;
  import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
  import static org.apache.cassandra.db.SystemKeyspace.LEGACY_SSTABLE_ACTIVITY;
  import static org.apache.cassandra.db.SystemKeyspace.SSTABLE_ACTIVITY_V2;
  import static org.apache.cassandra.distributed.shared.FutureUtils.waitOn;
  import static org.apache.cassandra.distributed.test.ExecUtil.rethrow;
  import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
  import static org.assertj.core.api.Assertions.assertThat;
  
  public class SSTableIdGenerationTest extends TestBaseImpl
  {
      private final static String ENABLE_UUID_FIELD_NAME = "uuid_sstable_identifiers_enabled";
      private final static String SNAPSHOT_TAG = "test";
  
      private int v;
  
      private static SecurityManager originalSecurityManager;
  
      @BeforeClass
      public static void beforeClass() throws Throwable
      {
          TestBaseImpl.beforeClass();
  
          originalSecurityManager = System.getSecurityManager();
      }
  
      @Before
      public void beforeEach()
      {
          // we prevent system exit and convert it to exception becuase this is one of the expected test outcomes,
          // and we want to make an assertion on that
          ClusterUtils.preventSystemExit();
      }
  
      @AfterClass
      public static void afterClass() throws Throwable
      {
          System.setSecurityManager(originalSecurityManager);
      }
  
      /**
       * This test verifies that a node with uuid disabled actually creates sstables with sequential ids and
       * both the current and legacy sstable activity tables are updated.
       * Then, when enable uuid, we actually create sstables with uuid but keep and can read the old sstables. Also, only
       * update the current sstable activity table.
       */
      @Test
      public void testRestartWithUUIDEnabled() throws IOException
      {
          try (Cluster cluster = init(Cluster.build(1)
                                             .withDataDirCount(1)
                                             .withConfig(config -> config.set(ENABLE_UUID_FIELD_NAME, false))
                                             .start()))
          {
              cluster.schemaChange(createTableStmt(KEYSPACE, "tbl", null));
              createSSTables(cluster.get(1), KEYSPACE, "tbl", 1, 2);
              assertSSTablesCount(cluster.get(1), 2, 0, KEYSPACE, "tbl");
              verfiySSTableActivity(cluster, true);
  
              restartNode(cluster, 1, true);
  
              createSSTables(cluster.get(1), KEYSPACE, "tbl", 3, 4);
              assertSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, "tbl");
              verfiySSTableActivity(cluster, false);
  
              checkRowsNumber(cluster.get(1), KEYSPACE, "tbl", 9);
          }
      }
  
      /**
       * This test verifies that we should not be able to start a node with uuid disabled when there are uuid sstables
       */
      @Test
      public void testRestartWithUUIDDisabled() throws IOException
      {
          try (Cluster cluster = init(Cluster.build(1)
                                             .withDataDirCount(1)
                                             .withConfig(config -> config.set(ENABLE_UUID_FIELD_NAME, true))
                                             .start()))
          {
              cluster.schemaChange(createTableStmt(KEYSPACE, "tbl", null));
              for (IInvokableInstance instance : cluster)
              {
                  instance.runOnInstance(() -> {
                      for (ColumnFamilyStore cs : Keyspace.open(KEYSPACE).getColumnFamilyStores())
                          cs.disableAutoCompaction();
                  });
              }
              createSSTables(cluster.get(1), KEYSPACE, "tbl", 1, 2);
              assertSSTablesCount(cluster.get(1), 0, 2, KEYSPACE, "tbl");
              verfiySSTableActivity(cluster, false);
  
              Assertions.assertThatExceptionOfType(RuntimeException.class)
                        .isThrownBy(() -> restartNode(cluster, 1, false))
                        .withCauseInstanceOf(SystemExitException.class);
          }
      }
  
      @Test
      public final void testCompactionStrategiesWithMixedSSTables() throws Exception
      {
          testCompactionStrategiesWithMixedSSTables(SizeTieredCompactionStrategy.class,
                                                    TimeWindowCompactionStrategy.class,
                                                    LeveledCompactionStrategy.class);
      }
  
      /**
       * The purpose of this test is to verify that we can compact using the given strategy the mix of sstables created
       * with sequential id and with uuid. Then we verify whether the number results matches the number of rows which we
       * would get by merging data from the initial sstables.
       */
      @SafeVarargs
      private final void testCompactionStrategiesWithMixedSSTables(final Class<? extends AbstractCompactionStrategy>... compactionStrategyClasses) throws Exception
      {
          try (Cluster cluster = init(Cluster.build(1)
                                             .withDataDirCount(1)
                                             .withConfig(config -> config.set(ENABLE_UUID_FIELD_NAME, false))
                                             .start()))
          {
              // create a table and two sstables with sequential id for each strategy, the sstables will contain overlapping partitions
              for (Class<? extends AbstractCompactionStrategy> compactionStrategyClass : compactionStrategyClasses)
              {
                  String tableName = "tbl_" + toLowerCaseLocalized(compactionStrategyClass.getSimpleName());
                  cluster.schemaChange(createTableStmt(KEYSPACE, tableName, compactionStrategyClass));
  
                  createSSTables(cluster.get(1), KEYSPACE, tableName, 1, 2);
                  assertSSTablesCount(cluster.get(1), 2, 0, KEYSPACE, tableName);
              }
  
              // restart the node with uuid enabled
              restartNode(cluster, 1, true);
  
              // create another two sstables with uuid for each previously created table
              for (Class<? extends AbstractCompactionStrategy> compactionStrategyClass : compactionStrategyClasses)
              {
                  String tableName = "tbl_" + toLowerCaseLocalized(compactionStrategyClass.getSimpleName());
  
                  createSSTables(cluster.get(1), KEYSPACE, tableName, 3, 4);
  
                  // expect to have a mix of sstables with sequential id and uuid
                  assertSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, tableName);
  
                  // after compaction, we expect to have a single sstable with uuid
                  cluster.get(1).forceCompact(KEYSPACE, tableName);
                  assertSSTablesCount(cluster.get(1), 0, 1, KEYSPACE, tableName);
  
                  // verify the number of rows
                  checkRowsNumber(cluster.get(1), KEYSPACE, tableName, 9);
              }
          }
      }
  
      @Test
      public void testStreamingToNodeWithUUIDEnabled() throws Exception
      {
          testStreaming(true);
      }
  
      @Test
      public void testStreamingToNodeWithUUIDDisabled() throws Exception
      {
          testStreaming(false);
      }
  
      /**
       * The purpose of this test case is to verify the scenario when we need to stream mixed UUID and seq sstables to
       * a node which have: 1) UUID disabled, and 2) UUID enabled; then verify that we can read all the data properly
       * from that node alone.
       */
      private void testStreaming(boolean uuidEnabledOnTargetNode) throws Exception
      {
          // start both nodes with uuid disabled
          try (Cluster cluster = init(Cluster.build(2)
                                             .withDataDirCount(1)
                                             .withConfig(config -> config.set(ENABLE_UUID_FIELD_NAME, false).with(Feature.NETWORK))
                                             .start()))
          {
              // create an empty table and shutdown nodes 2, 3
              cluster.schemaChange(createTableStmt(KEYSPACE, "tbl", null));
              waitOn(cluster.get(2).shutdown());
  
              // create 2 sstables with overlapping partitions on node 1 (with seq ids)
              createSSTables(cluster.get(1), KEYSPACE, "tbl", 1, 2);
  
              // restart node 1 with uuid enabled
              restartNode(cluster, 1, true);
  
              // create 2 sstables with overlapping partitions on node 1 (with UUID ids)
              createSSTables(cluster.get(1), KEYSPACE, "tbl", 3, 4);
  
              assertSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, "tbl");
  
              // now start node with UUID disabled and perform repair
              cluster.get(2).config().set(ENABLE_UUID_FIELD_NAME, uuidEnabledOnTargetNode);
              cluster.get(2).startup();
  
              assertSSTablesCount(cluster.get(2), 0, 0, KEYSPACE, "tbl");
  
              // at this point we have sstables with seq and uuid on nodes and no sstables on node
              // when we run repair, we expect streaming all 4 sstables from node 1 to node 2
  
              cluster.get(2).nodetool("repair", KEYSPACE);
  
              if (uuidEnabledOnTargetNode)
                  assertSSTablesCount(cluster.get(2), 0, 4, KEYSPACE, "tbl");
              else
                  assertSSTablesCount(cluster.get(2), 4, 0, KEYSPACE, "tbl");
  
              waitOn(cluster.get(1).shutdown());
  
              checkRowsNumber(cluster.get(2), KEYSPACE, "tbl", 9);
          }
      }
  
      @Test
      public void testSnapshot() throws Exception
      {
          File tmpDir = new File(Files.createTempDirectory("test"));
          Set<String> seqOnlyBackupDirs;
          Set<String> seqAndUUIDBackupDirs;
          Set<String> uuidOnlyBackupDirs;
          try (Cluster cluster = init(Cluster.build(1)
                                             .withDataDirCount(1)
                                             .withConfig(config -> config.with(Feature.NETWORK)
                                                                         .set("incremental_backups", true)
                                                                         .set("snapshot_before_compaction", false)
                                                                         .set("auto_snapshot", false)
                                                                         .set(ENABLE_UUID_FIELD_NAME, false))
                                             .start()))
          {
              // create the tables
  
              cluster.schemaChange("CREATE KEYSPACE new_ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};");
  
              cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_seq_only", null));
              cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_seq_and_uuid", null));
              cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_uuid_only", null));
              cluster.schemaChange(createTableStmt("new_ks", "tbl_seq_only", null));
              cluster.schemaChange(createTableStmt("new_ks", "tbl_seq_and_uuid", null));
              cluster.schemaChange(createTableStmt("new_ks", "tbl_uuid_only", null));
  
              // creating sstables
              createSSTables(cluster.get(1), KEYSPACE, "tbl_seq_only", 1, 2, 3, 4);
              createSSTables(cluster.get(1), KEYSPACE, "tbl_seq_and_uuid", 1, 2);
              createSSTables(cluster.get(1), "new_ks", "tbl_seq_only", 5, 6, 7, 8);
              createSSTables(cluster.get(1), "new_ks", "tbl_seq_and_uuid", 5, 6);
  
              restartNode(cluster, 1, true);
  
              createSSTables(cluster.get(1), KEYSPACE, "tbl_seq_and_uuid", 3, 4);
              createSSTables(cluster.get(1), KEYSPACE, "tbl_uuid_only", 1, 2, 3, 4);
              createSSTables(cluster.get(1), "new_ks", "tbl_seq_and_uuid", 7, 8);
              createSSTables(cluster.get(1), "new_ks", "tbl_uuid_only", 5, 6, 7, 8);
  
              Set<String> seqOnlySnapshotDirs = snapshot(cluster.get(1), KEYSPACE, "tbl_seq_only");
              Set<String> seqAndUUIDSnapshotDirs = snapshot(cluster.get(1), KEYSPACE, "tbl_seq_and_uuid");
              Set<String> uuidOnlySnapshotDirs = snapshot(cluster.get(1), KEYSPACE, "tbl_uuid_only");
  
              seqOnlyBackupDirs = getBackupDirs(cluster.get(1), KEYSPACE, "tbl_seq_only");
              seqAndUUIDBackupDirs = getBackupDirs(cluster.get(1), KEYSPACE, "tbl_seq_and_uuid");
              uuidOnlyBackupDirs = getBackupDirs(cluster.get(1), KEYSPACE, "tbl_uuid_only");
  
              // at this point, we should have sstables with backups and snapshots for all tables
              assertSSTablesCount(cluster.get(1), 4, 0, KEYSPACE, "tbl_seq_only");
              assertSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, "tbl_seq_and_uuid");
              assertSSTablesCount(cluster.get(1), 0, 4, KEYSPACE, "tbl_uuid_only");
  
              assertBackupSSTablesCount(cluster.get(1), 4, 0, KEYSPACE, "tbl_seq_only");
              assertBackupSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, "tbl_seq_and_uuid");
              assertBackupSSTablesCount(cluster.get(1), 0, 4, KEYSPACE, "tbl_uuid_only");
  
              assertSnapshotSSTablesCount(cluster.get(1), 4, 0, KEYSPACE, "tbl_seq_only");
              assertSnapshotSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, "tbl_seq_and_uuid");
              assertSnapshotSSTablesCount(cluster.get(1), 0, 4, KEYSPACE, "tbl_uuid_only");
  
              checkRowsNumber(cluster.get(1), KEYSPACE, "tbl_seq_only", 9);
              checkRowsNumber(cluster.get(1), KEYSPACE, "tbl_seq_and_uuid", 9);
              checkRowsNumber(cluster.get(1), KEYSPACE, "tbl_uuid_only", 9);
  
              // truncate the first set of tables
              truncateAndAssertEmpty(cluster.get(1), KEYSPACE, "tbl_seq_only", "tbl_seq_and_uuid", "tbl_uuid_only");
  
              restore(cluster.get(1), seqOnlySnapshotDirs, "tbl_seq_only", 9);
              restore(cluster.get(1), seqAndUUIDSnapshotDirs, "tbl_seq_and_uuid", 9);
              restore(cluster.get(1), uuidOnlySnapshotDirs, "tbl_uuid_only", 9);
  
              truncateAndAssertEmpty(cluster.get(1), KEYSPACE, "tbl_seq_only", "tbl_seq_and_uuid", "tbl_uuid_only");
  
              restore(cluster.get(1), seqOnlyBackupDirs, "tbl_seq_only", 9);
              restore(cluster.get(1), seqAndUUIDBackupDirs, "tbl_seq_and_uuid", 9);
              restore(cluster.get(1), uuidOnlyBackupDirs, "tbl_uuid_only", 9);
  
              ImmutableSet<String> allBackupDirs = ImmutableSet.<String>builder().addAll(seqOnlyBackupDirs).addAll(seqAndUUIDBackupDirs).addAll(uuidOnlyBackupDirs).build();
              cluster.get(1).runOnInstance(rethrow(() -> allBackupDirs.forEach(dir -> bulkLoadSSTables(new File(dir), "new_ks"))));
  
              checkRowsNumber(cluster.get(1), "new_ks", "tbl_seq_only", 17);
              checkRowsNumber(cluster.get(1), "new_ks", "tbl_seq_and_uuid", 17);
              checkRowsNumber(cluster.get(1), "new_ks", "tbl_uuid_only", 17);
  
  
              for (String dir : allBackupDirs)
              {
                  File src = new File(dir);
                  File dest = relativizePath(tmpDir, src, 3);
                  Files.createDirectories(dest.parent().toPath());
                  FileUtils.moveDirectory(src.toJavaIOFile(), dest.toJavaIOFile());
              }
          }
  
          try (Cluster cluster = init(Cluster.build(1)
                                             .withDataDirCount(1)
                                             .withConfig(config -> config.with(Feature.NETWORK, Feature.NATIVE_PROTOCOL)
                                                                         .set("incremental_backups", true)
                                                                         .set("snapshot_before_compaction", false)
                                                                         .set("auto_snapshot", false)
                                                                         .set(ENABLE_UUID_FIELD_NAME, false))
                                             .start()))
          {
              cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_seq_only", null));
              cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_seq_and_uuid", null));
              cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_uuid_only", null));
  
              Function<String, String> relativeToTmpDir = d -> relativizePath(tmpDir, new File(d), 3).toString();
              restore(cluster.get(1), seqOnlyBackupDirs.stream().map(relativeToTmpDir).collect(Collectors.toSet()), "tbl_seq_only", 9);
              restore(cluster.get(1), seqAndUUIDBackupDirs.stream().map(relativeToTmpDir).collect(Collectors.toSet()), "tbl_seq_and_uuid", 9);
              restore(cluster.get(1), uuidOnlyBackupDirs.stream().map(relativeToTmpDir).collect(Collectors.toSet()), "tbl_uuid_only", 9);
          }
      }
  
      private static void restore(IInvokableInstance instance, Set<String> dirs, String targetTableName, int expectedRowsNum)
      {
          List<String> failedImports = instance.callOnInstance(() -> ColumnFamilyStore.getIfExists(KEYSPACE, targetTableName)
                                                                                      .importNewSSTables(dirs, false, false, true, true, true, true, true));
          assertThat(failedImports).isEmpty();
          checkRowsNumber(instance, KEYSPACE, targetTableName, expectedRowsNum);
      }
  
      private static void truncateAndAssertEmpty(IInvokableInstance instance, String ks, String... tableNames)
      {
          for (String tableName : tableNames)
          {
              instance.executeInternal(format("TRUNCATE %s.%s", ks, tableName));
              assertSSTablesCount(instance, 0, 0, ks, tableName);
              checkRowsNumber(instance, ks, tableName, 0);
          }
      }
  
      private static Set<String> snapshot(IInvokableInstance instance, String ks, String tableName)
      {
          Set<String> snapshotDirs = instance.callOnInstance(() -> {
  
              ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(ks, tableName);
  
              if (cfs == null)
                  return Set.of();
  
              TableSnapshot tableSnapshot = SnapshotManager.instance.takeSnapshot(SnapshotOptions.userSnapshot(SNAPSHOT_TAG, cfs.getKeyspaceTableName())).iterator().next();
              Set<String> dirs = new HashSet<>();
              for (File dir : tableSnapshot.getDirectories())
                  dirs.add(dir.toString());
  
              return dirs;
          });
          assertThat(snapshotDirs).isNotEmpty();
          return snapshotDirs;
      }
  
      private static String createTableStmt(String ks, String name, Class<? extends AbstractCompactionStrategy> compactionStrategy)
      {
          if (compactionStrategy == null)
              compactionStrategy = SizeTieredCompactionStrategy.class;
          return format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) " +
                        "WITH compaction = {'class':'%s', 'enabled':'false'}",
                        ks, name, compactionStrategy.getCanonicalName());
      }
  
      private void createSSTables(IInstance instance, String ks, String tableName, int... records)
      {
          String insert = format("INSERT INTO %s.%s (pk, ck, v) VALUES (?, ?, ?)", ks, tableName);
          for (int record : records)
          {
              instance.executeInternal(insert, record, record, ++v);
              instance.executeInternal(insert, record, record + 1, ++v);
              instance.executeInternal(insert, record + 1, record + 1, ++v);
              instance.flush(ks);
          }
      }
  
      private static void assertSSTablesCount(Set<Descriptor> descs, String tableName, int expectedSeqGenIds, int expectedUUIDGenIds)
      {
          List<String> seqSSTables = descs.stream()
                                          .filter(desc -> desc.id instanceof SequenceBasedSSTableId)
                                          .map(descriptor -> descriptor.baseFile().toString())
                                          .sorted()
                                          .collect(Collectors.toList());
          List<String> uuidSSTables = descs.stream()
                                           .filter(desc -> desc.id instanceof UUIDBasedSSTableId)
                                           .map(descriptor -> descriptor.baseFile().toString())
                                           .sorted()
                                           .collect(Collectors.toList());
          assertThat(seqSSTables).describedAs("SSTables of %s with sequence based id", tableName).hasSize(expectedSeqGenIds);
          assertThat(uuidSSTables).describedAs("SSTables of %s with UUID based id", tableName).hasSize(expectedUUIDGenIds);
      }
  
      private static void assertSSTablesCount(IInvokableInstance instance, int expectedSeqGenIds, int expectedUUIDGenIds, String ks, String... tableNames)
      {
          instance.runOnInstance(rethrow(() -> Arrays.stream(tableNames).forEach(tableName -> assertSSTablesCount(getSSTables(ks, tableName), tableName, expectedSeqGenIds, expectedUUIDGenIds))));
      }
  
      private static void assertSnapshotSSTablesCount(IInvokableInstance instance, int expectedSeqGenIds, int expectedUUIDGenIds, String ks, String... tableNames)
      {
          instance.runOnInstance(rethrow(() -> Arrays.stream(tableNames).forEach(tableName -> assertSSTablesCount(TableSnapshot.getSnapshotDescriptors(ks, tableName, SNAPSHOT_TAG), tableName, expectedSeqGenIds, expectedUUIDGenIds))));
      }
  
      private static void assertBackupSSTablesCount(IInvokableInstance instance, int expectedSeqGenIds, int expectedUUIDGenIds, String ks, String... tableNames)
      {
          instance.runOnInstance(rethrow(() -> Arrays.stream(tableNames).forEach(tableName -> assertSSTablesCount(getBackups(ks, tableName), tableName, expectedSeqGenIds, expectedUUIDGenIds))));
      }
  
      private static Set<String> getBackupDirs(IInvokableInstance instance, String ks, String tableName)
      {
          return instance.callOnInstance(() -> getBackups(ks, tableName).stream()
                                                                        .map(d -> d.directory)
                                                                        .map(File::toString)
                                                                        .collect(Collectors.toSet()));
      }
  
      private static void verfiySSTableActivity(Cluster cluster, boolean expectLegacyTableIsPopulated)
      {
          cluster.get(1).runOnInstance(() -> {
              RestorableMeter meter = new RestorableMeter(15, 120);
              SequenceBasedSSTableId seqGenId = new SequenceBasedSSTableId(1);
              SystemKeyspace.persistSSTableReadMeter("ks", "tab", seqGenId, meter);
              assertThat(SystemKeyspace.getSSTableReadMeter("ks", "tab", seqGenId)).matches(m -> m.fifteenMinuteRate() == meter.fifteenMinuteRate()
                                                                                                 && m.twoHourRate() == meter.twoHourRate());
  
              checkSSTableActivityRow(SSTABLE_ACTIVITY_V2, seqGenId.toString(), true);
              if (expectLegacyTableIsPopulated)
                  checkSSTableActivityRow(LEGACY_SSTABLE_ACTIVITY, seqGenId.generation, true);
  
              SystemKeyspace.clearSSTableReadMeter("ks", "tab", seqGenId);
  
              checkSSTableActivityRow(SSTABLE_ACTIVITY_V2, seqGenId.toString(), false);
              if (expectLegacyTableIsPopulated)
                  checkSSTableActivityRow(LEGACY_SSTABLE_ACTIVITY, seqGenId.generation, false);
  
              UUIDBasedSSTableId uuidGenId = new UUIDBasedSSTableId(TimeUUID.Generator.nextTimeUUID());
              SystemKeyspace.persistSSTableReadMeter("ks", "tab", uuidGenId, meter);
              assertThat(SystemKeyspace.getSSTableReadMeter("ks", "tab", uuidGenId)).matches(m -> m.fifteenMinuteRate() == meter.fifteenMinuteRate()
                                                                                                  && m.twoHourRate() == meter.twoHourRate());
  
              checkSSTableActivityRow(SSTABLE_ACTIVITY_V2, uuidGenId.toString(), true);
  
              SystemKeyspace.clearSSTableReadMeter("ks", "tab", uuidGenId);
  
              checkSSTableActivityRow(SSTABLE_ACTIVITY_V2, uuidGenId.toString(), false);
          });
      }
  
      private static void checkSSTableActivityRow(String table, Object genId, boolean expectExists)
      {
          String tableColName = SSTABLE_ACTIVITY_V2.equals(table) ? "table_name" : "columnfamily_name";
          String idColName = SSTABLE_ACTIVITY_V2.equals(table) ? "id" : "generation";
          String cql = "SELECT rate_15m, rate_120m FROM system.%s WHERE keyspace_name=? and %s=? and %s=?";
          UntypedResultSet results = executeInternal(format(cql, table, tableColName, idColName), "ks", "tab", genId);
          assertThat(results).isNotNull();
  
          if (expectExists)
          {
              assertThat(results.isEmpty()).isFalse();
              UntypedResultSet.Row row = results.one();
              assertThat(row.getDouble("rate_15m")).isEqualTo(15d, Offset.offset(0.001d));
              assertThat(row.getDouble("rate_120m")).isEqualTo(120d, Offset.offset(0.001d));
          }
          else
          {
              assertThat(results.isEmpty()).isTrue();
          }
      }
  
      private static void restartNode(Cluster cluster, int node, boolean uuidEnabled)
      {
          waitOn(cluster.get(node).shutdown());
          cluster.get(node).config().set(ENABLE_UUID_FIELD_NAME, uuidEnabled);
          cluster.get(node).startup();
      }
  
      private static void checkRowsNumber(IInstance instance, String ks, String tableName, int expectedNumber)
      {
          SimpleQueryResult result = instance.executeInternalWithResult(format("SELECT * FROM %s.%s", ks, tableName));
          Object[][] rows = result.toObjectArrays();
          assertThat(rows).withFailMessage("Invalid results for %s.%s - should have %d rows but has %d: \n%s", ks, tableName, expectedNumber,
                                           rows.length, result.toString()).hasNumberOfRows(expectedNumber);
      }
  }
  