From 73e5b6a536f36e9a6ada3150961dcad6739bcde3 Mon Sep 17 00:00:00 2001 From: Nicholas Jiang Date: Thu, 15 Oct 2026 09:44:09 -0700 Subject: [PATCH] fix(global-index): exclude deleted rows from full-text ranking Deletion vectors were applied only when rows were read, so a full-text search on a table with deletion vectors still ranked deleted rows. They could take every top-k slot, and the read then dropped them, returning fewer than limit rows even when enough live rows matched. As Java GlobalIndexLiveRowFilter does, GlobalIndexScan now computes the rows live in the snapshot it reads the index from: the union of the row id ranges of the data files in the scanned partitions, minus the rows deleted by their deletion vectors. Every range is added before any deletion vector is subtracted, so a deleted row is not added back by another file covering it. DataFileMeta::NonNullRowIdRange computes each range as Java nonNullRowIdRange does, and rejects one that would pass the largest int64 value. As in Java, DataEvolutionSplitGenerator takes its ranges from it too, so every data-evolution scan, including the one live rows are read from, rejects such a range instead of overflowing. A deletion vector that cannot be read is reported with its data file, snapshot and table. Live rows are read only for tables with deletion vectors and row tracking, once per scan on its first full-text search that ranks rows. A search without a positive limit, or whose include row ids miss the rows every shard ranks, neither reads them nor fails on them. As Java DataEvolutionFullTextRead#includeRowIds does, each index shard then ranks only the live rows of its search range, intersected with the include row ids if given, before the shard results are merged into the global top-k. BM25 statistics still come from the whole index, and a shard without deleted rows gets no extra filter. Live rows are loaded before the search fans out to the shards, so loading them never waits for the executor from one of its own threads. Predicate and vector searches are delegated unchanged. --- docs/source/user_guide/global_index.rst | 25 ++ src/paimon/CMakeLists.txt | 3 + .../global_index_live_row_filter.cpp | 213 ++++++++++++++ .../global_index_live_row_filter.h | 63 +++++ .../global_index/global_index_scan_impl.cpp | 38 ++- .../global_index/global_index_scan_impl.h | 7 +- .../live_row_global_index_reader.cpp | 232 ++++++++++++++++ .../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 +- .../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(-) create mode 100644 src/paimon/core/global_index/global_index_live_row_filter.cpp create mode 100644 src/paimon/core/global_index/global_index_live_row_filter.h create mode 100644 src/paimon/core/global_index/live_row_global_index_reader.cpp create mode 100644 src/paimon/core/global_index/live_row_global_index_reader.h create mode 100644 src/paimon/core/global_index/live_row_global_index_reader_test.cpp diff --git a/docs/source/user_guide/global_index.rst b/docs/source/user_guide/global_index.rst index afee33914..f3dca7445 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 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 reader, + index_scan->CreateReader("f0", "full-text", + /*row_range_index=*/std::nullopt)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr result, + reader->VisitFullTextSearch(std::make_shared( + "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 da5213dc0..b3d1a1e24 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 000000000..faf9e0408 --- /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 +#include +#include +#include +#include +#include +#include + +#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> CreateSnapshotReader( + const std::string& root_path, const std::shared_ptr& table_schema, + const std::shared_ptr& partition_filter, const CoreOptions& options, + const std::shared_ptr& executor, const std::shared_ptr& pool) { + std::shared_ptr fs = options.GetFileSystem(); + std::string branch = BranchManager::NormalizeBranch(options.GetBranch()); + auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields()); + PAIMON_ASSIGN_OR_RAISE(std::vector external_paths, options.CreateExternalPaths()); + PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, + options.CreateGlobalIndexExternalPath()); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr 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 manifest_list, + ManifestList::Create(fs, options.GetManifestFormat(), options.GetManifestCompression(), + path_factory, options.GetCache(), pool)); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr partition_schema, + FieldMapping::GetPartitionSchema(arrow_schema, table_schema->PartitionKeys())); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr manifest_file, + ManifestFile::Create(fs, options.GetManifestFormat(), options.GetManifestCompression(), + path_factory, options.GetManifestTargetFileSize(), pool, options, + partition_schema)); + auto snapshot_manager = std::make_shared(fs, root_path, branch); + auto schema_manager = std::make_shared(fs, root_path, branch); + auto scan_filter = std::make_shared( + /*predicate=*/nullptr, + /*partition_filters=*/std::vector>(), + /*bucket_filter=*/std::nullopt); + + std::unique_ptr scan; + std::unique_ptr 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( + 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( + options.GetSourceSplitTargetSize(), options.GetSourceSplitOpenFileCost(), + BucketMode::BUCKET_UNAWARE); + } + scan->WithTablePath(root_path)->WithPartitionFilter(partition_filter, partition_schema); + + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr index_manifest_file, + IndexManifestFile::Create(fs, options.GetManifestFormat(), options.GetManifestCompression(), + path_factory, options.GetBucket(), pool, options)); + auto index_file_handler = std::make_unique( + fs, std::move(index_manifest_file), std::make_shared(path_factory), + options.DeletionVectorsBitmap64(), pool); + return std::make_unique( + 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& fs, + const std::shared_ptr& pool, RoaringBitmap64* live_rows) { + const std::vector>& 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> 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 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> GlobalIndexLiveRowFilter::LiveRows( + const std::string& root_path, const std::shared_ptr& table_schema, + const Snapshot& snapshot, const std::shared_ptr& partition_filter, + const CoreOptions& options, const std::shared_ptr& executor, + const std::shared_ptr& 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(); + } + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr 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, snapshot_reader->Read()); + + std::vector> splits; + splits.reserve(plan->Splits().size()); + for (const auto& split : plan->Splits()) { + auto data_split = std::dynamic_pointer_cast(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(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 000000000..4f607e699 --- /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 +#include +#include + +#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> LiveRows( + const std::string& root_path, const std::shared_ptr& table_schema, + const Snapshot& snapshot, const std::shared_ptr& partition_filter, + const CoreOptions& options, const std::shared_ptr& executor, + const std::shared_ptr& 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 511be25c3..b20437d4a 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& table_schema, - const CoreOptions& options, - const std::shared_ptr& path_factory, - IndexMetaMap&& index_metas, - const std::shared_ptr& executor, - const std::shared_ptr& pool) +GlobalIndexScanImpl::GlobalIndexScanImpl( + const std::shared_ptr& table_schema, const CoreOptions& options, + const std::shared_ptr& path_factory, IndexMetaMap&& index_metas, + LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier, + const std::shared_ptr& executor, const std::shared_ptr& pool) : pool_(pool), table_schema_(table_schema), options_(options), index_file_manager_(std::make_shared( 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> GlobalIndexScanImpl::Create( @@ -122,8 +123,18 @@ Result> GlobalIndexScanImpl::Create( index_metas[index_meta->index_field_id][index_file_meta->IndexType()][range].push_back( index_file_meta); } - return std::unique_ptr(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( + new GlobalIndexScanImpl(table_schema, options, path_factory, std::move(index_metas), + std::move(live_rows_supplier), final_executor, pool)); } Result> GlobalIndexScanImpl::GetOrCreateIndexEvaluator() { @@ -235,8 +246,15 @@ Result>> GlobalIndexScanImpl::Cre if (union_readers.empty()) { continue; } - readers.push_back(std::make_shared( - 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( + 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 a7921d77c..bb26ae8ee 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& table_schema, const CoreOptions& options, const std::shared_ptr& path_factory, - IndexMetaMap&& index_metas, const std::shared_ptr& executor, + IndexMetaMap&& index_metas, + LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier, + const std::shared_ptr& executor, const std::shared_ptr& pool); Result> GetOrCreateIndexEvaluator(); @@ -88,6 +91,8 @@ class GlobalIndexScanImpl : public GlobalIndexScan { CoreOptions options_; std::shared_ptr 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_; std::shared_ptr 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 000000000..9bdf1449b --- /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 +#include +#include +#include + +#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>&& readers, + std::vector&& row_ranges, + std::vector>&& search_ranges, + std::vector&& ranked_ranges, + const std::shared_ptr& 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> VisitFullTextSearch( + const std::shared_ptr& 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 ranked_ranges_; + LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier_; +}; +} // namespace + +LiveRowGlobalIndexReader::LiveRowsSupplier LiveRowGlobalIndexReader::CacheLiveRows( + std::function>()> load) { + struct State { + std::mutex mutex; + bool loaded = false; + std::shared_ptr live_rows; + }; + auto state = std::make_shared(); + return [state, load = std::move(load)]() -> Result> { + std::lock_guard lock(state->mutex); + if (!state->loaded) { + PAIMON_ASSIGN_OR_RAISE(std::optional live_rows, load()); + if (live_rows) { + state->live_rows = + std::make_shared(std::move(live_rows).value()); + } + state->loaded = true; + } + return state->live_rows; + }; +} + +std::shared_ptr LiveRowGlobalIndexReader::CreateUnion( + std::vector>&& shard_readers, + std::vector&& shard_ranges, std::vector>&& search_ranges, + const std::shared_ptr& executor, const LiveRowsSupplier& live_rows_supplier) { + assert(shard_readers.size() == shard_ranges.size()); + std::vector> readers; + std::vector 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( + 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( + std::move(readers), std::move(shard_ranges), std::move(search_ranges), + std::move(ranked_ranges), executor, live_rows_supplier); +} + +LiveRowGlobalIndexReader::LiveRowGlobalIndexReader(std::shared_ptr&& 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> LiveRowGlobalIndexReader::VisitIsNotNull() { + return wrapped_->VisitIsNotNull(); +} + +Result> LiveRowGlobalIndexReader::VisitIsNull() { + return wrapped_->VisitIsNull(); +} + +Result> LiveRowGlobalIndexReader::VisitEqual( + const Literal& literal) { + return wrapped_->VisitEqual(literal); +} + +Result> LiveRowGlobalIndexReader::VisitNotEqual( + const Literal& literal) { + return wrapped_->VisitNotEqual(literal); +} + +Result> LiveRowGlobalIndexReader::VisitLessThan( + const Literal& literal) { + return wrapped_->VisitLessThan(literal); +} + +Result> LiveRowGlobalIndexReader::VisitLessOrEqual( + const Literal& literal) { + return wrapped_->VisitLessOrEqual(literal); +} + +Result> LiveRowGlobalIndexReader::VisitGreaterThan( + const Literal& literal) { + return wrapped_->VisitGreaterThan(literal); +} + +Result> LiveRowGlobalIndexReader::VisitGreaterOrEqual( + const Literal& literal) { + return wrapped_->VisitGreaterOrEqual(literal); +} + +Result> LiveRowGlobalIndexReader::VisitIn( + const std::vector& literals) { + return wrapped_->VisitIn(literals); +} + +Result> LiveRowGlobalIndexReader::VisitNotIn( + const std::vector& literals) { + return wrapped_->VisitNotIn(literals); +} + +Result> LiveRowGlobalIndexReader::VisitStartsWith( + const Literal& prefix) { + return wrapped_->VisitStartsWith(prefix); +} + +Result> LiveRowGlobalIndexReader::VisitEndsWith( + const Literal& suffix) { + return wrapped_->VisitEndsWith(suffix); +} + +Result> LiveRowGlobalIndexReader::VisitContains( + const Literal& literal) { + return wrapped_->VisitContains(literal); +} + +Result> LiveRowGlobalIndexReader::VisitLike( + const Literal& literal) { + return wrapped_->VisitLike(literal); +} + +Result> LiveRowGlobalIndexReader::VisitVectorSearch( + const std::shared_ptr& vector_search) { + return wrapped_->VisitVectorSearch(vector_search); +} + +Result> LiveRowGlobalIndexReader::VisitFullTextSearch( + const std::shared_ptr& 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 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( + full_text_search->field_name, full_text_search->query, full_text_search->limit)); + } + return wrapped_->VisitFullTextSearch( + std::make_shared(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 000000000..4e2589c7e --- /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 +#include +#include +#include +#include + +#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>()>; + + /// 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>()> 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 CreateUnion( + std::vector>&& shard_readers, + std::vector&& shard_ranges, std::vector>&& search_ranges, + const std::shared_ptr& 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&& wrapped, const Range& shard_range, + LiveRowsSupplier live_rows_supplier); + + Result> VisitIsNotNull() override; + Result> VisitIsNull() override; + Result> VisitEqual(const Literal& literal) override; + Result> VisitNotEqual(const Literal& literal) override; + Result> VisitLessThan(const Literal& literal) override; + Result> VisitLessOrEqual(const Literal& literal) override; + Result> VisitGreaterThan(const Literal& literal) override; + Result> VisitGreaterOrEqual(const Literal& literal) override; + Result> VisitIn( + const std::vector& literals) override; + Result> VisitNotIn( + const std::vector& literals) override; + Result> VisitStartsWith(const Literal& prefix) override; + Result> VisitEndsWith(const Literal& suffix) override; + Result> VisitContains(const Literal& literal) override; + Result> VisitLike(const Literal& literal) override; + + Result> VisitVectorSearch( + const std::shared_ptr& 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> VisitFullTextSearch( + const std::shared_ptr& full_text_search) override; + + bool IsThreadSafe() const override { + return wrapped_->IsThreadSafe(); + } + + std::string GetIndexType() const override { + return wrapped_->GetIndexType(); + } + + private: + std::shared_ptr 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 000000000..bc727394c --- /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 +#include +#include +#include +#include +#include + +#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& live_rows, int32_t* call_count) { + std::shared_ptr shared_live_rows; + if (live_rows) { + shared_live_rows = std::make_shared(live_rows.value()); + } + return [shared_live_rows, call_count]() -> Result> { + (*call_count)++; + return shared_live_rows; + }; + } + + static RoaringBitmap64 LiveRows(int64_t from, int64_t to, const std::vector& deleted) { + RoaringBitmap64 live_rows; + live_rows.AddRange(from, to + 1); + live_rows -= RoaringBitmap64::From(deleted); + return live_rows; + } + + static std::shared_ptr Search( + const std::optional& include_row_ids = std::nullopt) { + return std::make_shared("f0", R"({"match":{"query":"apple"}})", + /*limit=*/3, include_row_ids); + } +}; + +TEST_F(LiveRowGlobalIndexReaderTest, TestSearchUnchangedWhenEveryRowIsLive) { + auto mock_reader = std::make_shared(); + 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 search = Search(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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(); + 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 result, + reader.VisitFullTextSearch(Search())); + ASSERT_EQ(result->ToString(), "row ids: {11,13}, scores: {0.90,0.50}"); + std::shared_ptr 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(); + 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 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(); + auto second_shard = std::make_shared(); + second_shard->SetScoredResult({kSecondShardFrom}, {0.5f}); + auto live_rows = std::make_shared( + RoaringBitmap64::From({kSecondShardFrom, kSecondShardFrom + 1})); + std::mutex mutex; + std::optional loading_thread; + LiveRowGlobalIndexReader::LiveRowsSupplier supplier = + [&]() -> Result> { + std::lock_guard lock(mutex); + if (!loading_thread) { + loading_thread = std::this_thread::get_id(); + } + return live_rows; + }; + ASSERT_OK_AND_ASSIGN(std::shared_ptr executor, + CreateDefaultExecutor(/*thread_count=*/1)); + std::shared_ptr reader = LiveRowGlobalIndexReader::CreateUnion( + std::vector>({first_shard, second_shard}), + {Range(0, 1), Range(kSecondShardFrom, kSecondShardFrom + 1)}, + {Range(0, 1), Range(kSecondShardFrom, kSecondShardFrom + 1)}, executor, supplier); + + std::shared_ptr search = Search(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr result, + reader->VisitFullTextSearch(search)); + ASSERT_TRUE(loading_thread == std::this_thread::get_id()); + std::shared_ptr 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(); + first_shard->SetScoredResult({0, 1}, {0.9f, 0.2f}); + auto second_shard = std::make_shared(); + second_shard->SetScoredResult({10, 11}, {0.8f, 0.7f}); + int32_t call_count = 0; + std::shared_ptr reader = LiveRowGlobalIndexReader::CreateUnion( + std::vector>({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 result, + reader->VisitFullTextSearch(Search())); + ASSERT_EQ(result->ToString(), "row ids: {0,10,11}, scores: {0.90,0.80,0.70}"); + std::shared_ptr 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 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> { + load_count++; + if (fail) { + return Status::IOError("cannot read manifest"); + } + return std::optional(RoaringBitmap64::From({1, 2})); + }); + ASSERT_NOK_WITH_MSG(supplier(), "cannot read manifest"); + fail = false; + ASSERT_OK_AND_ASSIGN(std::shared_ptr live_rows, supplier()); + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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> { + load_count++; + return std::optional(); + }); + 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(); + LiveRowGlobalIndexReader::LiveRowsSupplier failing_supplier = + []() -> Result> { + 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 union_reader = LiveRowGlobalIndexReader::CreateUnion( + std::vector>({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( + "f0", R"({"match":{"query":"apple"}})", /*limit=*/0)), + "full-text search requires a positive limit"); + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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 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(); + 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 result, reader.VisitEqual(Literal(1))); + ASSERT_EQ(result->ToString(), "{1,2}"); + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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 cd8666098..6ac7f5436 100644 --- a/src/paimon/core/io/data_file_meta.cpp +++ b/src/paimon/core/io/data_file_meta.cpp @@ -20,6 +20,7 @@ #include #include +#include #include #include "arrow/type_fwd.h" @@ -167,6 +168,17 @@ Result DataFileMeta::CreationTimeEpochMillis() const { return utc_ts.GetMillisecond(); } +Result DataFileMeta::NonNullRowIdRange() const { + PAIMON_ASSIGN_OR_RAISE(int64_t start, NonNullFirstRowId()); + if (start < 0 || row_count <= 0 || + row_count - 1 > std::numeric_limits::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> DataFileMeta::ToFileSelection( const std::optional>& 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 ee3d4a9f5..0c9409c50 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 NonNullRowIdRange() const; + // empty row_ranges indicates all rows in the file are needed, return null bitmap Result> ToFileSelection( const std::optional>& 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 f76bc0ae8..b35631e42 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 + #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& first_row_id) { + return std::make_shared( + "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>(), + /*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::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( diff --git a/src/paimon/core/operation/file_store_scan.h b/src/paimon/core/operation/file_store_scan.h index b73ad095d..25911d0e5 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& partition_filter, + const std::shared_ptr& 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 0a76e75a0..62f334db3 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> DataEvolutionSplitGenerator::SplitForBatch( @@ -37,8 +38,8 @@ Result> DataEvolutionSplitGenerator::Spl return meta->NonNullFirstRowId(); }, [](const std::shared_ptr& meta) -> Result { - 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>> ranges, diff --git a/src/paimon/core/table/source/split_generator_test.cpp b/src/paimon/core/table/source/split_generator_test.cpp index 9f1368700..4ba05c3d5 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 #include +#include #include #include #include @@ -260,6 +261,22 @@ TEST_F(SplitGeneratorTest, TestDataEvolutionBlobSplitByFileSize) { } } +TEST_F(SplitGeneratorTest, TestDataEvolutionRowIdRangeAtInt64Max) { + constexpr int64_t kMax = std::numeric_limits::max(); + DataEvolutionSplitGenerator split_generator(/*target_split_size=*/1000, /*open_file_cost=*/10, + /*count_blob_size=*/true); + ASSERT_OK_AND_ASSIGN(std::vector 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> 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 c508fd199..2924e5413 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 CreateDataEvolutionTable( - bool deletion_vectors_enabled, - const std::map& extra_options = {}) const { + bool deletion_vectors_enabled, const std::map& extra_options = {}, + const std::vector& partition_keys = {}) const { std::map 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> group_msgs0, + WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0, 1, 2, 3}, + /*partition=*/{{"f1", "p0"}})); + ASSERT_OK_AND_ASSIGN( + std::vector> group_msgs1, + WriteAndCommitGroup(table_path, /*first_row_id=*/4, /*f0_values=*/{4, 5, 6, 7}, + /*partition=*/{{"f1", "p1"}})); + ASSERT_OK_AND_ASSIGN(std::vector anchor_file_names, + PlannedAnchorFileNames(table_path)); + ASSERT_EQ(anchor_file_names.size(), 2); + + std::shared_ptr fs = dir_->GetFileSystem(); + SchemaManager schema_manager(fs, table_path); + ASSERT_OK_AND_ASSIGN(std::optional> 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_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_manager.LatestSnapshot()); + ASSERT_TRUE(snapshot); + + std::shared_ptr executor = CreateDefaultExecutor(); + auto read_live_rows = [&](const Snapshot& read_snapshot, + const std::shared_ptr& partition_filter) { + return GlobalIndexLiveRowFilter::LiveRows(table_path, table_schema.value(), read_snapshot, + partition_filter, options, executor, + GetDefaultPool()); + }; + ASSERT_OK_AND_ASSIGN(std::optional 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 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 9c36766b2..97b169823 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& index_options) { + arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), + arrow::field("f1", arrow::int32())}; + auto schema = arrow::schema(fields); + std::map 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 scan_before_deletion, create_scan()); + + ASSERT_OK_AND_ASSIGN(auto split, ScanData(table_path, /*partition_filters=*/{})); + ASSERT_OK_AND_ASSIGN(std::vector anchor_file_names, + DeletionVectorTestHelper::RetrieveAnchorFileNames(split->DataFiles())); + ASSERT_EQ(anchor_file_names.size(), 2u); + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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 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 index_result, + readers[0]->VisitFullTextSearch( + std::make_shared("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> ScanGlobalIndexAndData( const std::string& table_path, const std::shared_ptr& predicate, const std::map& 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) {