From 93f490689e4afbfd16d8531ce5e9c07cbc35f6c1 Mon Sep 17 00:00:00 2001 From: Lukasz Antoniak Date: Fri, 28 Aug 2026 17:09:24 +0200 Subject: [PATCH 1/3] CASSANALYTICS-104: Eliminate redundant filesystem lookups in SSTable direct streaming --- .../spark/bulkwriter/DirectStreamSession.java | 87 ++++----- .../spark/bulkwriter/SortedSSTableWriter.java | 174 ++++++++++++++---- .../CloudStorageStreamSession.java | 19 +- .../bulkwriter/DirectStreamSessionTest.java | 2 +- .../bulkwriter/SortedSSTableWriterTest.java | 12 +- 5 files changed, 189 insertions(+), 105 deletions(-) diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java index 7af617090..0d04da823 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java @@ -21,13 +21,12 @@ import java.io.IOException; import java.math.BigInteger; -import java.nio.file.DirectoryStream; import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; @@ -42,8 +41,6 @@ import org.apache.cassandra.bridge.SSTableDescriptor; import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler; import org.apache.cassandra.spark.common.Digest; -import org.apache.cassandra.spark.common.SSTables; -import org.apache.cassandra.spark.data.FileType; import org.apache.cassandra.util.IntWrapper; public class DirectStreamSession extends StreamSession @@ -85,20 +82,19 @@ protected void onSSTablesProduced(Set sstables) // 2. validate the sstables // 3. send the sstables to all replicas // 4. remove the sstables once sent - Map fileDigests = sstableWriter.prepareSStablesToSend(writerContext, sstables); + SortedSSTableWriter.PreparedSSTables preparedSSTables = sstableWriter.prepareSStablesToSend(writerContext, sstables); // retain only the SSTable data components IntWrapper sstableCounter = new IntWrapper(); - fileDigests.keySet() - .stream() - .filter(p -> p.getFileName().toString().endsWith(FileType.DATA.getFileSuffix())) - .forEach(sstable -> { - sstableCounter.value++; - sendSStableToReplicas(sstable); - }); + preparedSSTables.sstables() + .forEach(preparedSSTable -> { + sstableCounter.value++; + sendSStableToReplicas(preparedSSTable); + }); LOGGER.info("[{}]: Sent newly produced SSTables. sstables={}", sessionID, sstableCounter.value); - LOGGER.info("[{}]: Removing temporary files after streaming. files={}", sessionID, fileDigests); - fileDigests.keySet().forEach(path -> { + Set allSSTableFiles = preparedSSTables.files(); + LOGGER.info("[{}]: Removing temporary files after streaming. files={}", sessionID, allSSTableFiles); + allSSTableFiles.forEach(path -> { try { Files.deleteIfExists(path); @@ -153,22 +149,15 @@ protected StreamResult doFinalizeStream() @Override protected void sendRemainingSSTables() { - try (DirectoryStream dataFileStream = Files.newDirectoryStream(sstableWriter.getOutDir(), "*Data.db")) + try { - for (Path dataFile : dataFileStream) - { - if (isFileStreamed(dataFile)) - { - // the file is already streamed or being streamed; skipping it - continue; - } - - sendSStableToReplicas(dataFile); - } + sstableWriter.remainingSSTablesAfterClose() + .sstables() + .forEach(this::sendSStableToReplicas); LOGGER.info("[{}]: Sent SSTables. sstables={}", sessionID, sstableWriter.sstableCount()); } - catch (IOException exception) + catch (Exception exception) { LOGGER.error("[{}]: Unexpected exception while streaming SSTables {}", sessionID, sstableWriter.getOutDir()); @@ -182,24 +171,23 @@ protected void sendRemainingSSTables() } } - private void sendSStableToReplicas(Path dataFile) + private void sendSStableToReplicas(SortedSSTableWriter.PreparedSSTable preparedSSTable) { int ssTableIdx = nextSSTableIdx.getAndIncrement(); LOGGER.info("[{}]: Pushing SSTable {} to replicas {}", - sessionID, dataFile, + sessionID, preparedSSTable.dataFile(), replicas.stream().map(RingInstance::nodeName).collect(Collectors.joining(","))); - replicas.removeIf(replica -> !trySendSSTableToOneReplica(dataFile, ssTableIdx, replica, sstableWriter.fileDigestMap())); + replicas.removeIf(replica -> !trySendSSTableToOneReplica(preparedSSTable, ssTableIdx, replica)); } - private boolean trySendSSTableToOneReplica(Path dataFile, + private boolean trySendSSTableToOneReplica(SortedSSTableWriter.PreparedSSTable preparedSSTable, int ssTableIdx, - RingInstance replica, - Map fileDigests) + RingInstance replica) { try { - sendSSTableToOneReplica(dataFile, ssTableIdx, replica, fileDigests); + sendSSTableToOneReplica(preparedSSTable, ssTableIdx, replica); return true; } catch (Exception exception) @@ -207,32 +195,35 @@ private boolean trySendSSTableToOneReplica(Path dataFile, LOGGER.error("[{}]: Failed to stream range {} to instance {}", sessionID, tokenRange, replica.nodeName(), exception); writerContext.cluster().refreshClusterInfo(); - failureHandler.addFailure(this.tokenRange, replica, exception.getMessage()); - errors.add(new StreamError(this.tokenRange, replica, exception.getMessage())); + // Sometimes error can contain just file name (e.g. when it is missing). + // Let us return 3 latest stacktrace lines for easier troubleshooting. + String stackTrace = Arrays.stream(exception.getStackTrace()) + .limit(3) + .map(StackTraceElement::toString) + .collect(Collectors.joining("\n")); + String errorMessage = exception.getClass().getName() + ": " + exception.getMessage() + + "\n" + String.join("\n", stackTrace); + failureHandler.addFailure(this.tokenRange, replica, errorMessage); + errors.add(new StreamError(this.tokenRange, replica, errorMessage)); clean(replica, sessionID); return false; } } - private void sendSSTableToOneReplica(Path dataFile, + private void sendSSTableToOneReplica(SortedSSTableWriter.PreparedSSTable preparedSSTable, int ssTableIdx, - RingInstance instance, - Map fileHashes) throws IOException + RingInstance instance) throws IOException { - try (DirectoryStream componentFileStream = Files.newDirectoryStream(dataFile.getParent(), - SSTables.getSSTableBaseName(dataFile) + "*")) + for (Path componentFile : preparedSSTable.files()) { - for (Path componentFile : componentFileStream) + // send data component the last + if (preparedSSTable.dataFile().equals(componentFile)) { - // send data component the last - if (componentFile.getFileName().toString().endsWith("Data.db")) - { - continue; - } - sendSSTableComponent(componentFile, ssTableIdx, instance, fileHashes.get(componentFile)); + continue; } - sendSSTableComponent(dataFile, ssTableIdx, instance, fileHashes.get(dataFile)); + sendSSTableComponent(componentFile, ssTableIdx, instance, preparedSSTable.getDigest(componentFile)); } + sendSSTableComponent(preparedSSTable.dataFile(), ssTableIdx, instance, preparedSSTable.getDigest(preparedSSTable.dataFile())); } private void sendSSTableComponent(Path componentFile, diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java index 593208dda..b48f131bb 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java @@ -28,11 +28,14 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.function.Consumer; +import java.util.stream.Collectors; import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; import com.google.common.collect.Range; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -99,6 +102,8 @@ public class SortedSSTableWriter // Fields protected by synchronization - accessed from both RecordWriter thread and executor threads private final Map overallFileDigests = new HashMap<>(); + // holds the list of newly created sstables after SSTableWriter.close() method was called + private PreparedSSTables remainingSSTablesAfterClose = PreparedSSTables.EMPTY; private boolean isClosed = false; private int sstableCount = 0; private long bytesWritten = 0; @@ -209,17 +214,17 @@ public int sstableCount() * * @param writerContext the bulk writer context * @param sstables the set of SSTable descriptors to prepare - * @return a map of file paths to their digests, or an empty map if the writer is already closed + * @return an object containing list of prepared sstables, or {@link PreparedSSTables#EMPTY} if the writer is already closed * @throws IOException if an I/O error occurs */ - public synchronized Map prepareSStablesToSend(@NotNull BulkWriterContext writerContext, Set sstables) throws IOException + public synchronized PreparedSSTables prepareSStablesToSend(@NotNull BulkWriterContext writerContext, Set sstables) throws IOException { // If the writer is already closed, return empty map // The remaining SSTables will be handled by sendRemainingSSTables() if (isClosed) { LOGGER.debug("Writer is already closed, returning empty digest map. Remaining SSTables will be handled by sendRemainingSSTables()"); - return Collections.emptyMap(); + return PreparedSSTables.EMPTY; } // Filter for SSTables that match the requested descriptors AND haven't been hashed yet @@ -227,8 +232,7 @@ public synchronized Map prepareSStablesToSend(@NotNull BulkWriterC SSTableDescriptor baseName = SSTables.getSSTableDescriptor(path); return sstables.contains(baseName) && !overallFileDigests.containsKey(path); }; - Set dataFilePaths = new HashSet<>(); - Map fileDigests = new HashMap<>(); + PreparedSSTables preparedSSTables = new PreparedSSTables(); // FIXME: CQLSSTableWriter may produce incomplete Filter.db file, rebuilding it manually (see CASSANDRA-21423). // rebuild Filter.db files before calculating their digest rebuildFilterComponents(writerContext, sstableFilter); @@ -236,21 +240,23 @@ public synchronized Map prepareSStablesToSend(@NotNull BulkWriterC { for (Path path : stream) { - if (path.getFileName().toString().endsWith("-" + FileType.DATA.getFileSuffix())) + PreparedSSTable preparedSSTable = preparedSSTables.addIfAbsent(path); + + if (isDataFile(path)) { - dataFilePaths.add(path); sstableCount += 1; } Digest digest = digestAlgorithm.calculateFileDigest(path); - fileDigests.put(path, digest); + preparedSSTable.addComponent(path, digest); LOGGER.debug("Calculated digest={} for path={}", digest, path); } } + Map fileDigests = preparedSSTables.digests(); bytesWritten += calculatedTotalSize(fileDigests.keySet()); overallFileDigests.putAll(fileDigests); - validateSSTables(writerContext, getOutDir(), dataFilePaths); - return fileDigests; + validateSSTables(writerContext, getOutDir(), preparedSSTables.dataFiles()); + return preparedSSTables; } /** @@ -289,19 +295,31 @@ public synchronized void close(BulkWriterContext writerContext) throws IOExcepti // FIXME: CQLSSTableWriter may produce incomplete Filter.db file, rebuilding it manually (see CASSANDRA-21423). rebuildFilterComponents(writerContext, unhashedFilter); - try (DirectoryStream dataFileStream = getDataFileStream(unhashedFilter)) + PreparedSSTables prepared = new PreparedSSTables(); + + try (DirectoryStream fileStream = Files.newDirectoryStream(getOutDir(), unhashedFilter)) { - for (Path dataFile : dataFileStream) + for (Path path : fileStream) { // NOTE: We calculate file hashes before re-reading so that we know what we hashed // is what we validated. Then we send these along with the files and the // receiving end re-hashes the files to make sure they still match. - Map newFileDigests = calculateFileDigestMap(dataFile); - overallFileDigests.putAll(newFileDigests); - newlyHashedFiles.addAll(newFileDigests.keySet()); - sstableCount += 1; + Digest digest = digestAlgorithm.calculateFileDigest(path); + LOGGER.debug("Calculated digest={} for path={}", digest, path); + + overallFileDigests.put(path, digest); + newlyHashedFiles.add(path); + + prepared.addIfAbsent(path) + .addComponent(path, digest); + + if (isDataFile(path)) + { + sstableCount += 1; + } } } + remainingSSTablesAfterClose = prepared; // Only calculate size for newly hashed files, not all files in overallFileDigests // (previously hashed files may have been deleted by prepareSStablesToSend) bytesWritten += calculatedTotalSize(newlyHashedFiles); @@ -388,29 +406,10 @@ private LocalDataLayer buildLocalDataLayer(@NotNull BulkWriterContext writerCont private DirectoryStream getDataFileStream(DirectoryStream.Filter filter) throws IOException { // Combine the data file filter with the provided filter - DirectoryStream.Filter combinedFilter = path -> { - String fileName = path.getFileName().toString(); - return fileName.endsWith("Data.db") && filter.accept(path); - }; + DirectoryStream.Filter combinedFilter = path -> isDataFile(path) && filter.accept(path); return Files.newDirectoryStream(getOutDir(), combinedFilter); } - private Map calculateFileDigestMap(Path dataFile) throws IOException - { - Map fileHashes = new HashMap<>(); - try (DirectoryStream filesToHash = - Files.newDirectoryStream(dataFile.getParent(), SSTables.getSSTableBaseName(dataFile) + "*")) - { - for (Path path : filesToHash) - { - Digest digest = digestAlgorithm.calculateFileDigest(path); - fileHashes.put(path, digest); - LOGGER.debug("Calculated digest={} for path={}", digest, path); - } - } - return fileHashes; - } - private long calculatedTotalSize(Collection paths) throws IOException { long totalSize = 0; @@ -438,4 +437,107 @@ public Map fileDigestMap() { return Collections.unmodifiableMap(overallFileDigests); } + + public PreparedSSTables remainingSSTablesAfterClose() + { + return remainingSSTablesAfterClose; + } + + /** + * Helper class representing list of newly generated sstables. + */ + public static class PreparedSSTables + { + private static final PreparedSSTables EMPTY = new PreparedSSTables(); + + private final Map sstables = new HashMap<>(); // indexed by base file name + + public PreparedSSTable addIfAbsent(Path path) + { + String baseName = SSTables.getSSTableDescriptor(path).baseFilename; + return sstables.computeIfAbsent(baseName, (__) -> new PreparedSSTable()); + } + + /** + * @return all files from every generated sstable + */ + public Set files() + { + return sstables.values().stream() + .flatMap(c -> c.components.keySet().stream()) + .collect(Collectors.toSet()); + } + + /** + * @return data files from all sstables + */ + public Set dataFiles() + { + return sstables.values().stream() + .map(c -> c.dataFile) + .collect(Collectors.toSet()); + } + + /** + * @return digests for all files from every generated sstable + */ + public Map digests() + { + return sstables.values().stream() + .map(c -> c.components) + .flatMap(c -> c.entrySet().stream()) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + } + + public Collection sstables() + { + return sstables.values(); + } + } + + public static class PreparedSSTable + { + private Path dataFile; + private final Map components = new HashMap<>(); + + public void addComponent(Path path, Digest digest) + { + if (isDataFile(path)) + { + dataFile = path; + } + components.put(path, digest); + } + + /** + * @return path to data file + */ + public Path dataFile() + { + return dataFile; + } + + /** + * @return list of all sstable files + */ + public List files() + { + return ImmutableList.copyOf(components.keySet()); + } + + /** + * @param path component path + * @return digest value, if it is known for given component + */ + @Nullable + public Digest getDigest(Path path) + { + return components.get(path); + } + } + + private static boolean isDataFile(Path path) + { + return path.getFileName().toString().endsWith("-" + FileType.DATA.getFileSuffix()); + } } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/CloudStorageStreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/CloudStorageStreamSession.java index b749a3597..f3b3a233a 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/CloudStorageStreamSession.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/CloudStorageStreamSession.java @@ -23,23 +23,21 @@ import java.math.BigInteger; import java.nio.file.Path; import java.util.HashSet; -import java.util.Map; import java.util.Set; import java.util.concurrent.ExecutorService; -import java.util.stream.Collectors; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Range; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import o.a.c.sidecar.client.shaded.client.SidecarInstance; import o.a.c.sidecar.client.shaded.common.request.data.CreateSliceRequestPayload; import o.a.c.sidecar.client.shaded.common.response.data.RestoreJobSummaryResponsePayload; import org.apache.cassandra.bridge.CassandraBridge; import org.apache.cassandra.bridge.CassandraBridgeFactory; import org.apache.cassandra.bridge.SSTableDescriptor; import org.apache.cassandra.clients.Sidecar; -import o.a.c.sidecar.client.shaded.client.SidecarInstance; import org.apache.cassandra.spark.bulkwriter.BulkWriteValidator; import org.apache.cassandra.spark.bulkwriter.BulkWriterContext; import org.apache.cassandra.spark.bulkwriter.JobInfo; @@ -51,8 +49,6 @@ import org.apache.cassandra.spark.bulkwriter.TransportContext; import org.apache.cassandra.spark.bulkwriter.cloudstorage.coordinated.CoordinatedCloudStorageDataTransferApi; import org.apache.cassandra.spark.bulkwriter.token.ReplicaAwareFailureHandler; -import org.apache.cassandra.spark.common.Digest; -import org.apache.cassandra.spark.common.SSTables; import org.apache.cassandra.spark.data.QualifiedTableName; import org.apache.cassandra.spark.exception.ConsistencyNotSatisfiedException; import org.apache.cassandra.spark.exception.S3ApiCallException; @@ -119,15 +115,12 @@ protected void onSSTablesProduced(Set sstables) executorService.submit(() -> { try { - Map fileDigests = sstableWriter.prepareSStablesToSend(writerContext, sstables); - sstablesBundler.includeFileDigests(fileDigests); + SortedSSTableWriter.PreparedSSTables preparedSSTables = sstableWriter.prepareSStablesToSend(writerContext, sstables); + sstablesBundler.includeFileDigests(preparedSSTables.digests()); // sstablesBundler keeps track of the known files. No need to record the streamed files. - // group the files by sstable (unique) basename and add to bundler - fileDigests.keySet() - .stream() - .collect(Collectors.groupingBy(SSTables::getSSTableBaseName)) - .values() - .forEach(sstablesBundler::includeSSTable); + // add to bundler files grouped by base sstable name + preparedSSTables.sstables() + .forEach(s -> sstablesBundler.includeSSTable(s.files())); if (!sstablesBundler.hasNext()) { diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java index f6fe8c5e8..de876f674 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSessionTest.java @@ -184,7 +184,7 @@ void testOutDirCreationFailureCleansAllReplicas() fut.get(); }) .isInstanceOf(ExecutionException.class) - .hasRootCauseInstanceOf(NoSuchFileException.class); + .hasMessageContaining(NoSuchFileException.class.getName()); List actualInstances = writerContext.getCleanedInstances().stream() .map(CassandraInstance::nodeName) .collect(Collectors.toList()); diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriterTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriterTest.java index fba73361e..95e19e969 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriterTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriterTest.java @@ -32,7 +32,6 @@ import java.util.Collections; import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.SortedMap; import java.util.TreeMap; @@ -60,7 +59,6 @@ import org.apache.cassandra.bridge.SSTableDescriptor; import org.apache.cassandra.spark.bulkwriter.token.ConsistencyLevel; import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping; -import org.apache.cassandra.spark.common.Digest; import org.apache.cassandra.spark.data.CqlTable; import org.apache.cassandra.spark.data.FileSystemSSTable; import org.apache.cassandra.spark.data.ReplicationFactor; @@ -425,13 +423,13 @@ public void testBytesWrittenWithDeletedFiles(String version) throws Exception writer.addRow(BigInteger.valueOf(100), ImmutableMap.of("id", 2, "date", 2, "course", "test2", "marks", 200)); // Call prepareSStablesToSend with the existing SSTables - Map processedFiles = writer.prepareSStablesToSend(writerContext, new HashSet<>(existingSSTables)); - assertThat(processedFiles).as("Should have processed existing SSTables").isNotEmpty(); + SortedSSTableWriter.PreparedSSTables processedFiles = writer.prepareSStablesToSend(writerContext, new HashSet<>(existingSSTables)); + assertThat(processedFiles.sstables()).as("Should have processed existing SSTables").isNotEmpty(); long bytesAfterPrepare = writer.bytesWritten(); // Delete the files that were processed (simulating DirectStreamSession behavior) - for (Path path : processedFiles.keySet()) + for (Path path : processedFiles.files()) { Files.deleteIfExists(path); } @@ -474,10 +472,10 @@ public void testPrepareSStablesToSendAfterClose(String version) throws Exception int fileDigestCountAfterClose = writer.fileDigestMap().size(); // Try to call prepareSStablesToSend after close - it should return empty map - Map result = writer.prepareSStablesToSend(writerContext, new HashSet<>()); + SortedSSTableWriter.PreparedSSTables result = writer.prepareSStablesToSend(writerContext, new HashSet<>()); // Verify it returned an empty map - assertThat(result) + assertThat(result.sstables()) .as("prepareSStablesToSend should return empty map when called after close") .isEmpty(); From bac241c214ad47e206a2e83353b6f69eab517710 Mon Sep 17 00:00:00 2001 From: Lukasz Antoniak Date: Wed, 2 Sep 2026 09:11:27 +0200 Subject: [PATCH 2/3] Apply review comments --- CHANGES.txt | 1 + .../spark/bulkwriter/DirectStreamSession.java | 6 ++---- .../spark/bulkwriter/SortedSSTableWriter.java | 16 ++++++++++++---- .../spark/bulkwriter/StreamSession.java | 13 ------------- 4 files changed, 15 insertions(+), 21 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index c3cda0a5a..0b412acf3 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * Eliminate redundant filesystem lookups in SSTable direct streaming (CASSANALYTICS-104) * CDC reader stats silently dropped in SidecarCdcBuilder (CASSANALYTICS-191) * Add CapturePublishedSchema metric to SidecarCdcStats (CASSANALYTICS-189) * Expand list of architecture that supports unaligned access in FastByteOperations (CASSANALYTICS-188) diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java index 0d04da823..ab1473b16 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java @@ -83,7 +83,6 @@ protected void onSSTablesProduced(Set sstables) // 3. send the sstables to all replicas // 4. remove the sstables once sent SortedSSTableWriter.PreparedSSTables preparedSSTables = sstableWriter.prepareSStablesToSend(writerContext, sstables); - // retain only the SSTable data components IntWrapper sstableCounter = new IntWrapper(); preparedSSTables.sstables() .forEach(preparedSSTable -> { @@ -202,7 +201,7 @@ private boolean trySendSSTableToOneReplica(SortedSSTableWriter.PreparedSSTable p .map(StackTraceElement::toString) .collect(Collectors.joining("\n")); String errorMessage = exception.getClass().getName() + ": " + exception.getMessage() - + "\n" + String.join("\n", stackTrace); + + "\n" + stackTrace; failureHandler.addFailure(this.tokenRange, replica, errorMessage); errors.add(new StreamError(this.tokenRange, replica, errorMessage)); clean(replica, sessionID); @@ -217,7 +216,7 @@ private void sendSSTableToOneReplica(SortedSSTableWriter.PreparedSSTable prepare for (Path componentFile : preparedSSTable.files()) { // send data component the last - if (preparedSSTable.dataFile().equals(componentFile)) + if (componentFile.equals(preparedSSTable.dataFile())) { continue; } @@ -235,7 +234,6 @@ private void sendSSTableComponent(Path componentFile, LOGGER.info("[{}]: Uploading {} to {}: size={} digest={}", sessionID, componentFile, instance.nodeName(), Files.size(componentFile), digest); directDataTransferApi.uploadSSTableComponent(componentFile, ssTableIdx, instance, this.sessionID, digest); - recordStreamedFile(componentFile); } private List commit(DirectStreamResult streamResult) throws ExecutionException, InterruptedException diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java index b48f131bb..435df3985 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java @@ -240,7 +240,7 @@ public synchronized PreparedSSTables prepareSStablesToSend(@NotNull BulkWriterCo { for (Path path : stream) { - PreparedSSTable preparedSSTable = preparedSSTables.addIfAbsent(path); + PreparedSSTable preparedSSTable = preparedSSTables.getOrPrepareSSTable(path); if (isDataFile(path)) { @@ -310,7 +310,7 @@ public synchronized void close(BulkWriterContext writerContext) throws IOExcepti overallFileDigests.put(path, digest); newlyHashedFiles.add(path); - prepared.addIfAbsent(path) + prepared.getOrPrepareSSTable(path) .addComponent(path, digest); if (isDataFile(path)) @@ -448,11 +448,19 @@ public PreparedSSTables remainingSSTablesAfterClose() */ public static class PreparedSSTables { - private static final PreparedSSTables EMPTY = new PreparedSSTables(); + private static final PreparedSSTables EMPTY = new PreparedSSTables() + { + @Override + public PreparedSSTable getOrPrepareSSTable(Path path) + { + // assert that state is never modified + throw new IllegalStateException(); + } + }; private final Map sstables = new HashMap<>(); // indexed by base file name - public PreparedSSTable addIfAbsent(Path path) + public PreparedSSTable getOrPrepareSSTable(Path path) { String baseName = SSTables.getSSTableDescriptor(path).baseFilename; return sstables.computeIfAbsent(baseName, (__) -> new PreparedSSTable()); diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java index 40cbbdeaa..fb0cc9d2f 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/StreamSession.java @@ -22,14 +22,12 @@ import java.io.File; import java.io.IOException; import java.math.BigInteger; -import java.nio.file.Path; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicReference; @@ -61,7 +59,6 @@ public abstract class StreamSession protected final SortedSSTableWriter sstableWriter; protected final ExecutorService executorService; - private final Set streamedFiles = ConcurrentHashMap.newKeySet(); private final AtomicReference lastStreamFailure = new AtomicReference<>(); private volatile boolean isStreamFinalized = false; @@ -164,16 +161,6 @@ protected boolean setLastStreamFailure(Exception streamFailure) return lastStreamFailure.compareAndSet(null, streamFailure); } - protected void recordStreamedFile(Path file) - { - streamedFiles.add(file); - } - - protected boolean isFileStreamed(Path file) - { - return streamedFiles.contains(file); - } - @VisibleForTesting List getReplicas() { From c71d3503ae78c96075a20f4952a176543f2b0170 Mon Sep 17 00:00:00 2001 From: Lukasz Antoniak Date: Wed, 2 Sep 2026 09:40:20 +0200 Subject: [PATCH 3/3] Apply review comments --- .../spark/bulkwriter/DirectStreamSession.java | 1 + .../spark/bulkwriter/SortedSSTableWriter.java | 10 ++++++++-- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java index ab1473b16..420acd441 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java @@ -222,6 +222,7 @@ private void sendSSTableToOneReplica(SortedSSTableWriter.PreparedSSTable prepare } sendSSTableComponent(componentFile, ssTableIdx, instance, preparedSSTable.getDigest(componentFile)); } + Preconditions.checkNotNull(preparedSSTable.dataFile(), "Data file not present in SSTable: {}", preparedSSTable); sendSSTableComponent(preparedSSTable.dataFile(), ssTableIdx, instance, preparedSSTable.getDigest(preparedSSTable.dataFile())); } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java index 435df3985..b163ca4fc 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java @@ -451,7 +451,7 @@ public static class PreparedSSTables private static final PreparedSSTables EMPTY = new PreparedSSTables() { @Override - public PreparedSSTable getOrPrepareSSTable(Path path) + protected PreparedSSTable getOrPrepareSSTable(Path path) { // assert that state is never modified throw new IllegalStateException(); @@ -460,7 +460,7 @@ public PreparedSSTable getOrPrepareSSTable(Path path) private final Map sstables = new HashMap<>(); // indexed by base file name - public PreparedSSTable getOrPrepareSSTable(Path path) + protected PreparedSSTable getOrPrepareSSTable(Path path) { String baseName = SSTables.getSSTableDescriptor(path).baseFilename; return sstables.computeIfAbsent(baseName, (__) -> new PreparedSSTable()); @@ -542,6 +542,12 @@ public Digest getDigest(Path path) { return components.get(path); } + + @Override + public String toString() + { + return components.keySet().toString(); + } } private static boolean isDataFile(Path path)