This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new ab4343e0 fix(global-index): exclude deleted rows from full-text
ranking (#448)
ab4343e0 is described below
commit ab4343e0d9b6b2fd9944b5be01cc9d09962c46d2
Author: Nicholas Jiang <[email protected]>
AuthorDate: Thu Oct 22 13:17:13 2026 +0700
fix(global-index): exclude deleted rows from full-text ranking (#448)
---
docs/source/user_guide/global_index.rst | 25 ++
src/paimon/CMakeLists.txt | 3 +
.../global_index/global_index_live_row_filter.cpp | 213 +++++++++++++++++
.../global_index/global_index_live_row_filter.h | 63 +++++
.../core/global_index/global_index_scan_impl.cpp | 38 ++-
.../core/global_index/global_index_scan_impl.h | 7 +-
.../global_index/live_row_global_index_reader.cpp | 232 ++++++++++++++++++
.../global_index/live_row_global_index_reader.h | 115 +++++++++
.../live_row_global_index_reader_test.cpp | 262 +++++++++++++++++++++
src/paimon/core/io/data_file_meta.cpp | 12 +
src/paimon/core/io/data_file_meta.h | 5 +
src/paimon/core/io/data_file_meta_test.cpp | 30 +++
src/paimon/core/operation/file_store_scan.h | 9 +
.../source/data_evolution_split_generator.cpp | 5 +-
.../core/table/source/split_generator_test.cpp | 17 ++
test/inte/data_evolution_table_test.cpp | 79 ++++++-
test/inte/global_index_test.cpp | 120 ++++++++++
17 files changed, 1219 insertions(+), 16 deletions(-)
diff --git a/docs/source/user_guide/global_index.rst
b/docs/source/user_guide/global_index.rst
index afee3391..f3dca744 100644
--- a/docs/source/user_guide/global_index.rst
+++ b/docs/source/user_guide/global_index.rst
@@ -181,6 +181,31 @@ ranges is ranked only by the range that starts first, or
ends first if both star
A shard ignores the ``include_row_ids`` outside the rows it ranks and is not
searched if none
remain. The result iterates rows by ascending row id, not by score.
+When ``deletion-vectors.enabled`` and ``row-tracking.enabled`` are set,
readers created by
+``GlobalIndexScan`` rank only the rows that are live in the snapshot the scan
reads, as Java Paimon
+does, so rows deleted by deletion vectors do not take ``limit`` slots. BM25
statistics still come
+from the whole index. A later snapshot can delete some of the matched rows, so
pass one snapshot id
+to both the index scan and the data scan:
+
+.. code-block:: cpp
+
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<paimon::GlobalIndexScan> index_scan,
+ paimon::GlobalIndexScan::Create(table_path, snapshot_id,
/*partitions=*/std::nullopt,
+ /*options=*/{}, /*file_system=*/nullptr,
+ /*executor=*/nullptr, pool));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::GlobalIndexReader> reader,
+ index_scan->CreateReader("f0", "full-text",
+
/*row_range_index=*/std::nullopt));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::ScoredGlobalIndexResult>
result,
+
reader->VisitFullTextSearch(std::make_shared<paimon::FullTextSearch>(
+ "f0", R"({"match":{"query":"paimon lake"}})",
/*limit=*/10)));
+
+ paimon::ScanContextBuilder scan_builder(table_path);
+ scan_builder.AddOption(paimon::Options::SCAN_MODE, "from-snapshot")
+ .AddOption(paimon::Options::SCAN_SNAPSHOT_ID,
std::to_string(snapshot_id))
+ .SetGlobalIndexResult(result);
+
**Configuration:**
Table options with the ``full-text.`` prefix are passed to the engine with the
prefix removed.
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index da5213dc..b3d1a1e2 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -269,9 +269,11 @@ set(PAIMON_CORE_SRCS
core/deletionvectors/deletion_vectors_index_file.cpp
core/deletionvectors/deletion_vector_index_file_writer.cpp
core/global_index/global_index_evaluator_impl.cpp
+ core/global_index/global_index_live_row_filter.cpp
core/global_index/global_index_scan.cpp
core/global_index/global_index_scan_impl.cpp
core/global_index/global_index_write_task.cpp
+ core/global_index/live_row_global_index_reader.cpp
core/index/index_file_handler.cpp
core/index/global_index_meta.cpp
core/index/index_file_meta_serializer.cpp
@@ -865,6 +867,7 @@ if(PAIMON_BUILD_TESTS)
core/global_index/global_index_file_manager_test.cpp
core/global_index/global_index_evaluator_impl_test.cpp
core/global_index/indexed_split_test.cpp
+ core/global_index/live_row_global_index_reader_test.cpp
core/manifest/file_source_test.cpp
core/manifest/file_kind_test.cpp
core/manifest/manifest_entry_writer_test.cpp
diff --git a/src/paimon/core/global_index/global_index_live_row_filter.cpp
b/src/paimon/core/global_index/global_index_live_row_filter.cpp
new file mode 100644
index 00000000..faf9e040
--- /dev/null
+++ b/src/paimon/core/global_index/global_index_live_row_filter.cpp
@@ -0,0 +1,213 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/global_index/global_index_live_row_filter.h"
+
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "fmt/format.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/core/deletionvectors/deletion_vector.h"
+#include "paimon/core/index/index_file_handler.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/manifest/index_manifest_file.h"
+#include "paimon/core/manifest/manifest_file.h"
+#include "paimon/core/manifest/manifest_list.h"
+#include "paimon/core/operation/append_only_file_store_scan.h"
+#include "paimon/core/operation/data_evolution_file_store_scan.h"
+#include "paimon/core/operation/file_store_scan.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/table/bucket_mode.h"
+#include "paimon/core/table/source/append_only_split_generator.h"
+#include "paimon/core/table/source/data_evolution_split_generator.h"
+#include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/core/table/source/scan_mode.h"
+#include "paimon/core/table/source/snapshot/snapshot_reader.h"
+#include "paimon/core/table/source/split_generator.h"
+#include "paimon/core/utils/branch_manager.h"
+#include "paimon/core/utils/field_mapping.h"
+#include "paimon/core/utils/file_store_path_factory.h"
+#include "paimon/core/utils/index_file_path_factories.h"
+#include "paimon/core/utils/snapshot_manager.h"
+#include "paimon/scan_context.h"
+#include "paimon/table/source/plan.h"
+#include "paimon/utils/range.h"
+
+namespace paimon {
+namespace {
+/// Creates a reader over the data files of an append table in the partitions
accepted by
+/// `partition_filter`, as Java `FileStoreTable#newSnapshotReader` with
`withPartitionFilter` does.
+Result<std::unique_ptr<SnapshotReader>> CreateSnapshotReader(
+ const std::string& root_path, const std::shared_ptr<TableSchema>&
table_schema,
+ const std::shared_ptr<PredicateFilter>& partition_filter, const
CoreOptions& options,
+ const std::shared_ptr<Executor>& executor, const
std::shared_ptr<MemoryPool>& pool) {
+ std::shared_ptr<FileSystem> fs = options.GetFileSystem();
+ std::string branch = BranchManager::NormalizeBranch(options.GetBranch());
+ auto arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> external_paths,
options.CreateExternalPaths());
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::string>
global_index_external_path,
+ options.CreateGlobalIndexExternalPath());
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileStorePathFactory> path_factory,
+ FileStorePathFactory::Create(
+ root_path, arrow_schema, table_schema->PartitionKeys(),
+ options.GetPartitionDefaultName(),
options.GetFileFormat()->Identifier(),
+ options.DataFilePrefix(), options.LegacyPartitionNameEnabled(),
external_paths,
+ global_index_external_path, options.IndexFileInDataFileDir(),
pool));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<ManifestList> manifest_list,
+ ManifestList::Create(fs, options.GetManifestFormat(),
options.GetManifestCompression(),
+ path_factory, options.GetCache(), pool));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<arrow::Schema> partition_schema,
+ FieldMapping::GetPartitionSchema(arrow_schema,
table_schema->PartitionKeys()));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<ManifestFile> manifest_file,
+ ManifestFile::Create(fs, options.GetManifestFormat(),
options.GetManifestCompression(),
+ path_factory,
options.GetManifestTargetFileSize(), pool, options,
+ partition_schema));
+ auto snapshot_manager = std::make_shared<SnapshotManager>(fs, root_path,
branch);
+ auto schema_manager = std::make_shared<SchemaManager>(fs, root_path,
branch);
+ auto scan_filter = std::make_shared<ScanFilter>(
+ /*predicate=*/nullptr,
+ /*partition_filters=*/std::vector<std::map<std::string,
std::string>>(),
+ /*bucket_filter=*/std::nullopt);
+
+ std::unique_ptr<FileStoreScan> scan;
+ std::unique_ptr<SplitGenerator> split_generator;
+ if (options.DataEvolutionEnabled()) {
+ PAIMON_ASSIGN_OR_RAISE(
+ scan, DataEvolutionFileStoreScan::Create(
+ snapshot_manager, schema_manager, manifest_list,
manifest_file, table_schema,
+ arrow_schema, scan_filter, options, executor, pool));
+ split_generator = std::make_unique<DataEvolutionSplitGenerator>(
+ options.GetSourceSplitTargetSize(),
options.GetSourceSplitOpenFileCost(),
+ options.BlobSplitByFileSize());
+ } else {
+ PAIMON_ASSIGN_OR_RAISE(
+ scan, AppendOnlyFileStoreScan::Create(snapshot_manager,
schema_manager, manifest_list,
+ manifest_file, table_schema,
arrow_schema,
+ scan_filter, options,
executor, pool));
+ split_generator = std::make_unique<AppendOnlySplitGenerator>(
+ options.GetSourceSplitTargetSize(),
options.GetSourceSplitOpenFileCost(),
+ BucketMode::BUCKET_UNAWARE);
+ }
+ scan->WithTablePath(root_path)->WithPartitionFilter(partition_filter,
partition_schema);
+
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<IndexManifestFile> index_manifest_file,
+ IndexManifestFile::Create(fs, options.GetManifestFormat(),
options.GetManifestCompression(),
+ path_factory, options.GetBucket(), pool,
options));
+ auto index_file_handler = std::make_unique<IndexFileHandler>(
+ fs, std::move(index_manifest_file),
std::make_shared<IndexFilePathFactories>(path_factory),
+ options.DeletionVectorsBitmap64(), pool);
+ return std::make_unique<SnapshotReader>(
+ std::move(scan), path_factory, std::move(split_generator),
std::move(index_file_handler));
+}
+
+/// Removes the rows deleted by the deletion vectors of `split` from
`live_rows`. Each deletion
+/// vector is keyed by the data file whose first row id its positions are
relative to.
+Status SubtractDeletedRows(const DataSplitImpl& split, const std::string&
root_path,
+ const Snapshot& snapshot, const
std::shared_ptr<FileSystem>& fs,
+ const std::shared_ptr<MemoryPool>& pool,
RoaringBitmap64* live_rows) {
+ const std::vector<std::shared_ptr<DataFileMeta>>& files =
split.DataFiles();
+ DeletionVector::Factory deletion_vector_factory =
DeletionVector::CreateFactory(
+ fs, DeletionVector::CreateDeletionFileMap(files,
split.DeletionFiles()), pool);
+ for (const auto& file : files) {
+ if (!file->first_row_id) {
+ continue;
+ }
+ int64_t first_row_id = file->first_row_id.value();
+ Result<std::shared_ptr<DeletionVector>> deletion_vector_result =
+ deletion_vector_factory(file->file_name);
+ if (!deletion_vector_result.ok()) {
+ const Status& status = deletion_vector_result.status();
+ return status.WithMessage(fmt::format(
+ "failed to read deletion vector of data file {} in snapshot {}
of table {}: {}",
+ file->file_name, snapshot.Id(), root_path, status.message()));
+ }
+ std::shared_ptr<DeletionVector> deletion_vector =
std::move(deletion_vector_result).value();
+ if (!deletion_vector || deletion_vector->IsEmpty()) {
+ continue;
+ }
+ RoaringBitmap64 deleted_rows;
+ PAIMON_RETURN_NOT_OK(deletion_vector->ForEachDeletedPosition(
+ [&deleted_rows, first_row_id](int64_t position) {
+ deleted_rows.Add(first_row_id + position);
+ }));
+ *live_rows -= deleted_rows;
+ }
+ return Status::OK();
+}
+} // namespace
+
+Result<std::optional<RoaringBitmap64>> GlobalIndexLiveRowFilter::LiveRows(
+ const std::string& root_path, const std::shared_ptr<TableSchema>&
table_schema,
+ const Snapshot& snapshot, const std::shared_ptr<PredicateFilter>&
partition_filter,
+ const CoreOptions& options, const std::shared_ptr<Executor>& executor,
+ const std::shared_ptr<MemoryPool>& pool) {
+ // Without row tracking, as in primary-key tables, data files carry no row
ids to filter.
+ if (!options.DeletionVectorsEnabled() || !options.RowTrackingEnabled()) {
+ return std::optional<RoaringBitmap64>();
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<SnapshotReader> snapshot_reader,
+ CreateSnapshotReader(root_path, table_schema, partition_filter,
options, executor, pool));
+ snapshot_reader->WithMode(ScanMode::ALL)->WithSnapshot(snapshot);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Plan> plan,
snapshot_reader->Read());
+
+ std::vector<std::shared_ptr<DataSplitImpl>> splits;
+ splits.reserve(plan->Splits().size());
+ for (const auto& split : plan->Splits()) {
+ auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+ if (!data_split) {
+ return Status::Invalid("cannot cast split to DataSplitImpl when
reading live rows");
+ }
+ splits.push_back(std::move(data_split));
+ }
+
+ RoaringBitmap64 live_rows;
+ // Phase 1: union the row id range of every data file.
+ for (const auto& split : splits) {
+ for (const auto& file : split->DataFiles()) {
+ if (file->first_row_id && file->row_count > 0) {
+ PAIMON_ASSIGN_OR_RAISE(Range row_id_range,
file->NonNullRowIdRange());
+ // AddRange is half-open and `row_id_range.to` may be the
largest int64 value.
+ live_rows.AddRange(row_id_range.from, row_id_range.to);
+ live_rows.Add(row_id_range.to);
+ }
+ }
+ }
+ // Phase 2: subtract each deletion vector. Every range is added first, so
a row deleted by one
+ // file's deletion vector is not added back by another file covering it,
and only one deletion
+ // vector is held at a time.
+ for (const auto& split : splits) {
+ PAIMON_RETURN_NOT_OK(SubtractDeletedRows(*split, root_path, snapshot,
+ options.GetFileSystem(),
pool, &live_rows));
+ }
+ return std::optional<RoaringBitmap64>(std::move(live_rows));
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/global_index/global_index_live_row_filter.h
b/src/paimon/core/global_index/global_index_live_row_filter.h
new file mode 100644
index 00000000..4f607e69
--- /dev/null
+++ b/src/paimon/core/global_index/global_index_live_row_filter.h
@@ -0,0 +1,63 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+#include <optional>
+#include <string>
+
+#include "paimon/common/predicate/predicate_filter.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/snapshot.h"
+#include "paimon/executor.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+/// Live-row filtering shared by global index search readers, aligned with Java
+/// `GlobalIndexLiveRowFilter`. Deletion vectors are applied only when rows
are read, so a search
+/// that ranks deleted rows can fill its top-k slots with rows the read then
drops.
+class GlobalIndexLiveRowFilter {
+ public:
+ GlobalIndexLiveRowFilter() = delete;
+ ~GlobalIndexLiveRowFilter() = delete;
+
+ /// Returns the global row ids of the rows that are live in `snapshot`:
the row id ranges of
+ /// the data files in the partitions accepted by `partition_filter`, minus
the rows deleted by
+ /// their deletion vectors. Returns nullopt when deletion vectors or row
tracking are disabled,
+ /// so every row counts as live.
+ /// @param root_path Root directory of the table.
+ /// @param table_schema Schema of the table.
+ /// @param snapshot The snapshot to read; pin it to the one the
global index is
+ /// read from, so row ids and live rows agree.
+ /// @param partition_filter Predicate over the partition fields, or null
for all partitions.
+ /// @param options Options of the table.
+ /// @param executor Executor for reading manifests.
+ /// @param pool Memory pool for reading manifests and deletion
vectors.
+ static Result<std::optional<RoaringBitmap64>> LiveRows(
+ const std::string& root_path, const std::shared_ptr<TableSchema>&
table_schema,
+ const Snapshot& snapshot, const std::shared_ptr<PredicateFilter>&
partition_filter,
+ const CoreOptions& options, const std::shared_ptr<Executor>& executor,
+ const std::shared_ptr<MemoryPool>& pool);
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/global_index/global_index_scan_impl.cpp
b/src/paimon/core/global_index/global_index_scan_impl.cpp
index 511be25c..b20437d4 100644
--- a/src/paimon/core/global_index/global_index_scan_impl.cpp
+++ b/src/paimon/core/global_index/global_index_scan_impl.cpp
@@ -31,24 +31,25 @@
#include "paimon/common/global_index/union_global_index_reader.h"
#include "paimon/common/utils/scope_guard.h"
#include "paimon/core/global_index/global_index_evaluator_impl.h"
+#include "paimon/core/global_index/global_index_live_row_filter.h"
#include "paimon/core/index/index_file_handler.h"
#include "paimon/global_index/bitmap_global_index_result.h"
#include "paimon/global_index/global_indexer.h"
#include "paimon/global_index/global_indexer_factory.h"
namespace paimon {
-GlobalIndexScanImpl::GlobalIndexScanImpl(const std::shared_ptr<TableSchema>&
table_schema,
- const CoreOptions& options,
- const
std::shared_ptr<IndexPathFactory>& path_factory,
- IndexMetaMap&& index_metas,
- const std::shared_ptr<Executor>&
executor,
- const std::shared_ptr<MemoryPool>&
pool)
+GlobalIndexScanImpl::GlobalIndexScanImpl(
+ const std::shared_ptr<TableSchema>& table_schema, const CoreOptions&
options,
+ const std::shared_ptr<IndexPathFactory>& path_factory, IndexMetaMap&&
index_metas,
+ LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier,
+ const std::shared_ptr<Executor>& executor, const
std::shared_ptr<MemoryPool>& pool)
: pool_(pool),
table_schema_(table_schema),
options_(options),
index_file_manager_(std::make_shared<GlobalIndexFileManager>(
options.GetFileSystem(), path_factory,
/*checkpoint_path_factory=*/nullptr)),
index_metas_(std::move(index_metas)),
+ live_rows_supplier_(std::move(live_rows_supplier)),
executor_(executor) {}
Result<std::unique_ptr<GlobalIndexScanImpl>> GlobalIndexScanImpl::Create(
@@ -122,8 +123,18 @@ Result<std::unique_ptr<GlobalIndexScanImpl>>
GlobalIndexScanImpl::Create(
index_metas[index_meta->index_field_id][index_file_meta->IndexType()][range].push_back(
index_file_meta);
}
- return std::unique_ptr<GlobalIndexScanImpl>(new GlobalIndexScanImpl(
- table_schema, options, path_factory, std::move(index_metas),
final_executor, pool));
+ // Full-text searches rank only the rows live in the snapshot the index is
read from.
+ LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier;
+ if (options.DeletionVectorsEnabled()) {
+ live_rows_supplier = LiveRowGlobalIndexReader::CacheLiveRows(
+ [root_path, table_schema, snapshot, partitions, options,
final_executor, pool]() {
+ return GlobalIndexLiveRowFilter::LiveRows(
+ root_path, table_schema, snapshot, partitions, options,
final_executor, pool);
+ });
+ }
+ return std::unique_ptr<GlobalIndexScanImpl>(
+ new GlobalIndexScanImpl(table_schema, options, path_factory,
std::move(index_metas),
+ std::move(live_rows_supplier), final_executor,
pool));
}
Result<std::shared_ptr<GlobalIndexEvaluator>>
GlobalIndexScanImpl::GetOrCreateIndexEvaluator() {
@@ -235,8 +246,15 @@ Result<std::vector<std::shared_ptr<GlobalIndexReader>>>
GlobalIndexScanImpl::Cre
if (union_readers.empty()) {
continue;
}
- readers.push_back(std::make_shared<UnionGlobalIndexReader>(
- std::move(union_readers), std::move(row_ranges),
std::move(search_ranges), executor_));
+ if (live_rows_supplier_) {
+ readers.push_back(LiveRowGlobalIndexReader::CreateUnion(
+ std::move(union_readers), std::move(row_ranges),
std::move(search_ranges),
+ executor_, live_rows_supplier_));
+ } else {
+ readers.push_back(std::make_shared<UnionGlobalIndexReader>(
+ std::move(union_readers), std::move(row_ranges),
std::move(search_ranges),
+ executor_));
+ }
}
return readers;
}
diff --git a/src/paimon/core/global_index/global_index_scan_impl.h
b/src/paimon/core/global_index/global_index_scan_impl.h
index a7921d77..bb26ae8e 100644
--- a/src/paimon/core/global_index/global_index_scan_impl.h
+++ b/src/paimon/core/global_index/global_index_scan_impl.h
@@ -29,6 +29,7 @@
#include "paimon/core/core_options.h"
#include "paimon/core/global_index/global_index_evaluator.h"
#include "paimon/core/global_index/global_index_file_manager.h"
+#include "paimon/core/global_index/live_row_global_index_reader.h"
#include "paimon/core/index/index_file_meta.h"
#include "paimon/core/index/index_path_factory.h"
#include "paimon/core/schema/table_schema.h"
@@ -67,7 +68,9 @@ class GlobalIndexScanImpl : public GlobalIndexScan {
GlobalIndexScanImpl(const std::shared_ptr<TableSchema>& table_schema,
const CoreOptions& options,
const std::shared_ptr<IndexPathFactory>& path_factory,
- IndexMetaMap&& index_metas, const
std::shared_ptr<Executor>& executor,
+ IndexMetaMap&& index_metas,
+ LiveRowGlobalIndexReader::LiveRowsSupplier
live_rows_supplier,
+ const std::shared_ptr<Executor>& executor,
const std::shared_ptr<MemoryPool>& pool);
Result<std::shared_ptr<GlobalIndexEvaluator>> GetOrCreateIndexEvaluator();
@@ -88,6 +91,8 @@ class GlobalIndexScanImpl : public GlobalIndexScan {
CoreOptions options_;
std::shared_ptr<GlobalIndexFileManager> index_file_manager_;
IndexMetaMap index_metas_;
+ /// Supplies the rows live in the scanned snapshot; empty when deletion
vectors are disabled.
+ LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier_;
std::shared_ptr<Executor> executor_;
std::shared_ptr<GlobalIndexEvaluator> evaluator_;
};
diff --git a/src/paimon/core/global_index/live_row_global_index_reader.cpp
b/src/paimon/core/global_index/live_row_global_index_reader.cpp
new file mode 100644
index 00000000..9bdf1449
--- /dev/null
+++ b/src/paimon/core/global_index/live_row_global_index_reader.cpp
@@ -0,0 +1,232 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/global_index/live_row_global_index_reader.h"
+
+#include <algorithm>
+#include <cassert>
+#include <mutex>
+#include <utility>
+
+#include "paimon/common/global_index/union_global_index_reader.h"
+
+namespace paimon {
+namespace {
+/// Returns whether `full_text_search` may rank a row of `range`: it has no
include row ids, or one
+/// of them is in `range`.
+bool MayRankRowsOf(const FullTextSearch& full_text_search, const Range& range)
{
+ if (!full_text_search.include_row_ids) {
+ return true;
+ }
+ const RoaringBitmap64& include_row_ids =
full_text_search.include_row_ids.value();
+ auto first = include_row_ids.EqualOrLarger(range.from);
+ return first != include_row_ids.End() && *first <= range.to;
+}
+
+/// A union of LiveRowGlobalIndexReaders that loads the live rows on the
visiting thread before a
+/// full-text search fans out to them.
+class LiveRowUnionGlobalIndexReader : public UnionGlobalIndexReader {
+ public:
+ /// @param ranked_ranges The global row ids the shards rank in a full-text
search, leaving out
+ /// the shards that rank none.
+
LiveRowUnionGlobalIndexReader(std::vector<std::shared_ptr<GlobalIndexReader>>&&
readers,
+ std::vector<Range>&& row_ranges,
+ std::vector<std::optional<Range>>&&
search_ranges,
+ std::vector<Range>&& ranked_ranges,
+ const std::shared_ptr<Executor>& executor,
+ LiveRowGlobalIndexReader::LiveRowsSupplier
live_rows_supplier)
+ : UnionGlobalIndexReader(std::move(readers), std::move(row_ranges),
+ std::move(search_ranges), executor),
+ ranked_ranges_(std::move(ranked_ranges)),
+ live_rows_supplier_(std::move(live_rows_supplier)) {}
+
+ Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitFullTextSearch(
+ const std::shared_ptr<FullTextSearch>& full_text_search) override {
+ // Only a shard search needs live rows: the union rejects a search
without a positive limit
+ // before searching any shard, and skips the shards whose rows the
include row ids miss.
+ if (full_text_search && full_text_search->limit > 0 &&
+ std::any_of(ranked_ranges_.begin(), ranked_ranges_.end(),
[&](const Range& range) {
+ return MayRankRowsOf(*full_text_search, range);
+ })) {
+ PAIMON_RETURN_NOT_OK(live_rows_supplier_());
+ }
+ return UnionGlobalIndexReader::VisitFullTextSearch(full_text_search);
+ }
+
+ private:
+ std::vector<Range> ranked_ranges_;
+ LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier_;
+};
+} // namespace
+
+LiveRowGlobalIndexReader::LiveRowsSupplier
LiveRowGlobalIndexReader::CacheLiveRows(
+ std::function<Result<std::optional<RoaringBitmap64>>()> load) {
+ struct State {
+ std::mutex mutex;
+ bool loaded = false;
+ std::shared_ptr<const RoaringBitmap64> live_rows;
+ };
+ auto state = std::make_shared<State>();
+ return [state, load = std::move(load)]() -> Result<std::shared_ptr<const
RoaringBitmap64>> {
+ std::lock_guard<std::mutex> lock(state->mutex);
+ if (!state->loaded) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<RoaringBitmap64> live_rows,
load());
+ if (live_rows) {
+ state->live_rows =
+ std::make_shared<const
RoaringBitmap64>(std::move(live_rows).value());
+ }
+ state->loaded = true;
+ }
+ return state->live_rows;
+ };
+}
+
+std::shared_ptr<GlobalIndexReader> LiveRowGlobalIndexReader::CreateUnion(
+ std::vector<std::shared_ptr<GlobalIndexReader>>&& shard_readers,
+ std::vector<Range>&& shard_ranges, std::vector<std::optional<Range>>&&
search_ranges,
+ const std::shared_ptr<Executor>& executor, const LiveRowsSupplier&
live_rows_supplier) {
+ assert(shard_readers.size() == shard_ranges.size());
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers;
+ std::vector<Range> ranked_ranges;
+ readers.reserve(shard_readers.size());
+ ranked_ranges.reserve(shard_readers.size());
+ for (size_t i = 0; i < shard_readers.size(); i++) {
+ readers.push_back(std::make_shared<LiveRowGlobalIndexReader>(
+ std::move(shard_readers[i]), shard_ranges[i], live_rows_supplier));
+ if (search_ranges.empty()) {
+ ranked_ranges.push_back(shard_ranges[i]);
+ } else if (search_ranges[i]) {
+ ranked_ranges.push_back(search_ranges[i].value());
+ }
+ }
+ return std::make_shared<LiveRowUnionGlobalIndexReader>(
+ std::move(readers), std::move(shard_ranges), std::move(search_ranges),
+ std::move(ranked_ranges), executor, live_rows_supplier);
+}
+
+LiveRowGlobalIndexReader::LiveRowGlobalIndexReader(std::shared_ptr<GlobalIndexReader>&&
wrapped,
+ const Range& shard_range,
+ LiveRowsSupplier
live_rows_supplier)
+ : wrapped_(std::move(wrapped)),
+ shard_range_(shard_range),
+ live_rows_supplier_(std::move(live_rows_supplier)) {}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitIsNotNull() {
+ return wrapped_->VisitIsNotNull();
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitIsNull() {
+ return wrapped_->VisitIsNull();
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitEqual(
+ const Literal& literal) {
+ return wrapped_->VisitEqual(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitNotEqual(
+ const Literal& literal) {
+ return wrapped_->VisitNotEqual(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitLessThan(
+ const Literal& literal) {
+ return wrapped_->VisitLessThan(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitLessOrEqual(
+ const Literal& literal) {
+ return wrapped_->VisitLessOrEqual(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitGreaterThan(
+ const Literal& literal) {
+ return wrapped_->VisitGreaterThan(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitGreaterOrEqual(
+ const Literal& literal) {
+ return wrapped_->VisitGreaterOrEqual(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitIn(
+ const std::vector<Literal>& literals) {
+ return wrapped_->VisitIn(literals);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitNotIn(
+ const std::vector<Literal>& literals) {
+ return wrapped_->VisitNotIn(literals);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitStartsWith(
+ const Literal& prefix) {
+ return wrapped_->VisitStartsWith(prefix);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitEndsWith(
+ const Literal& suffix) {
+ return wrapped_->VisitEndsWith(suffix);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
LiveRowGlobalIndexReader::VisitContains(
+ const Literal& literal) {
+ return wrapped_->VisitContains(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitLike(
+ const Literal& literal) {
+ return wrapped_->VisitLike(literal);
+}
+
+Result<std::shared_ptr<ScoredGlobalIndexResult>>
LiveRowGlobalIndexReader::VisitVectorSearch(
+ const std::shared_ptr<VectorSearch>& vector_search) {
+ return wrapped_->VisitVectorSearch(vector_search);
+}
+
+Result<std::shared_ptr<ScoredGlobalIndexResult>>
LiveRowGlobalIndexReader::VisitFullTextSearch(
+ const std::shared_ptr<FullTextSearch>& full_text_search) {
+ if (!full_text_search || !MayRankRowsOf(*full_text_search, shard_range_)) {
+ return wrapped_->VisitFullTextSearch(full_text_search);
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<const RoaringBitmap64> live_rows,
live_rows_supplier_());
+ if (!live_rows) {
+ return wrapped_->VisitFullTextSearch(full_text_search);
+ }
+ RoaringBitmap64 include_row_ids;
+ // `to` may be the largest int64 value, so add it apart from the half-open
range.
+ include_row_ids.AddRange(shard_range_.from, shard_range_.to);
+ include_row_ids.Add(shard_range_.to);
+ include_row_ids &= *live_rows;
+ if (full_text_search->include_row_ids) {
+ include_row_ids &= full_text_search->include_row_ids.value();
+ }
+ if (include_row_ids.Cardinality() == shard_range_.Count()) {
+ if (!full_text_search->include_row_ids) {
+ return wrapped_->VisitFullTextSearch(full_text_search);
+ }
+ return wrapped_->VisitFullTextSearch(std::make_shared<FullTextSearch>(
+ full_text_search->field_name, full_text_search->query,
full_text_search->limit));
+ }
+ return wrapped_->VisitFullTextSearch(
+ std::make_shared<FullTextSearch>(full_text_search->field_name,
full_text_search->query,
+ full_text_search->limit,
std::move(include_row_ids)));
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/global_index/live_row_global_index_reader.h
b/src/paimon/core/global_index/live_row_global_index_reader.h
new file mode 100644
index 00000000..4e2589c7
--- /dev/null
+++ b/src/paimon/core/global_index/live_row_global_index_reader.h
@@ -0,0 +1,115 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "paimon/executor.h"
+#include "paimon/global_index/global_index_reader.h"
+#include "paimon/utils/range.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+
+/// A GlobalIndexReader over one index shard that ranks only live rows in
full-text searches, so
+/// rows deleted by deletion vectors do not take top-k slots. As Java
+/// `DataEvolutionFullTextRead#includeRowIds` does, the shard is searched with
the live rows of its
+/// range, intersected with the include row ids if given, while BM25
statistics still come from the
+/// whole index. Predicate visits are delegated unchanged, as the read drops
deleted rows from their
+/// untruncated results, and so are vector searches.
+class LiveRowGlobalIndexReader : public GlobalIndexReader {
+ public:
+ /// Supplies the global row ids of live rows, or null when every row is
live.
+ using LiveRowsSupplier = std::function<Result<std::shared_ptr<const
RoaringBitmap64>>()>;
+
+ /// Returns a supplier that calls `load` on its first call and shares the
live rows it returns
+ /// with later calls, so they are read at most once. A failed load is
retried on the next call.
+ static LiveRowsSupplier CacheLiveRows(
+ std::function<Result<std::optional<RoaringBitmap64>>()> load);
+
+ /// Creates a union of the shard readers of one index whose full-text
searches rank only live
+ /// rows. Each shard reader is wrapped in a LiveRowGlobalIndexReader. When
a search may rank
+ /// rows of a shard, live rows are loaded on the visiting thread before it
fans out to the
+ /// shards, so a load that reads manifests on `executor` never waits for
it from one of its own
+ /// threads.
+ /// @param shard_readers Readers of the shards. They take and return
global row ids.
+ /// @param shard_ranges The global row id range (both ends
inclusive) of each shard.
+ /// @param search_ranges The global row ids each shard ranks in a
full-text search, as
+ /// passed to UnionGlobalIndexReader.
+ /// @param executor Executor to search the shards in parallel,
or null to search
+ /// them one after another.
+ /// @param live_rows_supplier Supplies the live rows to full-text searches.
+ static std::shared_ptr<GlobalIndexReader> CreateUnion(
+ std::vector<std::shared_ptr<GlobalIndexReader>>&& shard_readers,
+ std::vector<Range>&& shard_ranges, std::vector<std::optional<Range>>&&
search_ranges,
+ const std::shared_ptr<Executor>& executor, const LiveRowsSupplier&
live_rows_supplier);
+
+ /// Constructs a LiveRowGlobalIndexReader.
+ /// @param wrapped The reader of the shard. It takes and
returns global row ids.
+ /// @param shard_range The global row id range (both ends
inclusive) of the shard.
+ /// @param live_rows_supplier Supplies the live rows to full-text searches.
+ LiveRowGlobalIndexReader(std::shared_ptr<GlobalIndexReader>&& wrapped,
const Range& shard_range,
+ LiveRowsSupplier live_rows_supplier);
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitIsNotNull() override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitIsNull() override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitEqual(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitNotEqual(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitLessThan(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitLessOrEqual(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitGreaterThan(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitGreaterOrEqual(const
Literal& literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitIn(
+ const std::vector<Literal>& literals) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitNotIn(
+ const std::vector<Literal>& literals) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitStartsWith(const Literal&
prefix) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitEndsWith(const Literal&
suffix) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitContains(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitLike(const Literal&
literal) override;
+
+ Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitVectorSearch(
+ const std::shared_ptr<VectorSearch>& vector_search) override;
+
+ /// Searches only the live rows of the shard that are in the include row
ids, if given. The
+ /// search is delegated unchanged when none of its include row ids is in
the shard, or when it
+ /// has none and the whole shard is live.
+ Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitFullTextSearch(
+ const std::shared_ptr<FullTextSearch>& full_text_search) override;
+
+ bool IsThreadSafe() const override {
+ return wrapped_->IsThreadSafe();
+ }
+
+ std::string GetIndexType() const override {
+ return wrapped_->GetIndexType();
+ }
+
+ private:
+ std::shared_ptr<GlobalIndexReader> wrapped_;
+ Range shard_range_;
+ LiveRowsSupplier live_rows_supplier_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/global_index/live_row_global_index_reader_test.cpp
b/src/paimon/core/global_index/live_row_global_index_reader_test.cpp
new file mode 100644
index 00000000..bc727394
--- /dev/null
+++ b/src/paimon/core/global_index/live_row_global_index_reader_test.cpp
@@ -0,0 +1,262 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/global_index/live_row_global_index_reader.h"
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <thread>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/executor.h"
+#include "paimon/predicate/full_text_search.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/testing/mock/mock_global_index_reader.h"
+#include "paimon/testing/utils/testharness.h"
+#include "paimon/utils/range.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon::test {
+class LiveRowGlobalIndexReaderTest : public ::testing::Test {
+ public:
+ static LiveRowGlobalIndexReader::LiveRowsSupplier Supply(
+ const std::optional<RoaringBitmap64>& live_rows, int32_t* call_count) {
+ std::shared_ptr<const RoaringBitmap64> shared_live_rows;
+ if (live_rows) {
+ shared_live_rows = std::make_shared<const
RoaringBitmap64>(live_rows.value());
+ }
+ return [shared_live_rows, call_count]() ->
Result<std::shared_ptr<const RoaringBitmap64>> {
+ (*call_count)++;
+ return shared_live_rows;
+ };
+ }
+
+ static RoaringBitmap64 LiveRows(int64_t from, int64_t to, const
std::vector<int64_t>& deleted) {
+ RoaringBitmap64 live_rows;
+ live_rows.AddRange(from, to + 1);
+ live_rows -= RoaringBitmap64::From(deleted);
+ return live_rows;
+ }
+
+ static std::shared_ptr<FullTextSearch> Search(
+ const std::optional<RoaringBitmap64>& include_row_ids = std::nullopt) {
+ return std::make_shared<FullTextSearch>("f0",
R"({"match":{"query":"apple"}})",
+ /*limit=*/3, include_row_ids);
+ }
+};
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestSearchUnchangedWhenEveryRowIsLive) {
+ auto mock_reader = std::make_shared<MockGlobalIndexReader>();
+ mock_reader->SetScoredResult({11, 12}, {0.5f, 0.9f});
+ int32_t call_count = 0;
+ LiveRowGlobalIndexReader reader(mock_reader, Range(10, 14),
Supply(std::nullopt, &call_count));
+ std::shared_ptr<FullTextSearch> search = Search();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> result,
+ reader.VisitFullTextSearch(search));
+ ASSERT_EQ(mock_reader->CapturedFullTextSearch(), search);
+ ASSERT_EQ(result->ToString(), "row ids: {11,12}, scores: {0.50,0.90}");
+ ASSERT_EQ(call_count, 1);
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestSearchOnlyLiveRowsOfShard) {
+ auto mock_reader = std::make_shared<MockGlobalIndexReader>();
+ mock_reader->SetScoredResult({11, 13}, {0.9f, 0.5f});
+ int32_t call_count = 0;
+ LiveRowGlobalIndexReader reader(mock_reader, Range(10, 14),
+ Supply(LiveRows(0, 20, {1, 12}),
&call_count));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> result,
+ reader.VisitFullTextSearch(Search()));
+ ASSERT_EQ(result->ToString(), "row ids: {11,13}, scores: {0.90,0.50}");
+ std::shared_ptr<FullTextSearch> captured =
mock_reader->CapturedFullTextSearch();
+ ASSERT_TRUE(captured);
+ ASSERT_EQ(captured->field_name, "f0");
+ ASSERT_EQ(captured->query, R"({"match":{"query":"apple"}})");
+ ASSERT_EQ(captured->limit, 3);
+ ASSERT_TRUE(captured->include_row_ids);
+ ASSERT_EQ(captured->include_row_ids.value(), RoaringBitmap64::From({10,
11, 13, 14}));
+
+ constexpr int64_t kMax = RoaringBitmap64::MAX_VALUE;
+ LiveRowGlobalIndexReader max_reader(mock_reader, Range(kMax - 1, kMax),
+ Supply(RoaringBitmap64::From({kMax}),
&call_count));
+ ASSERT_OK(max_reader.VisitFullTextSearch(Search()));
+ captured = mock_reader->CapturedFullTextSearch();
+ ASSERT_TRUE(captured && captured->include_row_ids);
+ ASSERT_EQ(captured->include_row_ids.value(),
RoaringBitmap64::From({kMax}));
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestIncludeRowIdsClippedToShard) {
+ auto mock_reader = std::make_shared<MockGlobalIndexReader>();
+ int32_t call_count = 0;
+ LiveRowGlobalIndexReader reader(mock_reader, Range(10, 14),
+ Supply(LiveRows(0, 20, {}), &call_count));
+ ASSERT_OK(
+ reader.VisitFullTextSearch(Search(RoaringBitmap64::From({1, 12, 20,
int64_t{1} << 32}))));
+ std::shared_ptr<FullTextSearch> captured =
mock_reader->CapturedFullTextSearch();
+ ASSERT_TRUE(captured && captured->include_row_ids);
+ ASSERT_EQ(captured->include_row_ids.value(), RoaringBitmap64::From({12}));
+
+ ASSERT_OK(reader.VisitFullTextSearch(Search(LiveRows(9, 15, {}))));
+ captured = mock_reader->CapturedFullTextSearch();
+ ASSERT_TRUE(captured);
+ ASSERT_FALSE(captured->include_row_ids);
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest,
TestCreateUnionLoadsLiveRowsBeforeSearchingShards) {
+ constexpr int64_t kSecondShardFrom = int64_t{1} << 32;
+ auto first_shard = std::make_shared<MockGlobalIndexReader>();
+ auto second_shard = std::make_shared<MockGlobalIndexReader>();
+ second_shard->SetScoredResult({kSecondShardFrom}, {0.5f});
+ auto live_rows = std::make_shared<const RoaringBitmap64>(
+ RoaringBitmap64::From({kSecondShardFrom, kSecondShardFrom + 1}));
+ std::mutex mutex;
+ std::optional<std::thread::id> loading_thread;
+ LiveRowGlobalIndexReader::LiveRowsSupplier supplier =
+ [&]() -> Result<std::shared_ptr<const RoaringBitmap64>> {
+ std::lock_guard<std::mutex> lock(mutex);
+ if (!loading_thread) {
+ loading_thread = std::this_thread::get_id();
+ }
+ return live_rows;
+ };
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Executor> executor,
+ CreateDefaultExecutor(/*thread_count=*/1));
+ std::shared_ptr<GlobalIndexReader> reader =
LiveRowGlobalIndexReader::CreateUnion(
+ std::vector<std::shared_ptr<GlobalIndexReader>>({first_shard,
second_shard}),
+ {Range(0, 1), Range(kSecondShardFrom, kSecondShardFrom + 1)},
+ {Range(0, 1), Range(kSecondShardFrom, kSecondShardFrom + 1)},
executor, supplier);
+
+ std::shared_ptr<FullTextSearch> search = Search();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> result,
+ reader->VisitFullTextSearch(search));
+ ASSERT_TRUE(loading_thread == std::this_thread::get_id());
+ std::shared_ptr<FullTextSearch> first_search =
first_shard->CapturedFullTextSearch();
+ ASSERT_TRUE(first_search && first_search->include_row_ids);
+ ASSERT_TRUE(first_search->include_row_ids->IsEmpty());
+ ASSERT_EQ(second_shard->CapturedFullTextSearch(), search);
+ ASSERT_EQ(result->ToString(), "row ids: {4294967296}, scores: {0.50}");
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest,
TestCreateUnionRanksLiveRowsOfSearchRanges) {
+ auto first_shard = std::make_shared<MockGlobalIndexReader>();
+ first_shard->SetScoredResult({0, 1}, {0.9f, 0.2f});
+ auto second_shard = std::make_shared<MockGlobalIndexReader>();
+ second_shard->SetScoredResult({10, 11}, {0.8f, 0.7f});
+ int32_t call_count = 0;
+ std::shared_ptr<GlobalIndexReader> reader =
LiveRowGlobalIndexReader::CreateUnion(
+ std::vector<std::shared_ptr<GlobalIndexReader>>({first_shard,
second_shard}),
+ {Range(0, 9), Range(5, 14)}, {Range(0, 9), Range(10, 14)},
/*executor=*/nullptr,
+ Supply(LiveRows(0, 14, {3, 12}), &call_count));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> result,
+ reader->VisitFullTextSearch(Search()));
+ ASSERT_EQ(result->ToString(), "row ids: {0,10,11}, scores:
{0.90,0.80,0.70}");
+ std::shared_ptr<FullTextSearch> first_search =
first_shard->CapturedFullTextSearch();
+ ASSERT_TRUE(first_search && first_search->include_row_ids);
+ ASSERT_EQ(first_search->include_row_ids.value(), LiveRows(0, 9, {3}));
+ std::shared_ptr<FullTextSearch> second_search =
second_shard->CapturedFullTextSearch();
+ ASSERT_TRUE(second_search && second_search->include_row_ids);
+ ASSERT_EQ(second_search->include_row_ids.value(),
RoaringBitmap64::From({10, 11, 13, 14}));
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestCacheLiveRows) {
+ int32_t load_count = 0;
+ bool fail = true;
+ LiveRowGlobalIndexReader::LiveRowsSupplier supplier =
+ LiveRowGlobalIndexReader::CacheLiveRows([&]() ->
Result<std::optional<RoaringBitmap64>> {
+ load_count++;
+ if (fail) {
+ return Status::IOError("cannot read manifest");
+ }
+ return std::optional<RoaringBitmap64>(RoaringBitmap64::From({1,
2}));
+ });
+ ASSERT_NOK_WITH_MSG(supplier(), "cannot read manifest");
+ fail = false;
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<const RoaringBitmap64> live_rows,
supplier());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<const RoaringBitmap64>
cached_live_rows, supplier());
+ ASSERT_EQ(load_count, 2);
+ ASSERT_TRUE(live_rows);
+ ASSERT_EQ(*live_rows, RoaringBitmap64::From({1, 2}));
+ ASSERT_EQ(cached_live_rows, live_rows);
+
+ LiveRowGlobalIndexReader::LiveRowsSupplier every_row_live =
+ LiveRowGlobalIndexReader::CacheLiveRows([&]() ->
Result<std::optional<RoaringBitmap64>> {
+ load_count++;
+ return std::optional<RoaringBitmap64>();
+ });
+ ASSERT_OK_AND_ASSIGN(live_rows, every_row_live());
+ ASSERT_OK_AND_ASSIGN(cached_live_rows, every_row_live());
+ ASSERT_FALSE(live_rows);
+ ASSERT_FALSE(cached_live_rows);
+ ASSERT_EQ(load_count, 3);
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest,
TestLiveRowsErrorOnlyFailsSearchesThatRankRows) {
+ auto mock_reader = std::make_shared<MockGlobalIndexReader>();
+ LiveRowGlobalIndexReader::LiveRowsSupplier failing_supplier =
+ []() -> Result<std::shared_ptr<const RoaringBitmap64>> {
+ return Status::IOError("cannot read deletion vector");
+ };
+ LiveRowGlobalIndexReader reader(mock_reader, Range(0, 9),
failing_supplier);
+ ASSERT_NOK_WITH_MSG(reader.VisitFullTextSearch(Search()), "cannot read
deletion vector");
+ ASSERT_EQ(mock_reader->InvocationCount(), 0);
+
+ std::shared_ptr<GlobalIndexReader> union_reader =
LiveRowGlobalIndexReader::CreateUnion(
+ std::vector<std::shared_ptr<GlobalIndexReader>>({mock_reader}),
{Range(0, 9)},
+ {Range(5, 9)}, /*executor=*/nullptr, failing_supplier);
+ ASSERT_NOK_WITH_MSG(union_reader->VisitFullTextSearch(Search()), "cannot
read deletion vector");
+
ASSERT_NOK_WITH_MSG(union_reader->VisitFullTextSearch(std::make_shared<FullTextSearch>(
+ "f0", R"({"match":{"query":"apple"}})",
/*limit=*/0)),
+ "full-text search requires a positive limit");
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> result,
+
union_reader->VisitFullTextSearch(Search(RoaringBitmap64())));
+ ASSERT_EQ(result->ToString(), "row ids: {}, scores: {}");
+ ASSERT_OK_AND_ASSIGN(
+ result,
union_reader->VisitFullTextSearch(Search(RoaringBitmap64::From({2, 100}))));
+ ASSERT_EQ(result->ToString(), "row ids: {}, scores: {}");
+ ASSERT_EQ(mock_reader->InvocationCount(), 0);
+
+
ASSERT_OK(reader.VisitFullTextSearch(Search(RoaringBitmap64::From({100}))));
+ std::shared_ptr<FullTextSearch> captured =
mock_reader->CapturedFullTextSearch();
+ ASSERT_TRUE(captured && captured->include_row_ids);
+ ASSERT_EQ(captured->include_row_ids.value(), RoaringBitmap64::From({100}));
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestOtherVisitsDoNotReadLiveRows) {
+ auto mock_reader = std::make_shared<MockGlobalIndexReader>();
+ mock_reader->SetDefaultResult({1, 2});
+ mock_reader->SetScoredResult({1}, {0.5f});
+ int32_t call_count = 0;
+ LiveRowGlobalIndexReader reader(mock_reader, Range(0, 9),
+ Supply(LiveRows(0, 9, {1}), &call_count));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> result,
reader.VisitEqual(Literal(1)));
+ ASSERT_EQ(result->ToString(), "{1,2}");
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult>
scored_result,
+ reader.VisitVectorSearch(nullptr));
+ ASSERT_EQ(scored_result->ToString(), "row ids: {1}, scores: {0.50}");
+ ASSERT_OK_AND_ASSIGN(scored_result, reader.VisitFullTextSearch(nullptr));
+ ASSERT_EQ(call_count, 0);
+ ASSERT_EQ(reader.GetIndexType(), "fake");
+ ASSERT_TRUE(reader.IsThreadSafe());
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/io/data_file_meta.cpp
b/src/paimon/core/io/data_file_meta.cpp
index cd866609..6ac7f543 100644
--- a/src/paimon/core/io/data_file_meta.cpp
+++ b/src/paimon/core/io/data_file_meta.cpp
@@ -20,6 +20,7 @@
#include <algorithm>
#include <cstddef>
+#include <limits>
#include <utility>
#include "arrow/type_fwd.h"
@@ -167,6 +168,17 @@ Result<int64_t> DataFileMeta::CreationTimeEpochMillis()
const {
return utc_ts.GetMillisecond();
}
+Result<Range> DataFileMeta::NonNullRowIdRange() const {
+ PAIMON_ASSIGN_OR_RAISE(int64_t start, NonNullFirstRowId());
+ if (start < 0 || row_count <= 0 ||
+ row_count - 1 > std::numeric_limits<int64_t>::max() - start) {
+ return Status::Invalid(
+ fmt::format("Invalid row id range of {}: first row id {}, row
count {}.", file_name,
+ start, row_count));
+ }
+ return Range(start, start + (row_count - 1));
+}
+
Result<std::optional<RoaringBitmap32>> DataFileMeta::ToFileSelection(
const std::optional<std::vector<Range>>& row_ranges) const {
if (!row_ranges) {
diff --git a/src/paimon/core/io/data_file_meta.h
b/src/paimon/core/io/data_file_meta.h
index ee3d4a9f..0c9409c5 100644
--- a/src/paimon/core/io/data_file_meta.h
+++ b/src/paimon/core/io/data_file_meta.h
@@ -127,6 +127,11 @@ struct DataFileMeta {
return Status::Invalid(fmt::format("First row id of {} should not be
null.", file_name));
}
+ /// Returns the global row ids of this file, as Java `nonNullRowIdRange`
does. Fails if the
+ /// first row id is null or negative, the file has no rows, or its last
row id would pass the
+ /// largest int64 value.
+ Result<Range> NonNullRowIdRange() const;
+
// empty row_ranges indicates all rows in the file are needed, return null
bitmap
Result<std::optional<RoaringBitmap32>> ToFileSelection(
const std::optional<std::vector<Range>>& row_ranges) const;
diff --git a/src/paimon/core/io/data_file_meta_test.cpp
b/src/paimon/core/io/data_file_meta_test.cpp
index f76bc0ae..b35631e4 100644
--- a/src/paimon/core/io/data_file_meta_test.cpp
+++ b/src/paimon/core/io/data_file_meta_test.cpp
@@ -18,6 +18,8 @@
#include "paimon/core/io/data_file_meta.h"
+#include <limits>
+
#include "gtest/gtest.h"
#include "paimon/status.h"
#include "paimon/testing/utils/binary_row_generator.h"
@@ -180,6 +182,34 @@ TEST(DataFileMetaTest, TestNonNullFirstRowId) {
}
}
+TEST(DataFileMetaTest, TestNonNullRowIdRange) {
+ auto create_file_meta = [](int64_t row_count, const
std::optional<int64_t>& first_row_id) {
+ return std::make_shared<DataFileMeta>(
+ "data-0.orc", /*file_size=*/645, row_count, BinaryRow::EmptyRow(),
+ BinaryRow::EmptyRow(), SimpleStats::EmptyStats(),
SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/0,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, first_row_id,
/*write_cols=*/std::nullopt,
+ /*column_max_sequence_numbers=*/std::nullopt);
+ };
+ constexpr int64_t kMax = std::numeric_limits<int64_t>::max();
+ ASSERT_OK_AND_ASSIGN(Range range, create_file_meta(10,
100)->NonNullRowIdRange());
+ ASSERT_EQ(range, Range(100, 109));
+ ASSERT_OK_AND_ASSIGN(range, create_file_meta(2, kMax -
1)->NonNullRowIdRange());
+ ASSERT_EQ(range, Range(kMax - 1, kMax));
+ ASSERT_NOK_WITH_MSG(create_file_meta(3, kMax - 1)->NonNullRowIdRange(),
+ "Invalid row id range of data-0.orc");
+ ASSERT_NOK_WITH_MSG(create_file_meta(0, 100)->NonNullRowIdRange(),
+ "Invalid row id range of data-0.orc");
+ ASSERT_NOK_WITH_MSG(create_file_meta(10, -1)->NonNullRowIdRange(),
+ "Invalid row id range of data-0.orc");
+ ASSERT_NOK_WITH_MSG(create_file_meta(10,
std::nullopt)->NonNullRowIdRange(),
+ "First row id of data-0.orc should not be null.");
+}
+
TEST(DataFileMetaTest, TestToFileSelection) {
// row id range [100, 110)
auto file_meta = std::make_shared<DataFileMeta>(
diff --git a/src/paimon/core/operation/file_store_scan.h
b/src/paimon/core/operation/file_store_scan.h
index b73ad095..25911d0e 100644
--- a/src/paimon/core/operation/file_store_scan.h
+++ b/src/paimon/core/operation/file_store_scan.h
@@ -130,6 +130,15 @@ class FileStoreScan {
return this;
}
+ /// Replaces the partition filter taken from the scan filter; a null
filter accepts every
+ /// partition. `partition_schema` holds the partition fields the filter is
indexed by.
+ FileStoreScan* WithPartitionFilter(const std::shared_ptr<PredicateFilter>&
partition_filter,
+ const std::shared_ptr<arrow::Schema>&
partition_schema) {
+ partition_filter_ = partition_filter;
+ partition_schema_ = partition_schema;
+ return this;
+ }
+
FileStoreScan* WithTablePath(const std::string& table_path) {
table_path_ = table_path;
return this;
diff --git a/src/paimon/core/table/source/data_evolution_split_generator.cpp
b/src/paimon/core/table/source/data_evolution_split_generator.cpp
index 0a76e75a..62f334db 100644
--- a/src/paimon/core/table/source/data_evolution_split_generator.cpp
+++ b/src/paimon/core/table/source/data_evolution_split_generator.cpp
@@ -28,6 +28,7 @@
#include "paimon/common/utils/bin_packing.h"
#include "paimon/common/utils/range_helper.h"
#include "paimon/core/io/data_file_meta.h"
+#include "paimon/utils/range.h"
namespace paimon {
Result<std::vector<SplitGenerator::SplitGroup>>
DataEvolutionSplitGenerator::SplitForBatch(
@@ -37,8 +38,8 @@ Result<std::vector<SplitGenerator::SplitGroup>>
DataEvolutionSplitGenerator::Spl
return meta->NonNullFirstRowId();
},
[](const std::shared_ptr<DataFileMeta>& meta) -> Result<int64_t> {
- PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id,
meta->NonNullFirstRowId());
- return first_row_id + meta->row_count - 1;
+ PAIMON_ASSIGN_OR_RAISE(Range row_id_range,
meta->NonNullRowIdRange());
+ return row_id_range.to;
});
PAIMON_ASSIGN_OR_RAISE(std::vector<std::vector<std::shared_ptr<DataFileMeta>>>
ranges,
diff --git a/src/paimon/core/table/source/split_generator_test.cpp
b/src/paimon/core/table/source/split_generator_test.cpp
index 9f136870..4ba05c3d 100644
--- a/src/paimon/core/table/source/split_generator_test.cpp
+++ b/src/paimon/core/table/source/split_generator_test.cpp
@@ -21,6 +21,7 @@
#include <algorithm>
#include <cstdint>
+#include <limits>
#include <memory>
#include <optional>
#include <string>
@@ -260,6 +261,22 @@ TEST_F(SplitGeneratorTest,
TestDataEvolutionBlobSplitByFileSize) {
}
}
+TEST_F(SplitGeneratorTest, TestDataEvolutionRowIdRangeAtInt64Max) {
+ constexpr int64_t kMax = std::numeric_limits<int64_t>::max();
+ DataEvolutionSplitGenerator split_generator(/*target_split_size=*/1000,
/*open_file_cost=*/10,
+ /*count_blob_size=*/true);
+ ASSERT_OK_AND_ASSIGN(std::vector<SplitGenerator::SplitGroup> split_groups,
+ split_generator.SplitForBatch(
+ {CreateDataFileMetaWithRowId("f1",
/*file_size=*/100, /*row_count=*/2,
+
/*first_row_id=*/kMax - 1),
+ CreateDataFileMetaWithRowId("f2",
/*file_size=*/100, /*row_count=*/1,
+
/*first_row_id=*/kMax)}));
+ CheckResult(split_groups, {{"f1", "f2"}}, {false});
+
ASSERT_NOK_WITH_MSG(split_generator.SplitForBatch({CreateDataFileMetaWithRowId(
+ "f3", /*file_size=*/100, /*row_count=*/3,
/*first_row_id=*/kMax - 1)}),
+ "Invalid row id range of f3");
+}
+
TEST_F(SplitGeneratorTest, TestMergeTree) {
std::vector<std::shared_ptr<DataFileMeta>> files = {
CreateDataFileMeta("1", 0, 10), CreateDataFileMeta("2", 0, 12),
diff --git a/test/inte/data_evolution_table_test.cpp
b/test/inte/data_evolution_table_test.cpp
index c508fd19..2924e541 100644
--- a/test/inte/data_evolution_table_test.cpp
+++ b/test/inte/data_evolution_table_test.cpp
@@ -32,9 +32,16 @@
#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/scope_guard.h"
#include "paimon/common/utils/vector_store_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/global_index/global_index_live_row_filter.h"
#include "paimon/core/global_index/indexed_split_impl.h"
+#include "paimon/core/operation/file_store_scan.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/snapshot.h"
#include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/core/utils/snapshot_manager.h"
#include "paimon/defs.h"
+#include "paimon/executor.h"
#include "paimon/format/mosaic/mosaic_format_defs.h"
#include "paimon/fs/file_system.h"
#include "paimon/global_index/bitmap_global_index_result.h"
@@ -163,8 +170,8 @@ class DataEvolutionTableTest : public ::testing::Test,
/// created with, which a test that later evolves the schema has to pass
along.
/// `extra_options` is merged in for tests that also pin a read batch or
split target size.
std::map<std::string, std::string> CreateDataEvolutionTable(
- bool deletion_vectors_enabled,
- const std::map<std::string, std::string>& extra_options = {}) const {
+ bool deletion_vectors_enabled, const std::map<std::string,
std::string>& extra_options = {},
+ const std::vector<std::string>& partition_keys = {}) const {
std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
FileFormat()},
{Options::FILE_SYSTEM,
"local"},
{Options::ROW_TRACKING_ENABLED, "true"},
@@ -175,7 +182,7 @@ class DataEvolutionTableTest : public ::testing::Test,
std::get<2>(GetParam()) ? "true" : "false");
}
options.insert(extra_options.begin(), extra_options.end());
- CreateTable(/*partition_keys=*/{}, options);
+ CreateTable(partition_keys, options);
return options;
}
@@ -3386,6 +3393,72 @@ TEST_P(DataEvolutionDvTableTest,
TestGlobalIndexScoresWithMergedFileFiltering) {
CheckGlobalIndexScoresWithFiltering(/*merge_files=*/true);
}
+TEST_P(DataEvolutionDvTableTest, TestGlobalIndexLiveRows) {
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true,
/*extra_options=*/{},
+ /*partition_keys=*/{"f1"});
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<std::shared_ptr<CommitMessage>> group_msgs0,
+ WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0,
1, 2, 3},
+ /*partition=*/{{"f1", "p0"}}));
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<std::shared_ptr<CommitMessage>> group_msgs1,
+ WriteAndCommitGroup(table_path, /*first_row_id=*/4, /*f0_values=*/{4,
5, 6, 7},
+ /*partition=*/{{"f1", "p1"}}));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
+ PlannedAnchorFileNames(table_path));
+ ASSERT_EQ(anchor_file_names.size(), 2);
+
+ std::shared_ptr<FileSystem> fs = dir_->GetFileSystem();
+ SchemaManager schema_manager(fs, table_path);
+ ASSERT_OK_AND_ASSIGN(std::optional<std::shared_ptr<TableSchema>>
table_schema,
+ schema_manager.Latest());
+ ASSERT_TRUE(table_schema);
+ ASSERT_OK_AND_ASSIGN(CoreOptions options,
+ CoreOptions::FromMap(table_schema.value()->Options(),
fs));
+ SnapshotManager snapshot_manager(fs, table_path);
+ ASSERT_OK_AND_ASSIGN(std::optional<Snapshot> snapshot_before_deletion,
+ snapshot_manager.LatestSnapshot());
+ ASSERT_TRUE(snapshot_before_deletion);
+
+ ASSERT_OK(CommitDeletionVectors(table_path, group_msgs0[0],
+ {{anchor_file_names[0],
/*deleted_positions=*/{1, 3}}}));
+ ASSERT_OK(CommitDeletionVectors(table_path, group_msgs1[0],
+ {{anchor_file_names[1],
/*deleted_positions=*/{0}}}));
+ ASSERT_OK_AND_ASSIGN(std::optional<Snapshot> snapshot,
snapshot_manager.LatestSnapshot());
+ ASSERT_TRUE(snapshot);
+
+ std::shared_ptr<Executor> executor = CreateDefaultExecutor();
+ auto read_live_rows = [&](const Snapshot& read_snapshot,
+ const std::shared_ptr<PredicateFilter>&
partition_filter) {
+ return GlobalIndexLiveRowFilter::LiveRows(table_path,
table_schema.value(), read_snapshot,
+ partition_filter, options,
executor,
+ GetDefaultPool());
+ };
+ ASSERT_OK_AND_ASSIGN(std::optional<RoaringBitmap64> live_rows,
+ read_live_rows(snapshot.value(),
/*partition_filter=*/nullptr));
+ ASSERT_TRUE(live_rows);
+ ASSERT_EQ(live_rows.value(), RoaringBitmap64::From({0, 2, 5, 6, 7}));
+
+ auto arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema.value()->Fields());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<PredicateFilter> p0_filter,
+ FileStoreScan::CreatePartitionPredicate(
+ table_schema.value()->PartitionKeys(),
+ options.GetPartitionDefaultName(), arrow_schema,
{{{"f1", "p0"}}}));
+ ASSERT_OK_AND_ASSIGN(live_rows, read_live_rows(snapshot.value(),
p0_filter));
+ ASSERT_TRUE(live_rows);
+ ASSERT_EQ(live_rows.value(), RoaringBitmap64::From({0, 2}));
+
+ ASSERT_OK_AND_ASSIGN(live_rows,
read_live_rows(snapshot_before_deletion.value(),
+
/*partition_filter=*/nullptr));
+ ASSERT_TRUE(live_rows);
+ ASSERT_EQ(live_rows.value(), RoaringBitmap64::From({0, 1, 2, 3, 4, 5, 6,
7}));
+
+ ASSERT_OK(fs->Delete(PathUtil::JoinPath(table_path, "index")));
+ ASSERT_NOK_WITH_MSG(read_live_rows(snapshot.value(),
/*partition_filter=*/nullptr),
+ "failed to read deletion vector of data file");
+}
+
TEST_P(DataEvolutionDvTableTest, TestReadWithDeletionVectors) {
CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp
index 9c36766b..97b16982 100644
--- a/test/inte/global_index_test.cpp
+++ b/test/inte/global_index_test.cpp
@@ -47,6 +47,7 @@
#include "paimon/status.h"
#include "paimon/testing/utils/binary_row_generator.h"
#include "paimon/testing/utils/counting_cache_test_utils.h"
+#include "paimon/testing/utils/deletion_vector_test_helper.h"
#include "paimon/testing/utils/io_exception_helper.h"
#include "paimon/testing/utils/test_helper.h"
#include "paimon/testing/utils/testharness.h"
@@ -283,6 +284,113 @@ class GlobalIndexTest : public ::testing::Test, public
::testing::WithParamInter
CheckFullTextTopK(readers[0], query, std::nullopt, {3});
}
+ void CheckSearchSkipsDeletedRows(const std::string& index_type,
+ const std::map<std::string, std::string>&
index_options) {
+ arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
+ arrow::field("f1", arrow::int32())};
+ auto schema = arrow::schema(fields);
+ std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
file_format_},
+ {Options::FILE_SYSTEM,
"local"},
+
{Options::ROW_TRACKING_ENABLED, "true"},
+
{Options::DATA_EVOLUTION_ENABLED, "true"},
+
{Options::DELETION_VECTORS_ENABLED, "true"}};
+ CreateTable(/*partition_keys=*/{}, schema, options);
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+ auto src_array1 =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+["apple apple apple apple apple apple apple apple apple apple", 0],
+["apple apple apple apple apple apple apple apple apple", 1],
+["apple apple apple apple apple apple apple apple", 2],
+["apple apple apple apple apple apple apple", 3],
+["apple apple apple apple apple apple", 4]
+ ])")
+ .ValueOrDie();
+ auto src_array2 =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+["apple apple apple apple apple", 5],
+["apple apple apple apple", 6],
+["apple apple apple", 7],
+["apple apple", 8],
+["apple", 9]
+ ])")
+ .ValueOrDie();
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
+ WriteArray(table_path, schema->field_names(),
src_array1));
+ ASSERT_OK(Commit(table_path, commit_msgs1));
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs2,
+ WriteArray(table_path, schema->field_names(),
src_array2));
+ ASSERT_OK(Commit(table_path, commit_msgs2));
+ ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0",
index_type, index_options,
+ Range(0, 4)));
+ ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0",
index_type, index_options,
+ Range(5, 9)));
+
+ auto create_scan = [&]() {
+ return GlobalIndexScan::Create(table_path,
/*snapshot_id=*/std::nullopt,
+ /*partitions=*/std::nullopt,
/*options=*/{}, fs_,
+ /*executor=*/nullptr, pool_);
+ };
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexScan>
scan_before_deletion, create_scan());
+
+ ASSERT_OK_AND_ASSIGN(auto split, ScanData(table_path,
/*partition_filters=*/{}));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
+
DeletionVectorTestHelper::RetrieveAnchorFileNames(split->DataFiles()));
+ ASSERT_EQ(anchor_file_names.size(), 2u);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<CommitMessage> dv_commit_msg,
+
DeletionVectorTestHelper::CreateDeletionVectorCommitMessage(
+ dir_->GetFileSystem(), table_path,
file_format_, commit_msgs1[0],
+ {{anchor_file_names[0],
/*deleted_positions=*/{0, 1, 2, 3}},
+ {anchor_file_names[1],
/*deleted_positions=*/{0}}},
+ pool_));
+ ASSERT_OK(Commit(table_path, {dv_commit_msg}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexScan> scan,
create_scan());
+
+ std::string query = R"({"match":{"query":"apple"}})";
+ ASSERT_OK_AND_ASSIGN(
+ auto readers_before_deletion,
+ scan_before_deletion->CreateReaders("f0",
/*row_range_index=*/std::nullopt));
+ ASSERT_EQ(readers_before_deletion.size(), 1u);
+ CheckFullTextTopK(readers_before_deletion[0], query, std::nullopt,
+ {0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
+ ASSERT_OK_AND_ASSIGN(
+ ScoredRows rows_before_deletion,
+ SearchFullText(readers_before_deletion[0], query, /*limit=*/100,
std::nullopt));
+ RoaringBitmap64 deleted_rows = RoaringBitmap64::From({0, 1, 2, 3, 5});
+ ScoredRows expected_live_rows;
+ for (const auto& row : rows_before_deletion) {
+ if (!deleted_rows.Contains(row.first)) {
+ expected_live_rows.push_back(row);
+ }
+ }
+
+ ASSERT_OK_AND_ASSIGN(auto readers,
+ scan->CreateReaders("f0",
/*row_range_index=*/std::nullopt));
+ ASSERT_EQ(readers.size(), 1u);
+ ASSERT_OK_AND_ASSIGN(ScoredRows live_rows,
+ SearchFullText(readers[0], query, /*limit=*/100,
std::nullopt));
+ ASSERT_EQ(live_rows, expected_live_rows);
+ CheckFullTextTopK(readers[0], query, std::nullopt, {4, 6, 7, 8, 9});
+ CheckFullTextTopK(readers[0], query, RoaringBitmap64::From({0, 5, 8}),
{8});
+ CheckFullTextTopK(readers[0], query, RoaringBitmap64::From({0, 1}),
{});
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult>
index_result,
+ readers[0]->VisitFullTextSearch(
+ std::make_shared<FullTextSearch>("f0", query,
/*limit=*/5)));
+ ASSERT_OK_AND_ASSIGN(auto plan, ScanGlobalIndexAndData(table_path,
/*predicate=*/nullptr,
+ /*options=*/{},
index_result));
+ auto expected_array =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_({SpecialFields::ValueKind().ArrowField(),
fields[1]}),
+ R"([
+[0, 4],
+[0, 6],
+[0, 7],
+[0, 8],
+[0, 9]
+ ])")
+ .ValueOrDie();
+ ASSERT_OK(ReadData(table_path, {"f1"}, expected_array,
/*predicate=*/nullptr, plan));
+ }
+
Result<std::shared_ptr<Plan>> ScanGlobalIndexAndData(
const std::string& table_path, const std::shared_ptr<Predicate>&
predicate,
const std::map<std::string, std::string>& options = {},
@@ -2924,6 +3032,14 @@ TEST_P(GlobalIndexTest,
TestLuceneSearchOverlappingIndexRanges) {
{"lucene-fts.write.tmp.directory",
tmp_dir->Str()}},
R"({"match":{"query":"document","operator":"And"}})");
}
+
+TEST_P(GlobalIndexTest, TestLuceneSearchSkipsRowsDeletedByDeletionVectors) {
+ auto tmp_dir = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(tmp_dir);
+ CheckSearchSkipsDeletedRows("lucene-fts",
+
{{"lucene-fts.write.omit-term-freq-and-position", "false"},
+ {"lucene-fts.write.tmp.directory",
tmp_dir->Str()}});
+}
#endif
#ifdef PAIMON_ENABLE_FULL_TEXT
@@ -3071,6 +3187,10 @@ TEST_P(GlobalIndexTest,
TestFullTextSearchOverlappingIndexRanges) {
CheckSearchOverlappingIndexRanges("full-text", {{"full-text.tokenizer",
"default"}},
R"({"match":{"query":"document"}})");
}
+
+TEST_P(GlobalIndexTest, TestFullTextSearchSkipsRowsDeletedByDeletionVectors) {
+ CheckSearchSkipsDeletedRows("full-text", {{"full-text.tokenizer",
"default"}});
+}
#endif
TEST_P(GlobalIndexTest, TestBTreeWriteCommitScanReadIndex) {
