Skip to content

CASSANALYTICS-104: Eliminate redundant filesystem lookups in SSTable streaming #239

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lukasz-antoniak wants to merge 1 commit into
base: trunk
Choose a base branch
from
Open

CASSANALYTICS-104: Eliminate redundant filesystem lookups in SSTable streaming #239

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 39 additions & 48 deletions ...alytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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<TransportContext.DirectDataBulkWriterContext>
Expand Down Expand Up @@ -85,20 +82,19 @@ protected void onSSTablesProduced(Set<SSTableDescriptor> sstables)
// 2. validate the sstables
// 3. send the sstables to all replicas
// 4. remove the sstables once sent
Map<Path, Digest> fileDigests = sstableWriter.prepareSStablesToSend(writerContext, sstables);
SortedSSTableWriter.PreparedSSTables preparedSSTables = sstableWriter.prepareSStablesToSend(writerContext, sstables);
// retain only the SSTable data components

@yifan-c yifan-c Sep 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please remove this comment, it is stale with the removal of .filter

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<Path> allSSTableFiles = preparedSSTables.files();
LOGGER.info("[{}]: Removing temporary files after streaming. files={}", sessionID, allSSTableFiles);
allSSTableFiles.forEach(path -> {
try
{
Files.deleteIfExists(path);
Expand Down Expand Up @@ -153,22 +149,15 @@ protected StreamResult doFinalizeStream()
@Override
protected void sendRemainingSSTables()
{
try (DirectoryStream<Path> dataFileStream = Files.newDirectoryStream(sstableWriter.getOutDir(), "*Data.db"))
try
{
for (Path dataFile : dataFileStream)
{
if (isFileStreamed(dataFile))

@yifan-c yifan-c Sep 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

StreamSession.isFileStreamed is now dead. Can you remove the code in the base class?

{
// 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());
Expand All @@ -182,57 +171,59 @@ 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<Path, Digest> fileDigests)
RingInstance replica)
{
try
{
sendSSTableToOneReplica(dataFile, ssTableIdx, replica, fileDigests);
sendSSTableToOneReplica(preparedSSTable, ssTableIdx, replica);
return true;
}
catch (Exception exception)
{
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);

@yifan-c yifan-c Sep 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

String.join("\n", stackTrace) is equivalent to just stackTrace, right?
The stacktrace are already aggregated in the statement above.

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<Path, Digest> fileHashes) throws IOException
RingInstance instance) throws IOException
{
try (DirectoryStream<Path> 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))

@yifan-c yifan-c Sep 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can dataFile() return null?

{
// 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,
Expand Down
Loading
Loading