lukasz-antoniak commented on code in PR #254:
URL:
https://github.com/apache/cassandra-analytics/pull/254#discussion_r4351328349
##########
cassandra-six-zero-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridgeImplementation.java:
##########
@@ -227,6 +232,186 @@ public StreamScanner<RowData>
getCompactionScanner(@NotNull CqlTable table,
}));
}
+ @Override
+ public StreamScanner<RowData> getCompactionScanner(@NotNull CqlTable table,
+ @NotNull Partitioner
partitioner,
+ @NotNull
SSTablesSupplier ssTables,
+ @Nullable
SparkRangeFilter sparkRangeFilter,
+ @NotNull
Collection<PartitionKeyFilter> partitionKeyFilters,
+ @NotNull
SSTableTimeRangeFilter sstableTimeRangeFilter,
+ @Nullable
PruneColumnFilter columnFilter,
+ @NotNull TimeProvider
timeProvider,
+ boolean readIndexOffset,
+ boolean
useIncrementalRepair,
+ @NotNull Stats stats,
+ @NotNull
List<SaiFilter> saiFilters,
+ int
saiMaxCandidateTokens)
+ {
+ // A complete partition-key lookup is already more selective and
cheaper than consulting SAI.
+ if (!partitionKeyFilters.isEmpty() || saiFilters.isEmpty())
+ {
+ return getCompactionScanner(table, partitioner, ssTables,
sparkRangeFilter, partitionKeyFilters,
+ sstableTimeRangeFilter, columnFilter,
timeProvider, readIndexOffset,
+ useIncrementalRepair, stats);
+ }
+
+ // NOTE: Need to use SchemaBuilder to init keyspace if not already set
in Cassandra Schema instance
+ SchemaBuilder schemaBuilder = new SchemaBuilder(table, partitioner);
+ TableMetadata metadata = schemaBuilder.tableMetaData();
+
+ // Resolve each sstable token span before opening any SAI resources so
data files outside this
+ // Spark worker's token range are ignored.
+ Set<SaiSSTableReference> references = ssTables.openAll((sstable,
isRepairPrimary) -> {
+ if (sparkRangeFilter == null)
+ {
+ return new SaiSSTableReference(sstable, isRepairPrimary);
+ }
+ Pair<BigInteger, BigInteger> tokens = null;
+ if (sstable.isBigFormat())
+ {
+ SSTableSummary summary =
getSSTableSummary(metadata.partitioner,
+ sstable,
+
metadata.params.minIndexInterval,
+
metadata.params.maxIndexInterval);
+ tokens = Pair.of(summary.firstToken, summary.lastToken);
+ }
+ else
+ {
+ Pair<DecoratedKey, DecoratedKey> keys =
ReaderUtils.keysFromIndex(metadata.partitioner, sstable);
+ if (keys != null)
+ {
+ tokens =
Pair.of(ReaderUtils.tokenToBigInteger(keys.left.getToken()),
+
ReaderUtils.tokenToBigInteger(keys.right.getToken()));
+ }
+ }
+ if (tokens == null)
+ {
+ return new SaiSSTableReference(sstable, isRepairPrimary);
+ }
+ return new SaiSSTableReference(sstable,
+ isRepairPrimary,
+ tokens.left,
+ tokens.right,
+ sparkRangeFilter);
+ });
+ if (references.isEmpty())
+ {
+ return EmptyStreamScanner.INSTANCE;
+ }
+
+ Set<SSTable> sstables = references.stream()
+ .filter(p -> !p.ignore)
+ .map(reference -> reference.sstable)
+ .collect(Collectors.toSet());
+ Optional<CandidateTokens> candidates =
SaiIndexReader.findCandidateTokens(metadata,
+
sstables,
+
saiFilters,
+
sparkRangeFilter,
+
saiMaxCandidateTokens);
+ if (candidates.isEmpty())
+ {
+ // No candidate token ranges selected form SAI filter, use
standard full-table scan.
+ return openCompactionScanner(metadata, partitioner, timeProvider,
references, sparkRangeFilter,
+ Collections.emptyList(), null,
sstableTimeRangeFilter, columnFilter,
+ readIndexOffset,
useIncrementalRepair, stats);
+ }
+
+ CandidateTokens candidateTokens = candidates.get();
+ if (candidateTokens.isEmpty())
+ {
+ return EmptyStreamScanner.INSTANCE;
+ }
+
+ return openCompactionScanner(metadata, partitioner, timeProvider,
references, sparkRangeFilter,
+ Collections.emptyList(), candidateTokens,
sstableTimeRangeFilter,
+ columnFilter, readIndexOffset,
useIncrementalRepair, stats);
+ }
+
+ @NotNull
+ private static StreamScanner<RowData> openCompactionScanner(@NotNull
TableMetadata metadata,
+ @NotNull
Partitioner partitioner,
+ @NotNull
TimeProvider timeProvider,
+ @NotNull
Set<SaiSSTableReference> references,
+ @Nullable
SparkRangeFilter sparkRangeFilter,
+ @NotNull
Collection<PartitionKeyFilter> partitionKeyFilters,
+ @Nullable
CandidateTokens candidateTokens,
+ @NotNull
SSTableTimeRangeFilter sstableTimeRangeFilter,
+ @Nullable
PruneColumnFilter columnFilter,
+ boolean
readIndexOffset,
+ boolean
useIncrementalRepair,
+ @NotNull Stats
stats)
+ {
+ Set<org.apache.cassandra.spark.reader.SSTableReader> readers =
references.stream()
+ .map(reference -> {
+ try
+ {
+ return
org.apache.cassandra.spark.reader.SSTableReader.builder(metadata,
reference.sstable)
Review Comment:
This is a very good finding. SAI opens SSTables is a two-step process:
1. First we list SSTables and process their SAI indexes to compute candidate
token ranges.
2. Second step involves opening actual SSTables and running compaction.
For SAI-search correctness, both stages have to use same SSTables. I have
tried to refactor the code and introduce "planning" and "opening" stages in
`SSTablesSupplier`, but code was quite complicated. In the end, I settled on
passing executor service (like we already do in `getPartitionSizeIterator()`)
and use it to build `SSTableReader` in parallel.
I will push my changes tomorrow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
