arjunashok commented on code in PR #254:
URL:
https://github.com/apache/cassandra-analytics/pull/254#discussion_r4342706583
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/DataLayer.java:
##########
@@ -267,13 +274,24 @@ public boolean useIncrementalRepair()
return true;
}
+ public boolean saiFilteringEnabled()
+ {
+ return ClientConfig.DEFAULT_SAI_FILTERING_ENABLED;
+ }
+
+ public int saiMaxCandidateTokens()
+ {
+ return ClientConfig.DEFAULT_SAI_MAX_CANDIDATE_TOKENS;
+ }
+
/**
* @return CompactionScanner for iterating over one or more SSTables,
compacting data and purging tombstones
*/
public StreamScanner<RowData> openCompactionScanner(int partitionId,
List<PartitionKeyFilter> partitionKeyFilters,
SSTableTimeRangeFilter
sstableTimeRangeFilter,
- @Nullable
PruneColumnFilter columnFilter)
+ @Nullable
PruneColumnFilter columnFilter,
+ @NotNull
List<SaiFilter> saiFilters)
Review Comment:
We're changing the public signature here, potentially impacting subclasses
overriding it. Consider either keeping the 4-arg method for compatibility or
document this change in CHANGES.txt
##########
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);
Review Comment:
Could we use `SSTableCache` to prevent the network read ? If so, we might
not want to close the summary (done in `getSSTableSummary()` as it gets freed)
##########
cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/spark/reader/sai/SaiIndexReader.java:
##########
@@ -0,0 +1,596 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cassandra.spark.reader.sai;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.db.ClusteringComparator;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.PartitionRangeReadCommand;
+import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.dht.Bounds;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.index.sai.QueryContext;
+import org.apache.cassandra.index.sai.StorageAttachedIndex;
+import org.apache.cassandra.index.sai.disk.PrimaryKeyMap;
+import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
+import org.apache.cassandra.index.sai.disk.format.SSTableIndexFileAccess;
+import org.apache.cassandra.index.sai.disk.v1.MetadataSource;
+import org.apache.cassandra.index.sai.disk.v1.PerColumnIndexFiles;
+import org.apache.cassandra.index.sai.disk.v1.segment.IndexSegmentSearcher;
+import org.apache.cassandra.index.sai.disk.v1.segment.SegmentMetadata;
+import org.apache.cassandra.index.sai.iterators.KeyRangeIntersectionIterator;
+import org.apache.cassandra.index.sai.iterators.KeyRangeIterator;
+import org.apache.cassandra.index.sai.iterators.KeyRangeUnionIterator;
+import org.apache.cassandra.index.sai.plan.SaiPlanAccessor;
+import org.apache.cassandra.index.sai.plan.Expression;
+import org.apache.cassandra.index.sai.plan.QueryController;
+import org.apache.cassandra.index.sai.plan.SaiQueryPlanner;
+import org.apache.cassandra.index.sai.utils.PrimaryKey;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.spark.data.SSTable;
+import org.apache.cassandra.spark.sparksql.filters.SaiFilter;
+import org.apache.cassandra.spark.sparksql.filters.SparkRangeFilter;
+import org.apache.cassandra.spark.utils.Preconditions;
+import org.apache.cassandra.utils.TokenUtils;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Opens Cassandra 5 SAI components, executes the query plan once, and
materializes the resulting candidate tokens
+ * as an exact compact token set.
+ *
+ * <p>Cassandra's native SAI planner analyzes the conjunctive query, including
same-column range folding and
+ * intersection. For every planned expression, matches are first UNIONed
across all SSTables before Cassandra
+ * intersects different expressions. This ordering is important for
correctness: values contributing to a logically
+ * reconciled row may live in different SSTables. Applying conjunctions inside
each SSTable could therefore create
+ * false negatives.</p>
+ *
+ * <p>SAI is a read-planning phase only. The native iterator is fully consumed
before {@code Data.db} scanning starts,
+ * then all SAI resources are closed. The resulting exact tokens are shared by
every participating SSTable so the normal
+ * Cassandra compaction / reconciliation path can still apply newer values,
TTLs and tombstones. Spark keeps the
+ * original predicates as residual filters, so stale index entries can only
create false positives.</p>
+ */
+public final class SaiIndexReader
+{
+ private static final Logger LOGGER =
LoggerFactory.getLogger(SaiIndexReader.class);
+
+ private SaiIndexReader()
+ {
+ throw new IllegalStateException("Static utility class");
+ }
+
+ /**
+ * Executes all SAI predicates once and returns the sorted candidate
tokens as an exact compact token set.
+ *
+ * @return empty {@link Optional} when SAI cannot safely be used and the
caller should fall back to the ordinary SSTable
+ * scan; an empty {@link CandidateTokens} means SAI executed
successfully and found no candidates
+ */
+ @NotNull
+ public static Optional<CandidateTokens> findCandidateTokens(@NotNull
TableMetadata metadata,
+ @NotNull
Set<SSTable> sstables,
+ @NotNull
List<SaiFilter> filters,
+ @Nullable
SparkRangeFilter sparkRangeFilter,
+ int
maxCandidateTokens)
+ {
+ if (sstables.isEmpty() || filters.isEmpty())
+ {
+ return Optional.of(CandidateTokens.empty());
+ }
+
+ ResourceGroup resources = new ResourceGroup();
+ try
+ {
+ SaiQueryPlanner.Plan plan = SaiQueryPlanner.plan(metadata,
filters);
+ if (plan == null)
+ {
+ return Optional.empty();
+ }
+
+ List<StorageAttachedIndex> resourceIndexes =
plan.resourceIndexes();
+
+ List<SSTableResources> sstableResources = new
ArrayList<>(sstables.size());
+ for (SSTable sstable : sstables)
+ {
+ SSTableResources opened = SSTableResources.open(metadata,
sstable, resourceIndexes);
+ resources.add(opened);
+ sstableResources.add(opened);
+ }
+
+ QueryContext queryContext = new QueryContext(null, Long.MAX_VALUE);
+ try (KeyRangeIterator finalMatches = executePlan(metadata,
+ plan,
+ sstableResources,
+ queryContext,
+ sparkRangeFilter))
+ {
+ finalMatches.setOnClose(resources::close);
Review Comment:
Worth evaluating if implementations of `KeyRangeIterator` honor the
`close()` operation as expected here (eg. `EmptyRangeIterator `,
`PostingListRangeIterator `). We might want to explicitly close the resources
in a `finally` in the method that opened them instead.
##########
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:
The readers are opened sequentially in the task thread compared to
concurrently in trunk where `ssTables.openAll()` submits one task per SSTable
to an executor.
--
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]
