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

@@ -150,6 +150,8 @@
 import org.apache.cassandra.service.ActiveRepairService;
 import org.apache.cassandra.service.CacheService;
 import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.service.accord.LocalTransfers;
+import org.apache.cassandra.service.accord.PendingLocalTransfer;
 import org.apache.cassandra.service.paxos.Ballot;
 import org.apache.cassandra.service.paxos.PaxosRepairHistory;
 import org.apache.cassandra.service.paxos.TablePaxosRepairHistory;
@@ -545,6 +547,25 @@ public ColumnFamilyStore(Keyspace keyspace,
             data.addInitialSSTablesWithoutUpdatingSize(sstables);
         }
 
+        // Restores any pending SSTables from Accord bulk data transfer
+        Map<TimeUUID, List<File>> filePlanIDs = directories.getAccordBulkTransferPlanIds();
+        for (Map.Entry<TimeUUID, List<File>> entry: filePlanIDs.entrySet())
+        {
+            Set<SSTableReader> pendingSSTables = new HashSet<>();
+            for (File file : entry.getValue())
+            {
+                Directories.SSTableLister sstableLister = directories.sstableLister(file, Directories.OnTxnErr.IGNORE).skipTemporary(true);
+                pendingSSTables.addAll(SSTableReader.openAll(this, sstableLister.list(true).entrySet(), metadata));
+            }
+            if (pendingSSTables.isEmpty())
+            {
+                logger.info("SSTable import for TimeUUID {} is empty; removing the directory and skipping", entry.getValue());
+                continue;
+            }
+            PendingLocalTransfer pendingLocalTransfer = new PendingLocalTransfer(getTableId(), entry.getKey(), pendingSSTables);
+            LocalTransfers.instance.received(pendingLocalTransfer);
+        }
+
         // compaction strategy should be created after the CFS has been prepared
         compactionStrategyManager = new CompactionStrategyManager(this);
 
@@ -948,7 +969,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 +2229,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,84 @@ 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;
+    }
+
+    // Each TimeUUID can have multiple files spanning across different data directories
+    public Map<TimeUUID, List<File>> getAccordBulkTransferPlanIds()
+    {
+        Map<TimeUUID, List<File>> result = new HashMap<>();
+        Set<File> pendingLocations = getPendingLocations();
+        for (File pendingDir : pendingLocations)
+        {
+            // Each planID directory can contain multiple SSTables within it
+            for (File planID : pendingDir.listUnchecked())
+            {
+                TimeUUID timeUUID;
+                try
+                {
+                    timeUUID = TimeUUID.fromString(planID.name());
+                    if (!result.containsKey(timeUUID))
+                        result.put(timeUUID, new ArrayList<>());
+                    result.get(timeUUID).add(planID);
+                }
+                catch (IllegalArgumentException e)
+                {
+                    logger.warn("Unexpected: Invalid planID " + planID.name());
+                }
+            }
+        }
+
+        return result;
+    }
+
+    public List<File> getAccordBulkTransferSSTableDirectories()
+    {
+        List<File> result = new ArrayList<>();
+        Set<File> pendingLocations = getPendingLocations();
+        for (File pendingDir : pendingLocations)
+        {
+            for (File planID : pendingDir.listUnchecked())
+            {
+                try
+                {
+                    TimeUUID.fromString(planID.name());
+                    result.add(planID);
+                }
+                catch (IllegalArgumentException e)
+                {
+                    logger.warn("Unexpected: Invalid planID " + planID.name());
+                }
+            }
+        }
+
+        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 +891,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/lifecycle/LogTransaction.java

@@ -512,7 +512,10 @@ protected void doPrepare() { }
      */
     static boolean removeUnfinishedLeftovers(TableMetadata metadata)
     {
-        return removeUnfinishedLeftovers(new Directories(metadata).getCFDirectories());
+        Directories directories = new Directories(metadata);
+        List<File> files = directories.getCFDirectories();
+        files.addAll(directories.getAccordBulkTransferSSTableDirectories());
+        return removeUnfinishedLeftovers(files);
     }
 
     @VisibleForTesting
@@ -526,7 +529,7 @@ static boolean removeUnfinishedLeftovers(List<File> directories)
     private static final class LogFilesByName
     {
         // This maps a transaction log file name to a list of physical files. Each sstable
-        // can have multiple directories and a transaction is trakced by identical transaction log
+        // can have multiple directories and a transaction is tracked by identical transaction log
         // files, one per directory. So for each transaction file name we can have multiple
         // physical files.
         Map<String, List<File>> files = new HashMap<>();

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)