CASSANDRA-20595 SSTable import doesn't work with Accord transactions

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "modules/accord"]
 	path = modules/accord
-	url = https://github.com/apache/cassandra-accord.git
-	branch = trunk
+	url = https://github.com/alanwang67/cassandra-accord.git
+	branch = CASSANDRA-20595

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

@@ -948,7 +948,7 @@ public List<String> importNewSSTables(Set<String> srcPaths, boolean resetLevel,
                                                                         .build());
     }
 
-    Descriptor getUniqueDescriptorFor(Descriptor descriptor, File targetDirectory)
+    public Descriptor getUniqueDescriptorFor(Descriptor descriptor, File targetDirectory)
     {
         Descriptor newDescriptor;
         do
@@ -2208,6 +2208,31 @@ private void invalidateCaches()
             CacheService.instance.invalidateCounterCacheForCf(metadata());
     }
 
+    public void invalidateRowAndCounterCache(Collection<SSTableReader> sstables, Consumer<Integer> onRowCacheInvalidation, Consumer<Integer> onCounterCacheInvalidation)
+    {
+        if (isRowCacheEnabled() || metadata().isCounter())
+        {
+            List<Bounds<Token>> boundsToInvalidate = new ArrayList<>(sstables.size());
+            sstables.forEach(sstable -> boundsToInvalidate.add(new Bounds<>(sstable.getFirst().getToken(), sstable.getLast().getToken())));
+            Set<Bounds<Token>> nonOverlappingBounds = Bounds.getNonOverlappingBounds(boundsToInvalidate);
+
+            if (isRowCacheEnabled())
+            {
+                int invalidatedKeys = invalidateRowCache(nonOverlappingBounds);
+                if (invalidatedKeys > 0)
+                    onRowCacheInvalidation.accept(invalidatedKeys);
+            }
+
+            if (metadata().isCounter())
+            {
+                int invalidatedKeys = invalidateCounterCache(nonOverlappingBounds);
+                if (invalidatedKeys > 0)
+                    onCounterCacheInvalidation.accept(invalidatedKeys);
+            }
+        }
+    }
+
+
     public int invalidateRowCache(Collection<Bounds<Token>> boundsToInvalidate)
     {
         int invalidatedKeys = 0;

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

@@ -74,6 +74,7 @@
 import org.apache.cassandra.service.snapshot.SnapshotManifest;
 import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.TimeUUID;
 
 import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
 
@@ -117,6 +118,7 @@ public class Directories
     public static final String BACKUPS_SUBDIR = "backups";
     public static final String SNAPSHOT_SUBDIR = "snapshots";
     public static final String TMP_SUBDIR = "tmp";
+    public static final String PENDING_SUBDIR = "pending";
     public static final String SECONDARY_INDEX_NAME_SEPARATOR = ".";
     public static final String TABLE_DIRECTORY_NAME_SEPARATOR = "-";
 
@@ -316,10 +318,7 @@ public File getLocationForDisk(DataDirectory dataDirectory)
         if (dataDirectory != null)
             for (File dir : dataPaths)
             {
-                // Note that we must compare absolute paths (not canonical) here since keyspace directories might be symlinks
-                Path dirPath = dir.toAbsolute().toPath();
-                Path locationPath = dataDirectory.location.toAbsolute().toPath();
-                if (dirPath.startsWith(locationPath))
+                if (dataDirectory.contains(dir))
                     return dir;
             }
         return null;
@@ -726,6 +725,33 @@ public static File getSnapshotSchemaFile(File snapshotDir)
         return new File(snapshotDir, "schema.cql");
     }
 
+    @VisibleForTesting
+    public Set<File> getPendingLocations()
+    {
+        Set<File> result = new HashSet<>();
+        for (DataDirectory dataDirectory : dataDirectories.getAllDirectories())
+        {
+            for (File dir : dataPaths)
+            {
+                if (!dataDirectory.contains(dir))
+                    continue;
+                result.add(getOrCreate(dir, PENDING_SUBDIR));
+            }
+        }
+        return result;
+    }
+
+    public File getPendingLocationForDisk(DataDirectory dataDirectory, TimeUUID planId)
+    {
+        for (File dir : dataPaths)
+        {
+            if (!dataDirectory.contains(dir))
+                continue;
+            return getOrCreate(dir, PENDING_SUBDIR, planId.toString());
+        }
+        throw new RuntimeException("Could not find pending location");
+    }
+
     public static File getBackupsDirectory(Descriptor desc)
     {
         return getBackupsDirectory(desc.directory);
@@ -814,6 +840,13 @@ public DataDirectory(Path location)
             this.location = new File(location);
         }
 
+        public boolean contains(File file)
+        {
+            // Note that we must compare absolute paths (not canonical) here since keyspace directories might be symlinks
+            Path path = file.toAbsolute().toPath();
+            return path.startsWith(location.toAbsolute().toPath());
+        }
+
         public long getAvailableSpace()
         {
             long availableSpace = PathUtils.tryGetSpace(location.toPath(), FileStore::getUsableSpace) - DatabaseDescriptor.getMinFreeSpacePerDriveInBytes();

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

@@ -45,8 +45,10 @@
 import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.ActiveRepairService;
 import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.service.accord.AccordService;
 import org.apache.cassandra.utils.OutputHandler;
 import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.concurrent.Refs;
@@ -80,6 +82,8 @@ synchronized List<String> importNewSSTables(Options options)
         UUID importID = UUID.randomUUID();
         logger.info("[{}] Loading new SSTables for {}/{}: {}", importID, cfs.getKeyspaceName(), cfs.getTableName(), options);
 
+        TableMetadata metadata = cfs.metadata();
+        boolean isAccordEnabled = metadata.isAccordEnabled();
         List<Pair<Directories.SSTableLister, String>> listers = getSSTableListers(options.srcPaths);
 
         Set<Descriptor> currentDescriptors = new HashSet<>();
@@ -174,11 +178,14 @@ synchronized List<String> importNewSSTables(Options options)
                     if (currentDescriptors.contains(oldDescriptor))
                         continue;
 
-                    File targetDir = getTargetDirectory(dir, oldDescriptor, entry.getValue());
+                    File targetDir = dir == null ? oldDescriptor.directory : getTargetDirectory(cfs, oldDescriptor, entry.getValue());
                     Descriptor newDescriptor = cfs.getUniqueDescriptorFor(entry.getKey(), targetDir);
                     maybeMutateMetadata(entry.getKey(), options);
                     movedSSTables.add(new MovedSSTable(newDescriptor, entry.getKey(), entry.getValue()));
-                    SSTableReader sstable = SSTableReader.moveAndOpenSSTable(cfs, entry.getKey(), newDescriptor, entry.getValue(), options.copyData);
+                    // Don't move tracked SSTables, since that will move them to the live set on bounce
+                    SSTableReader sstable = isAccordEnabled
+                                            ? SSTableReader.open(cfs, oldDescriptor, metadata.ref)
+                                            : SSTableReader.moveAndOpenSSTable(cfs, oldDescriptor, newDescriptor, entry.getValue(), options.copyData);
                     newSSTablesPerDirectory.add(sstable);
                 }
                 catch (Throwable t)
@@ -228,7 +235,10 @@ synchronized List<String> importNewSSTables(Options options)
             if (!cfs.indexManager.validateSSTableAttachedIndexes(newSSTables, false, options.validateIndexChecksum))
                 cfs.indexManager.buildSSTableAttachedIndexesBlocking(newSSTables);
 
-            cfs.getTracker().addSSTables(newSSTables);
+            if (isAccordEnabled)
+                AccordService.instance().executeTransfer(importID, options.copyData, cfs.keyspace.getName(), newSSTables, metadata);
+            else
+                cfs.getTracker().addSSTables(newSSTables);
             for (SSTableReader reader : newSSTables)
             {
                 if (options.invalidateCaches && cfs.isRowCacheEnabled())
@@ -237,8 +247,16 @@ synchronized List<String> importNewSSTables(Options options)
         }
         catch (Throwable t)
         {
-            logger.error("[{}] Failed adding SSTables", importID, t);
-            throw new RuntimeException("Failed adding SSTables", t);
+            if (isAccordEnabled)
+            {
+                String msg = "Failed adding SSTables on local node; note the import may still have been committed by a recovery coordinator";
+                throw new RuntimeException(msg, t);
+            }
+            else
+            {
+                logger.error("[{}] Failed adding SSTables", importID, t);
+                throw new RuntimeException("Failed adding SSTables", t);
+            }
         }
 
         logger.info("[{}] Done loading load new SSTables for {}/{}", importID, cfs.getKeyspaceName(), cfs.getTableName());
@@ -282,15 +300,10 @@ private static String formatMetadata(SSTableReader sstable)
      * Opens the sstablereader described by descriptor and figures out the correct directory for it based
      * on the first token
      *
-     * srcPath == null means that the sstable is in a data directory and we can use that directly.
-     *
      * If we fail figuring out the directory we will pick the one with the most available disk space.
      */
-    private File getTargetDirectory(String srcPath, Descriptor descriptor, Set<Component> components)
+    public static File getTargetDirectory(ColumnFamilyStore cfs, Descriptor descriptor, Set<Component> components)
     {
-        if (srcPath == null)
-            return descriptor.directory;
-
         File targetDirectory = null;
         SSTableReader sstable = null;
         try
@@ -339,13 +352,13 @@ private List<Pair<Directories.SSTableLister, String>> getSSTableListers(Set<Stri
         return listers;
     }
 
-    private static class MovedSSTable
+    public static class MovedSSTable
     {
         private final Descriptor newDescriptor;
         private final Descriptor oldDescriptor;
         private final Set<Component> components;
 
-        private MovedSSTable(Descriptor newDescriptor, Descriptor oldDescriptor, Set<Component> components)
+        public MovedSSTable(Descriptor newDescriptor, Descriptor oldDescriptor, Set<Component> components)
         {
             this.newDescriptor = newDescriptor;
             this.oldDescriptor = oldDescriptor;
@@ -362,7 +375,7 @@ public String toString()
      * If we fail when opening the sstable (if for example the user passes in --no-verify and there are corrupt sstables)
      * we might have started copying sstables to the data directory, these need to be moved back to the original name/directory
      */
-    private void moveSSTablesBack(Set<MovedSSTable> movedSSTables)
+    public static void moveSSTablesBack(Set<MovedSSTable> movedSSTables)
     {
         for (MovedSSTable movedSSTable : movedSSTables)
         {
@@ -381,7 +394,7 @@ private void moveSSTablesBack(Set<MovedSSTable> movedSSTables)
      *
      * @param movedSSTables tables we have moved already (by copying) which need to be removed
      */
-    private void removeCopiedSSTables(Set<MovedSSTable> movedSSTables)
+    public static void removeCopiedSSTables(Set<MovedSSTable> movedSSTables)
     {
         logger.debug("Removing copied SSTables which were left in data directories after failed SSTable import.");
         for (MovedSSTable movedSSTable : movedSSTables)

src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java

@@ -45,6 +45,7 @@
 import org.apache.cassandra.io.util.SequentialWriterOption;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.streaming.ProgressInfo;
+import org.apache.cassandra.streaming.StreamOperation;
 import org.apache.cassandra.streaming.StreamReceiver;
 import org.apache.cassandra.streaming.StreamSession;
 import org.apache.cassandra.streaming.messages.StreamMessageHeader;
@@ -159,10 +160,14 @@ public SSTableMultiWriter read(DataInputPlus in) throws IOException
 
     private File getDataDir(ColumnFamilyStore cfs, long totalSize) throws IOException
     {
+        boolean performingAccordBulkDataImport = cfs.metadata().isAccordEnabled() && session.streamOperation() == StreamOperation.ACCORD_SSTABLE_IMPORT;
         Directories.DataDirectory localDir = cfs.getDirectories().getWriteableLocation(totalSize);
         if (localDir == null)
             throw new IOException(format("Insufficient disk space to store %s", prettyPrintMemory(totalSize)));
 
+        if (performingAccordBulkDataImport)
+            return cfs.getDirectories().getPendingLocationForDisk(localDir, session.planId());
+
         File dir = cfs.getDirectories().getLocationForDisk(cfs.getDiskBoundaries().getCorrectDiskForKey(header.firstKey));
 
         if (dir == null)

src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java

@@ -47,6 +47,7 @@
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.UnknownColumnException;
+import org.apache.cassandra.io.sstable.RangeAwarePendingSSTableWriter;
 import org.apache.cassandra.io.sstable.RangeAwareSSTableWriter;
 import org.apache.cassandra.io.sstable.SSTableMultiWriter;
 import org.apache.cassandra.io.sstable.SSTableSimpleIterator;
@@ -61,6 +62,7 @@
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.streaming.ProgressInfo;
+import org.apache.cassandra.streaming.StreamOperation;
 import org.apache.cassandra.streaming.StreamReceivedOutOfTokenRangeException;
 import org.apache.cassandra.streaming.StreamReceiver;
 import org.apache.cassandra.streaming.StreamSession;
@@ -185,7 +187,15 @@ protected SSTableTxnSingleStreamWriter createWriter(ColumnFamilyStore cfs, long
         StreamReceiver streamReceiver = session.getAggregator(tableId);
         Preconditions.checkState(streamReceiver instanceof CassandraStreamReceiver);
         ILifecycleTransaction txn = createTxn();
-        RangeAwareSSTableWriter writer = new RangeAwareSSTableWriter(cfs, estimatedKeys, repairedAt, pendingRepair, false, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()));
+        RangeAwareSSTableWriter writer;
+        if (session.streamOperation() == StreamOperation.ACCORD_SSTABLE_IMPORT)
+        {
+            Preconditions.checkState(cfs.metadata().isAccordEnabled());
+            writer = new RangeAwarePendingSSTableWriter(cfs, estimatedKeys, repairedAt, pendingRepair, false, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()), session.planId());
+        }
+        else
+            writer = new RangeAwareSSTableWriter(cfs, estimatedKeys, repairedAt, pendingRepair, false, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()));
+
         return new SSTableTxnSingleStreamWriter(txn, writer);
     }
 

src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java

@@ -32,6 +32,7 @@
 
 import accord.primitives.Ranges;
 import accord.primitives.TxnId;
+import accord.utils.Invariants;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ColumnFamilyStore;
@@ -51,9 +52,11 @@
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.service.accord.AccordService;
 import org.apache.cassandra.service.accord.IAccordService;
+import org.apache.cassandra.service.accord.PendingLocalTransfer;
 import org.apache.cassandra.service.accord.TimeOnlyRequestBookkeeping.LatencyRequestBookkeeping;
 import org.apache.cassandra.service.accord.topology.AccordTopology;
 import org.apache.cassandra.streaming.IncomingStream;
+import org.apache.cassandra.streaming.StreamOperation;
 import org.apache.cassandra.streaming.StreamReceiver;
 import org.apache.cassandra.streaming.StreamSession;
 import org.apache.cassandra.tcm.ClusterMetadata;
@@ -115,6 +118,7 @@ private static CassandraIncomingFile getFile(IncomingStream stream)
         return (CassandraIncomingFile) stream;
     }
 
+    // This method is called for every SSTable within a stream
     @Override
     public synchronized void received(IncomingStream stream)
     {
@@ -242,6 +246,7 @@ public void finished()
         {
             if (requiresWritePath)
             {
+                Invariants.require(session.streamOperation() != StreamOperation.ACCORD_SSTABLE_IMPORT);
                 sendThroughWritePath(cfs, readers);
             }
             else
@@ -257,6 +262,16 @@ public void finished()
 
                 // add sstables (this will build non-SSTable-attached secondary indexes too, see CASSANDRA-10130)
                 logger.debug("[Stream #{}] Received {} sstables from {} ({})", session.planId(), readers.size(), session.peer, readers);
+
+                // Accord will coordinate marking these SSTables as live
+                if (session.streamOperation() == StreamOperation.ACCORD_SSTABLE_IMPORT)
+                {
+                    Preconditions.checkState(cfs.metadata().isAccordEnabled());
+                    PendingLocalTransfer transfer = new PendingLocalTransfer(cfs.metadata().id, session.planId(), sstables);
+                    AccordService.instance().receivedSSTableImport(transfer);
+                    return;
+                }
+
                 cfs.addSSTables(readers);
 
                 //invalidate row and counter cache