leaves12138 commented on code in PR #8652: URL: https://github.com/apache/paimon/pull/8652#discussion_r3585840332
########## paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java: ########## @@ -0,0 +1,348 @@ +/* + * 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.paimon.table.source; + +import org.apache.paimon.CoreOptions.GlobalIndexSearchMode; +import org.apache.paimon.KeyValueFileStore; +import org.apache.paimon.deletionvectors.DeletionVector; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.globalindex.GlobalIndexReadThreadPool; +import org.apache.paimon.globalindex.GlobalIndexReader; +import org.apache.paimon.globalindex.GlobalIndexer; +import org.apache.paimon.globalindex.io.GlobalIndexFileReader; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileHandler; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.index.pk.PrimaryKeyIndexDefinition; +import org.apache.paimon.index.pkfulltext.PkFullTextDataFileReader; +import org.apache.paimon.index.pkfulltext.PkFullTextIndexBuilder; +import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile; +import org.apache.paimon.index.pkfulltext.PrimaryKeyFullTextBucketSearch; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.KeyValueFileReaderFactory; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.types.DataField; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; + +import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM; +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkNotNull; + +/** Reads primary-key full-text payloads and optional temporary fallback indexes. */ +public class PrimaryKeyFullTextRead implements FullTextRead { + + private final GlobalIndexSearchMode searchMode; + private final int limit; + private final BucketRankingSearch indexedSearch; + private final BucketRankingSearch rawSearch; + + public PrimaryKeyFullTextRead( + FileStoreTable table, + PrimaryKeyIndexDefinition definition, + DataField textField, + String query, + int limit) { + checkArgument(limit > 0, "Full-text search limit must be positive: %s.", limit); + checkArgument( + definition.family() == PrimaryKeyIndexDefinition.Family.FULL_TEXT, + "Primary-key full-text read requires a full-text definition."); + checkArgument( + definition.fieldId() == textField.id(), + "Full-text definition does not match field %s.", + textField.name()); + ProductionSearch production = + new ProductionSearch(table, definition, textField, query, limit); + this.searchMode = table.coreOptions().globalIndexSearchMode(); + this.limit = limit; + this.indexedSearch = production::searchIndexed; + this.rawSearch = production::searchRaw; + } + + PrimaryKeyFullTextRead( + GlobalIndexSearchMode searchMode, + int limit, + BucketRankingSearch indexedSearch, + BucketRankingSearch rawSearch) { + checkArgument(limit > 0, "Full-text search limit must be positive: %s.", limit); + this.searchMode = searchMode; + this.limit = limit; + this.indexedSearch = indexedSearch; + this.rawSearch = rawSearch; + } + + @Override + public PrimaryKeyScoredResult read(FullTextScan.Plan plan) { + checkArgument( + plan instanceof PrimaryKeyFullTextScan.Plan, + "Primary-key full-text read requires a PrimaryKeyFullTextScan plan."); + PrimaryKeyFullTextScan.Plan primaryKeyPlan = (PrimaryKeyFullTextScan.Plan) plan; + return read(primaryKeyPlan.snapshotId(), primaryKeyPlan.splits()); + } + + @Override + public PrimaryKeyScoredResult read(List<FullTextSearchSplit> splits) { + if (splits.isEmpty()) { + return new PrimaryKeyScoredResult(0, Collections.emptyList(), Collections.emptyList()); + } + checkArgument( + splits.get(0) instanceof PrimaryKeyFullTextSearchSplit, + "Primary-key full-text read requires primary-key full-text splits."); + long snapshotId = ((PrimaryKeyFullTextSearchSplit) splits.get(0)).dataSplit().snapshotId(); + return read(snapshotId, splits); + } + + private PrimaryKeyScoredResult read(long snapshotId, List<FullTextSearchSplit> splits) { + List<DataSplit> sourceSplits = new ArrayList<>(splits.size()); + List<List<PrimaryKeySearchPosition>> rankings = new ArrayList<>(); + for (FullTextSearchSplit searchSplit : splits) { + checkArgument( + searchSplit instanceof PrimaryKeyFullTextSearchSplit, + "Primary-key full-text read received an incompatible split."); + PrimaryKeyFullTextSearchSplit split = (PrimaryKeyFullTextSearchSplit) searchSplit; + checkArgument( + split.dataSplit().snapshotId() == snapshotId, + "Full-text bucket split snapshot does not match its plan."); + sourceSplits.add(split.dataSplit()); + rankings.addAll(indexedSearch.search(split)); + if (searchMode != GlobalIndexSearchMode.FAST && !split.uncoveredDataFiles().isEmpty()) { + rankings.addAll(rawSearch.search(split)); + } + } + List<PrimaryKeySearchPosition> positions = + rankings.isEmpty() + ? Collections.emptyList() + : PrimaryKeySearchRanker.rrf(rankings, limit); Review Comment: This treats every physical payload and fallback file as an independent RRF route, so the original full-text relevance scores are discarded. Because physical rows are disjoint across payloads, the first hit of every archive ties at `1 / 61`; for example, scores `[100, 99]` in archive A lose A's second hit to score `0.1` at rank 1 in archive B. Results also change when index compaction combines payloads even if table data is unchanged. This contradicts the stated merge-by-score behavior and the existing full-text path. Please globally select top-K by relevance score with a deterministic tie-break instead of applying RRF across storage shards. ########## paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextScan.java: ########## @@ -0,0 +1,303 @@ +/* + * 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.paimon.table.source; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.globalindex.IndexedSplit; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileHandler; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.index.pk.PrimaryKeyIndexDefinition; +import org.apache.paimon.index.pk.PrimaryKeyIndexSourcePolicy; +import org.apache.paimon.index.pkfulltext.PkFullTextBucketIndexState; +import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.IndexManifestEntry; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.source.snapshot.SnapshotReader; +import org.apache.paimon.utils.Pair; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Plans compaction-visible full-text inputs from one captured primary-key snapshot. */ +public class PrimaryKeyFullTextScan implements FullTextScan { + + private final FileStoreTable table; + private final PrimaryKeyIndexDefinition definition; + @Nullable private final PartitionPredicate partitionFilter; + + public PrimaryKeyFullTextScan( + FileStoreTable table, + PrimaryKeyIndexDefinition definition, + @Nullable PartitionPredicate partitionFilter) { + checkArgument( + definition.family() == PrimaryKeyIndexDefinition.Family.FULL_TEXT, + "Primary-key full-text scan requires a full-text index definition."); + this.table = table; + this.definition = definition; + this.partitionFilter = partitionFilter; + } + + @Override + public Plan scan() { + SnapshotReader snapshotReader = table.newSnapshotReader().keepStats(); + DataTableScan dataScan = table.newScan(ignored -> snapshotReader); + checkArgument( + dataScan instanceof PrimaryKeyBatchScan, + "Primary-key full-text search requires a primary-key batch scan."); + PrimaryKeyBatchScan batchScan = (PrimaryKeyBatchScan) dataScan; + if (partitionFilter != null) { + batchScan.withPartitionFilter(partitionFilter); + } + TableScan.Plan tablePlan = batchScan.planWithoutAuth(); + if (!(tablePlan instanceof SnapshotReader.Plan)) { + checkArgument( + tablePlan.splits().isEmpty(), + "Primary-key full-text search requires a snapshot plan."); + return new Plan(0, Collections.emptyList()); + } + SnapshotReader.Plan snapshotPlan = (SnapshotReader.Plan) tablePlan; + if (snapshotPlan.snapshotId() == null) { + return new Plan(0, Collections.emptyList()); + } + Snapshot snapshot = snapshotReader.snapshotManager().snapshot(snapshotPlan.snapshotId()); + checkArgument(snapshot != null, "Primary-key full-text snapshot does not exist."); + + IndexFileHandler indexFileHandler = snapshotReader.indexFileHandler(); + checkArgument( + indexFileHandler != null, "Primary-key full-text index handler is unavailable."); + List<IndexManifestEntry> payloadEntries = + indexFileHandler.scan( + snapshot, + entry -> + matchesDefinition(entry) + && (partitionFilter == null + || partitionFilter.test(entry.partition()))); + return plan(snapshot.id(), snapshotPlan.splits(), payloadEntries, definition.fieldId()); + } + + private boolean matchesDefinition(IndexManifestEntry entry) { + IndexFileMeta payload = entry.indexFile(); + GlobalIndexMeta globalMeta = payload.globalIndexMeta(); + if (!PkFullTextIndexFile.INDEX_TYPE.equals(payload.indexType()) + || globalMeta == null + || globalMeta.sourceMeta() == null + || globalMeta.indexFieldId() != definition.fieldId()) { + return false; + } + return true; + } + + static Plan plan( + long snapshotId, + List<? extends Split> dataSplits, + List<IndexManifestEntry> payloadEntries, + int textFieldId) { + Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> payloads = new LinkedHashMap<>(); + for (IndexManifestEntry entry : payloadEntries) { + checkArgument( + entry.kind() == FileKind.ADD, + "Primary-key full-text index file %s is not active.", + entry.indexFile().fileName()); + Pair<BinaryRow, Integer> key = Pair.of(entry.partition(), entry.bucket()); + payloads.computeIfAbsent(key, ignored -> new ArrayList<>()).add(entry.indexFile()); + } + + Map<Pair<BinaryRow, Integer>, BucketAccumulator> buckets = new LinkedHashMap<>(); + for (Split split : dataSplits) { + DataSplit dataSplit = unwrapDataSplit(split); + checkArgument( + dataSplit.snapshotId() == snapshotId, + "Data split snapshot %s does not match full-text scan snapshot %s.", + dataSplit.snapshotId(), + snapshotId); + checkArgument( + !dataSplit.isStreaming(), + "Primary-key full-text search requires a batch split."); + if (dataSplit.bucket() < 0) { + continue; + } + Pair<BinaryRow, Integer> key = Pair.of(dataSplit.partition(), dataSplit.bucket()); + BucketAccumulator accumulator = buckets.get(key); + if (accumulator == null) { + accumulator = new BucketAccumulator(dataSplit); + buckets.put(key, accumulator); + } + accumulator.add(dataSplit); + } + + List<PrimaryKeyFullTextSearchSplit> result = new ArrayList<>(); + for (Map.Entry<Pair<BinaryRow, Integer>, BucketAccumulator> entry : buckets.entrySet()) { + BucketAccumulator bucket = entry.getValue(); + if (bucket.isEmpty()) { + continue; + } + PkFullTextBucketIndexState state = + PkFullTextBucketIndexState.fromActivePayloads( + textFieldId, + payloads.getOrDefault(entry.getKey(), Collections.emptyList())); + Set<String> activeSources = bucket.dataFileNames(); + Map<String, IndexFileMeta> currentPayloads = new LinkedHashMap<>(); + Set<String> covered = new HashSet<>(); + for (Map.Entry<String, IndexFileMeta> payload : + state.payloadBySourceFile().entrySet()) { + if (activeSources.contains(payload.getKey())) { + currentPayloads.put(payload.getValue().fileName(), payload.getValue()); + covered.add(payload.getKey()); + } + } + List<String> uncovered = new ArrayList<>(); + for (DataFileMeta dataFile : bucket.dataFiles()) { + if (!covered.contains(dataFile.fileName())) { + uncovered.add(dataFile.fileName()); + } + } + result.add( + new PrimaryKeyFullTextSearchSplit( + bucket.build(), new ArrayList<>(currentPayloads.values()), uncovered)); + } + return new Plan(snapshotId, result); + } + + private static DataSplit unwrapDataSplit(Split split) { + if (split instanceof IndexedSplit) { + return ((IndexedSplit) split).dataSplit(); + } + checkArgument( + split instanceof DataSplit, + "Unsupported primary-key full-text source split: %s.", + split.getClass().getName()); + return (DataSplit) split; + } + + /** Immutable snapshot full-text plan. */ + public static class Plan implements FullTextScan.Plan { + + private final long snapshotId; + private final List<FullTextSearchSplit> splits; + + private Plan(long snapshotId, List<PrimaryKeyFullTextSearchSplit> splits) { + this.snapshotId = snapshotId; + this.splits = Collections.unmodifiableList(new ArrayList<FullTextSearchSplit>(splits)); + } + + public long snapshotId() { + return snapshotId; + } + + @Override + public List<FullTextSearchSplit> splits() { + return splits; + } + } + + private static class BucketAccumulator { + + private final long snapshotId; + private final BinaryRow partition; + private final int bucket; + private final String bucketPath; + @Nullable private final Integer totalBuckets; + private final List<DataFileMeta> dataFiles = new ArrayList<>(); + private final List<DeletionFile> deletionFiles = new ArrayList<>(); + private final Set<String> dataFileNames = new HashSet<>(); + private boolean hasDeletionMetadata; + + private BucketAccumulator(DataSplit split) { + this.snapshotId = split.snapshotId(); + this.partition = split.partition(); + this.bucket = split.bucket(); + this.bucketPath = split.bucketPath(); + this.totalBuckets = split.totalBuckets(); + } + + private void add(DataSplit split) { + checkArgument( + snapshotId == split.snapshotId() + && partition.equals(split.partition()) + && bucket == split.bucket() + && bucketPath.equals(split.bucketPath()), + "Cannot combine data splits from different snapshot buckets."); + checkArgument( + totalBuckets == null + ? split.totalBuckets() == null + : totalBuckets.equals(split.totalBuckets()), + "Bucket split total-bucket metadata is inconsistent."); + List<DeletionFile> splitDeletions = split.deletionFiles().orElse(null); + checkArgument( + splitDeletions == null || splitDeletions.size() == split.dataFiles().size(), + "Deletion files must align with data files in a full-text bucket split."); + hasDeletionMetadata |= splitDeletions != null; + for (int i = 0; i < split.dataFiles().size(); i++) { + DataFileMeta file = split.dataFiles().get(i); + if (!PrimaryKeyIndexSourcePolicy.shouldRead(file)) { Review Comment: Non-FAST modes can silently miss fresh or uncompacted rows here. `shouldRead` only retains COMPACT files at level > 0 (and the planner also skips negative buckets), while `PrimaryKeyFullTextRead` only raw-searches `uncoveredDataFiles` from this already-filtered set. Therefore L0, APPEND, and postpone-bucket files are searched by neither the index nor the fallback even when `global-index.search-mode` is `full` or `detail`, although those modes promise to scan unindexed data. Please keep these files in a merge-aware raw fallback (building each L0 file independently is not sufficient for primary-key semantics), or otherwise prevent this path from claiming complete-mode semantics. An end-to-end test with a fresh uncompacted matching row would catch this. ########## paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearch.java: ########## @@ -0,0 +1,278 @@ +/* + * 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.paimon.index.pkfulltext; + +import org.apache.paimon.deletionvectors.DeletionVector; +import org.apache.paimon.globalindex.GlobalIndexReader; +import org.apache.paimon.globalindex.ScoredGlobalIndexResult; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile; +import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.predicate.FullTextSearch; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.PrimaryKeyFullTextSearchSplit; +import org.apache.paimon.table.source.PrimaryKeySearchPosition; +import org.apache.paimon.table.source.PrimaryKeySearchRanker; +import org.apache.paimon.utils.IOUtils; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Searches source-backed full-text payloads and fuses their local ranks. */ +public class PrimaryKeyFullTextBucketSearch { + + private final ReaderFactory readerFactory; + + public PrimaryKeyFullTextBucketSearch(ReaderFactory readerFactory) { + this.readerFactory = readerFactory; + } + + public List<PrimaryKeySearchPosition> search( + PrimaryKeyFullTextSearchSplit split, + Map<String, DeletionVector> deletionVectors, + String column, + String query, + int limit) { + List<List<PrimaryKeySearchPosition>> localRankings = + searchRankings(split, deletionVectors, column, query, limit); + if (localRankings.isEmpty()) { + return Collections.emptyList(); + } + return PrimaryKeySearchRanker.rrf(localRankings, limit); + } + + public List<List<PrimaryKeySearchPosition>> searchRankings( + PrimaryKeyFullTextSearchSplit split, + Map<String, DeletionVector> deletionVectors, + String column, + String query, + int limit) { + checkArgument(limit > 0, "Full-text search limit must be positive: %s.", limit); + DataSplit dataSplit = split.dataSplit(); + Map<String, DataFileMeta> files = new HashMap<>(); + for (DataFileMeta file : dataSplit.dataFiles()) { + checkArgument( + files.put(file.fileName(), file) == null, + "Duplicate full-text source file %s.", + file.fileName()); + } + + List<PayloadRequest> requests = new ArrayList<>(); + for (IndexFileMeta payload : split.payloadFiles()) { + List<SourceRange> sourceRanges = new ArrayList<>(); + boolean needsInclude = false; + long totalRowCount = 0; + for (PrimaryKeyIndexSourceFile source : + PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFiles()) { + DataFileMeta file = files.get(source.fileName()); + if (file != null) { + checkArgument( + source.rowCount() == file.rowCount(), + "Full-text payload %s source row count does not match data file %s.", + payload.fileName(), + source.fileName()); + } + DeletionVector deletionVector = deletionVectors.get(source.fileName()); + needsInclude |= + file == null || (deletionVector != null && !deletionVector.isEmpty()); + sourceRanges.add( + new SourceRange( + source.fileName(), totalRowCount, source.rowCount(), file != null)); + totalRowCount = Math.addExact(totalRowCount, source.rowCount()); + } + RoaringNavigableMap64 include = + needsInclude ? liveRows(sourceRanges, deletionVectors) : null; + if (include != null && include.isEmpty()) { + continue; + } + GlobalIndexReader reader = readerFactory.create(payload); + CompletableFuture<Optional<ScoredGlobalIndexResult>> future; + try { + FullTextSearch predicate = new FullTextSearch(column, query, limit); + if (include != null) { + predicate.withIncludeRowIds(include); + } + future = + reader.visitFullTextSearch(predicate) + .whenComplete((ignored, error) -> IOUtils.closeQuietly(reader)); + } catch (RuntimeException | Error t) { + IOUtils.closeQuietly(reader); + throw t; + } + requests.add(new PayloadRequest(sourceRanges, totalRowCount, include, future)); + } + + CompletableFuture.allOf( + requests.stream() + .map(request -> request.future) + .toArray(CompletableFuture[]::new)) + .join(); + List<List<PrimaryKeySearchPosition>> localRankings = new ArrayList<>(requests.size()); + for (PayloadRequest request : requests) { + Optional<ScoredGlobalIndexResult> result = request.future.join(); + if (!result.isPresent()) { + continue; + } + ScoredGlobalIndexResult scored = result.get(); + List<PrimaryKeySearchPosition> ranking = new ArrayList<>(); + for (long rowId : scored.results()) { + checkArgument( + rowId >= 0 && rowId < request.totalRowCount, + "Full-text index returned archive row position %s outside row count %s.", + rowId, + request.totalRowCount); + if (request.include != null && !request.include.contains(rowId)) { + continue; + } + SourceRange source = request.source(rowId); + checkArgument( + source != null && source.active, + "Full-text index returned row position %s from an inactive source.", + rowId); + ranking.add( + new PrimaryKeySearchPosition( + dataSplit.partition(), + dataSplit.bucket(), + source.fileName, + rowId - source.offset, + scored.scoreGetter().score(rowId))); + } + ranking.sort( + (left, right) -> { + int scoreOrder = Float.compare(right.score(), left.score()); + if (scoreOrder != 0) { + return scoreOrder; + } + int fileOrder = left.dataFileName().compareTo(right.dataFileName()); + return fileOrder != 0 + ? fileOrder + : Long.compare(left.rowPosition(), right.rowPosition()); + }); + localRankings.add(ranking); + } + return Collections.unmodifiableList(localRankings); + } + + @Nullable + private static RoaringNavigableMap64 liveRows( + long rowCount, @Nullable DeletionVector deletionVector) { + if (deletionVector == null || deletionVector.isEmpty()) { + return null; + } + RoaringNavigableMap64 live = new RoaringNavigableMap64(); + if (rowCount > 0) { + live.addRange(new Range(0, rowCount - 1)); + } + RoaringNavigableMap64 deleted = new RoaringNavigableMap64(); + deletionVector.forEachDeletedPosition( + position -> { + checkArgument( + position >= 0 && position < rowCount, + "Deletion vector contains invalid row position %s.", + position); + deleted.add(position); + }); + live.andNot(deleted); + return live; + } + + private static RoaringNavigableMap64 liveRows( + List<SourceRange> sourceRanges, Map<String, DeletionVector> deletionVectors) { + RoaringNavigableMap64 include = new RoaringNavigableMap64(); + for (SourceRange source : sourceRanges) { + if (!source.active) { + continue; + } + RoaringNavigableMap64 local = + liveRows(source.rowCount, deletionVectors.get(source.fileName)); + if (local == null) { + if (source.rowCount > 0) { + include.addRange(new Range(source.offset, source.offset + source.rowCount - 1)); + } + } else { + for (long rowId : local) { Review Comment: For a source with any non-empty deletion vector, `liveRows(rowCount, dv)` contains almost every live row and this loop enumerates all of them just to apply the archive offset. A 100M-row file with one deletion performs roughly 100M iterations per query. Please build the global include bitmap from active source ranges, construct an archive-offset deleted bitmap by iterating only deleted positions, and apply `andNot`; the work then scales with source count plus deleted rows rather than all live rows. ########## paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java: ########## @@ -66,20 +68,51 @@ public FullTextSearchBuilder withQuery(String fieldName, String query) { @Override public FullTextScan newFullTextScan() { - return new FullTextScanImpl(table, partitionFilter, textColumns()); + DataField textColumn = textColumn(); + Optional<PrimaryKeyIndexDefinition> definition = primaryKeyFullTextDefinition(textColumn); + return definition.isPresent() + ? new PrimaryKeyFullTextScan(table, definition.get(), partitionFilter) Review Comment: This dispatch is also used by `HybridSearchBuilderImpl.newFullTextSearchBuilder`, so the primary-key path is exposed to hybrid search even though hybrid support is deferred. The returned `PrimaryKeyScoredResult` throws from both `results()` and `scoreGetter()`, while `HybridSearchBuilderImpl.rank` calls `results()` unconditionally, so any primary-key full-text hybrid route fails at runtime. Please explicitly reject or retain the old route for hybrid search until physical-position ranking is supported, or adapt hybrid ranking to `GlobalIndexSplitResult`. A primary-key full-text hybrid regression test would expose the issue. -- 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]
