[core] Prune primary-key scans and vector search with sorted indexes

paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java

@@ -33,8 +33,8 @@
 import org.apache.paimon.predicate.RowIdPredicateVisitor;
 import org.apache.paimon.predicate.TopN;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.AppendBatchTableScan;
 import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.table.source.DataTableBatchScan;
 import org.apache.paimon.table.source.DataTableScan;
 import org.apache.paimon.table.source.InnerTableScan;
 import org.apache.paimon.table.source.Split;
@@ -65,13 +65,13 @@ public class DataEvolutionBatchScan implements DataTableScan {
  private static final Logger LOG = LoggerFactory.getLogger(DataEvolutionBatchScan.class);
 
  private final FileStoreTable table;
- private final DataTableBatchScan batchScan;
+ private final AppendBatchTableScan batchScan;
 
  private Predicate filter;
  private RowRangeIndex pushedRowRangeIndex;
  private GlobalIndexResult globalIndexResult;
 
- public DataEvolutionBatchScan(FileStoreTable wrapped, DataTableBatchScan batchScan) {
+ public DataEvolutionBatchScan(FileStoreTable wrapped, AppendBatchTableScan batchScan) {
  this.table = wrapped;
  this.batchScan = batchScan;
  }

paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourceMeta.java

@@ -89,7 +89,16 @@ public static PrimaryKeyIndexSourceMeta deserialize(byte[] bytes) {
  checkArgument(version == VERSION, "Unsupported index source version: %s.", version);
  int sourceFileCount = input.readInt();
  checkArgument(sourceFileCount > 0, "An index must reference source files.");
- List<PrimaryKeyIndexSourceFile> sourceFiles = new ArrayList<>(sourceFileCount);
+ // Each entry needs at least the two-byte writeUTF length and one long.
+ int maximumSourceFileCount = input.available() / (Short.BYTES + Long.BYTES);
+ checkArgument(
+ sourceFileCount <= maximumSourceFileCount,
+ "Failed to deserialize index source metadata: source file count %s "
+ + "exceeds the maximum %s allowed by the remaining bytes.",
+ sourceFileCount,
+ maximumSourceFileCount);
+ List<PrimaryKeyIndexSourceFile> sourceFiles =
+ new ArrayList<>(Math.min(sourceFileCount, 1024));
  for (int i = 0; i < sourceFileCount; i++) {
  sourceFiles.add(new PrimaryKeyIndexSourceFile(input.readUTF(), input.readLong()));
  }

paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java

@@ -128,6 +128,26 @@ public List<PkVectorSearchResult> search(
  Map<String, DeletionVector> deletionVectors,
  Set<String> activeSourceFiles,
  Map<String, String> searchOptions) {
+ return search(
+ segment,
+ sourceMeta,
+ query,
+ limit,
+ deletionVectors,
+ activeSourceFiles,
+ Collections.emptyMap(),
+ searchOptions);
+ }
+
+ public List<PkVectorSearchResult> search(
+ IndexFileMeta segment,
+ PrimaryKeyIndexSourceMeta sourceMeta,
+ float[] query,
+ int limit,
+ Map<String, DeletionVector> deletionVectors,
+ Set<String> activeSourceFiles,
+ Map<String, List<Range>> rowRangesByFile,
+ Map<String, String> searchOptions) {
  checkArgument(limit > 0, "Vector search limit must be positive: %s.", limit);
  GlobalIndexMeta globalIndexMeta = segment.globalIndexMeta();
  checkArgument(
@@ -161,7 +181,11 @@ public List<PkVectorSearchResult> search(
  try {
  VectorSearch search = new VectorSearch(query, limit, vectorField.name(), searchOptions);
  RoaringNavigableMap64 liveRows =
- liveRowPositions(sourceMeta.sourceFiles(), activeSourceFiles, deletionVectors);
+ liveRowPositions(
+ sourceMeta.sourceFiles(),
+ activeSourceFiles,
+ deletionVectors,
+ rowRangesByFile);
  if (liveRows != null) {
  search.withIncludeRowIds(liveRows);
  }
@@ -193,6 +217,11 @@ public List<PkVectorSearchResult> search(
  "ANN segment %s returned snapshot-deleted row position %s.",
  segment.fileName(),
  filePosition.rowPosition);
+ List<Range> rowRanges = rowRangesByFile.get(filePosition.dataFileName);
+ checkArgument(
+ rowRanges == null || contains(rowRanges, filePosition.rowPosition),
+ "ANN segment %s returned a row outside the pre-filter.",
+ segment.fileName());
  candidates.add(
  new PkVectorSearchResult(
  filePosition.dataFileName,
@@ -211,15 +240,16 @@ public List<PkVectorSearchResult> search(
  private static RoaringNavigableMap64 liveRowPositions(
  List<PrimaryKeyIndexSourceFile> sourceFiles,
  Set<String> activeSourceFiles,
- Map<String, DeletionVector> deletionVectors) {
+ Map<String, DeletionVector> deletionVectors,
+ Map<String, List<Range>> rowRangesByFile) {
  boolean allSourcesActive = true;
  for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
  if (!activeSourceFiles.contains(sourceFile.fileName())) {
  allSourcesActive = false;
  break;
  }
  }
- if (allSourcesActive && deletionVectors.isEmpty()) {
+ if (allSourcesActive && deletionVectors.isEmpty() && rowRangesByFile.isEmpty()) {
  return null;
  }
  RoaringNavigableMap64 live = new RoaringNavigableMap64();
@@ -228,7 +258,18 @@ private static RoaringNavigableMap64 liveRowPositions(
  for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
  boolean active = activeSourceFiles.contains(sourceFile.fileName());
  if (active && sourceFile.rowCount() > 0) {
- live.addRange(new Range(fileOffset, fileOffset + sourceFile.rowCount() - 1));
+ List<Range> rowRanges = rowRangesByFile.get(sourceFile.fileName());
+ if (rowRanges == null) {
+ live.addRange(new Range(fileOffset, fileOffset + sourceFile.rowCount() - 1));
+ } else {
+ for (Range range : rowRanges) {
+ checkArgument(
+ range.from >= 0 && range.to < sourceFile.rowCount(),
+ "Pre-filter range is outside source file %s.",
+ sourceFile.fileName());
+ live.addRange(range.addOffset(fileOffset));
+ }
+ }
  }
  DeletionVector deletionVector =
  active ? deletionVectors.get(sourceFile.fileName()) : null;
@@ -242,6 +283,23 @@ private static RoaringNavigableMap64 liveRowPositions(
  return live;
  }
 
+ private static boolean contains(List<Range> ranges, long position) {
+ int low = 0;
+ int high = ranges.size() - 1;
+ while (low <= high) {
+ int middle = (low + high) >>> 1;
+ Range range = ranges.get(middle);
+ if (position < range.from) {
+ high = middle - 1;
+ } else if (position > range.to) {
+ low = middle + 1;
+ } else {
+ return true;
+ }
+ }
+ return false;
+ }
+
  private static long totalRowCount(List<PrimaryKeyIndexSourceFile> sourceFiles) {
  long total = 0;
  for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {

paimon-core/src/main/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorBucketSearch.java

@@ -24,6 +24,7 @@
 import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
 import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
 import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.utils.Range;
 
 import javax.annotation.Nullable;
 
@@ -95,6 +96,25 @@ public Result search(
  int indexedLimit,
  int exactLimit)
  throws IOException {
+ return search(
+ state,
+ activeFiles,
+ deletionVectors,
+ Collections.emptyMap(),
+ query,
+ indexedLimit,
+ exactLimit);
+ }
+
+ public Result search(
+ PkVectorBucketIndexState state,
+ List<DataFileMeta> activeFiles,
+ Map<String, DeletionVector> deletionVectors,
+ Map<String, List<Range>> rowRangesByFile,
+ float[] query,
+ int indexedLimit,
+ int exactLimit)
+ throws IOException {
  checkArgument(indexedLimit > 0, "Vector indexed search limit must be positive.");
  checkArgument(exactLimit > 0, "Vector exact search limit must be positive.");
  Map<String, DataFileMeta> filesByName = new HashMap<>();
@@ -121,15 +141,26 @@ public Result search(
  covered.add(source.fileName());
  }
  checkArgument(annSearcher != null, "ANN search is not configured.");
- for (PkVectorSearchResult result :
- annSearcher.search(
- ann,
- sourceMeta,
- query,
- indexedLimit,
- deletionVectors,
- activeSourceFiles,
- searchOptions)) {
+ List<PkVectorSearchResult> annResults =
+ rowRangesByFile.isEmpty()
+ ? annSearcher.search(
+ ann,
+ sourceMeta,
+ query,
+ indexedLimit,
+ deletionVectors,
+ activeSourceFiles,
+ searchOptions)
+ : annSearcher.search(
+ ann,
+ sourceMeta,
+ query,
+ indexedLimit,
+ deletionVectors,
+ activeSourceFiles,
+ rowRangesByFile,
+ searchOptions);
+ for (PkVectorSearchResult result : annResults) {
  add(indexedNearest, result, indexedLimit);
  }
  }
@@ -140,7 +171,14 @@ public Result search(
  continue;
  }
  DeletionVector dv = deletionVectors.get(file.fileName());
- LongPredicate excluded = dv == null ? position -> false : dv::isDeleted;
+ List<Range> rowRanges = rowRangesByFile.get(file.fileName());
+ if (rowRanges != null && rowRanges.isEmpty()) {
+ continue;
+ }
+ LongPredicate excluded =
+ position ->
+ (dv != null && dv.isDeleted(position))
+ || (rowRanges != null && !contains(rowRanges, position));
  try (PkVectorReader reader = vectorReaderFactory.create(file)) {
  for (PkVectorSearchResult result :
  PkVectorExactSearcher.search(
@@ -153,6 +191,23 @@ public Result search(
  return new Result(sorted(indexedNearest), sorted(exactNearest));
  }
 
+ private static boolean contains(List<Range> ranges, long position) {
+ int low = 0;
+ int high = ranges.size() - 1;
+ while (low <= high) {
+ int middle = (low + high) >>> 1;
+ Range range = ranges.get(middle);
+ if (position < range.from) {
+ high = middle - 1;
+ } else if (position > range.to) {
+ low = middle + 1;
+ } else {
+ return true;
+ }
+ }
+ return false;
+ }
+
  private static List<PkVectorSearchResult> sorted(PriorityQueue<PkVectorSearchResult> nearest) {
  List<PkVectorSearchResult> result = new ArrayList<>(nearest);
  Collections.sort(result, BEST_FIRST);

paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java

@@ -228,6 +228,12 @@ public DataTableScan newScan() {
  return wrapped.newScan();
  }
 
+ @Override
+ public DataTableScan newScan(SnapshotReaderFactory snapshotReaderFactory) {
+ privilegeChecker.assertCanSelect(identifier);
+ return wrapped.newScan(snapshotReaderFactory);
+ }
+
  @Override
  public StreamDataTableScan newStreamScan() {
  privilegeChecker.assertCanSelect(identifier);

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -24,7 +24,6 @@
 import org.apache.paimon.consumer.ConsumerManager;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
-import org.apache.paimon.globalindex.DataEvolutionBatchScan;
 import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.ManifestFileMeta;
@@ -45,8 +44,6 @@
 import org.apache.paimon.table.sink.RowKindGenerator;
 import org.apache.paimon.table.sink.TableCommitImpl;
 import org.apache.paimon.table.sink.WriteSelector;
-import org.apache.paimon.table.source.DataTableBatchScan;
-import org.apache.paimon.table.source.DataTableScan;
 import org.apache.paimon.table.source.DataTableStreamScan;
 import org.apache.paimon.table.source.SplitGenerator;
 import org.apache.paimon.table.source.StreamDataTableScan;
@@ -287,26 +284,6 @@ public SnapshotReader newSnapshotReader() {
  dvmetaCache);
  }
 
- @Override
- public DataTableScan newScan() {
- DataTableBatchScan scan =
- new DataTableBatchScan(
- tableSchema,
- schemaManager(),
- coreOptions(),
- newSnapshotReader(),
- catalogEnvironment.tableQueryAuth(coreOptions()));
- Integer scanBucket = coreOptions().scanBucket();
- if (scanBucket != null) {
- DataTableBatchScan.validateScanBucketOption(tableSchema, coreOptions(), scanBucket);
- scan.withBucket(scanBucket);
- }
- if (coreOptions().dataEvolutionEnabled()) {
- return new DataEvolutionBatchScan(this, scan);
- }
- return scan;
- }
-
  @Override
  public StreamDataTableScan newStreamScan() {
  DataTableStreamScan scan =
@@ -321,7 +298,6 @@ public StreamDataTableScan newStreamScan() {
  !tableSchema.primaryKeys().isEmpty());
  Integer scanBucket = coreOptions().scanBucket();
  if (scanBucket != null) {
- DataTableBatchScan.validateScanBucketOption(tableSchema, coreOptions(), scanBucket);
  scan.withBucket(scanBucket);
  }
  return scan;

paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java

@@ -23,17 +23,20 @@
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.globalindex.DataEvolutionBatchScan;
 import org.apache.paimon.operation.AppendOnlyFileStoreScan;
 import org.apache.paimon.operation.BaseAppendFileStoreWrite;
 import org.apache.paimon.operation.FileStoreScan;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.query.LocalTableQuery;
 import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.table.source.AppendBatchTableScan;
 import org.apache.paimon.table.source.AppendOnlySplitGenerator;
 import org.apache.paimon.table.source.AppendTableRead;
 import org.apache.paimon.table.source.DataEvolutionSplitGenerator;
 import org.apache.paimon.table.source.DataEvolutionTableRead;
+import org.apache.paimon.table.source.DataTableScan;
 import org.apache.paimon.table.source.InnerTableRead;
 import org.apache.paimon.table.source.SplitGenerator;
 import org.apache.paimon.table.source.splitread.AppendTableRawFileSplitReadProvider;
@@ -133,6 +136,24 @@ public InnerTableRead newRead() {
  : new AppendTableRead(providerFactories, schema());
  }
 
+ @Override
+ public DataTableScan newScan() {
+ return newScan(FileStoreTable::newSnapshotReader);
+ }
+
+ @Override
+ public DataTableScan newScan(SnapshotReaderFactory snapshotReaderFactory) {
+ CoreOptions options = coreOptions();
+ AppendBatchTableScan scan =
+ new AppendBatchTableScan(
+ schema(),
+ schemaManager(),
+ options,
+ snapshotReaderFactory.create(this),
+ catalogEnvironment.tableQueryAuth(options));
+ return options.dataEvolutionEnabled() ? new DataEvolutionBatchScan(this, scan) : scan;
+ }
+
  @Override
  public TableWriteImpl<InternalRow> newWrite(String commitUser) {
  return newWrite(commitUser, null);