This is an automated email from the ASF dual-hosted git repository.
SteNicholas 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 da24fbdf feat(scan): support file indexes in real-time and PK scans
(#370)
da24fbdf is described below
commit da24fbdf55f9264076b43dac4eeba2f64514620a
Author: lxy <[email protected]>
AuthorDate: Mon Sep 21 14:52:24 2026 +0800
feat(scan): support file indexes in real-time and PK scans (#370)
---
.../bitmap/apply_bitmap_index_batch_reader.h | 18 +-
src/paimon/core/operation/abstract_split_read.cpp | 121 ++++++++-
src/paimon/core/operation/abstract_split_read.h | 18 +-
.../core/operation/append_only_file_store_scan.cpp | 43 +---
.../core/operation/append_only_file_store_scan.h | 6 -
src/paimon/core/operation/file_store_scan.cpp | 31 +++
src/paimon/core/operation/file_store_scan.h | 8 +
.../core/operation/key_value_file_store_scan.cpp | 12 +-
.../core/operation/merge_file_split_read.cpp | 63 +----
src/paimon/core/operation/merge_file_split_read.h | 8 -
src/paimon/core/operation/raw_file_split_read.cpp | 96 -------
src/paimon/core/operation/raw_file_split_read.h | 8 -
test/inte/realtime_write_inte_test.cpp | 286 +++++++++++++++++++++
test/inte/scan_and_read_inte_test.cpp | 161 ++++++++++++
14 files changed, 652 insertions(+), 227 deletions(-)
diff --git
a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h
b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h
index efb159ea..114e56b9 100644
--- a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h
+++ b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h
@@ -101,11 +101,25 @@ class ApplyBitmapIndexBatchReader : public
FileBatchReader {
private:
Result<RoaringBitmap32> Filter(int32_t batch_size) const {
RoaringBitmap32 result;
- auto bitmap_iter = bitmap_.Begin();
+ if (batch_size == 0) {
+ return result;
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(uint64_t first_file_row_id,
+
reader_->GetPreviousBatchFileRowId(/*batch_row_id=*/0));
+ if (first_file_row_id >
static_cast<uint64_t>(RoaringBitmap32::MAX_VALUE)) {
+ return result;
+ }
+ // Avoid rescanning bitmap entries before this batch while retaining a
single linear
+ // iterator for all rows in the batch.
+ auto bitmap_iter =
bitmap_.EqualOrLarger(static_cast<int32_t>(first_file_row_id));
auto bitmap_end = bitmap_.End();
for (int32_t i = 0; i < batch_size; ++i) {
- PAIMON_ASSIGN_OR_RAISE(uint64_t file_row_id,
reader_->GetPreviousBatchFileRowId(i));
+ uint64_t file_row_id = first_file_row_id;
+ if (i > 0) {
+ PAIMON_ASSIGN_OR_RAISE(file_row_id,
reader_->GetPreviousBatchFileRowId(i));
+ }
while (bitmap_iter != bitmap_end &&
static_cast<uint64_t>(*bitmap_iter) < file_row_id) {
++bitmap_iter;
}
diff --git a/src/paimon/core/operation/abstract_split_read.cpp
b/src/paimon/core/operation/abstract_split_read.cpp
index fadf5178..53b9a40e 100644
--- a/src/paimon/core/operation/abstract_split_read.cpp
+++ b/src/paimon/core/operation/abstract_split_read.cpp
@@ -26,6 +26,8 @@
#include <set>
#include <utility>
+#include "arrow/c/abi.h"
+#include "arrow/c/bridge.h"
#include "arrow/type.h"
#include "fmt/format.h"
#include "paimon/common/data/blob_defs.h"
@@ -37,6 +39,7 @@
#include "paimon/common/data/variant/variant_type_utils.h"
#include "paimon/common/executor/future.h"
#include "paimon/common/executor/reader_build_executor.h"
+#include "paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h"
#include "paimon/common/reader/data_file_reader_factory.h"
#include "paimon/common/reader/delegating_prefetch_reader.h"
#include "paimon/common/reader/late_materializing_reader_builder.h"
@@ -44,11 +47,14 @@
#include "paimon/common/reader/prefetch_file_batch_reader_impl.h"
#include "paimon/common/table/special_fields.h"
#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/object_utils.h"
+#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/io/complete_row_tracking_fields_reader.h"
#include "paimon/core/io/data_file_meta.h"
#include "paimon/core/io/data_file_path_factory.h"
#include "paimon/core/io/field_mapping_reader.h"
+#include "paimon/core/io/file_index_evaluator.h"
#include "paimon/core/io/vector_file_batch_reader.h"
#include "paimon/core/operation/internal_read_context.h"
#include "paimon/core/partition/partition_info.h"
@@ -56,10 +62,13 @@
#include "paimon/core/table/source/data_split_impl.h"
#include "paimon/core/utils/field_mapping.h"
#include "paimon/core/utils/nested_projection_utils.h"
+#include "paimon/file_index/bitmap_index_result.h"
+#include "paimon/file_index/file_index_result.h"
#include "paimon/format/file_format.h"
#include "paimon/format/file_format_factory.h"
#include "paimon/fs/file_system.h"
#include "paimon/status.h"
+#include "paimon/utils/roaring_bitmap32.h"
namespace paimon {
class BinaryRow;
@@ -83,19 +92,38 @@ AbstractSplitRead::AbstractSplitRead(const
std::shared_ptr<FileStorePathFactory>
schema_manager_(std::move(schema_manager)) {}
Result<std::vector<std::unique_ptr<FileBatchReader>>>
AbstractSplitRead::CreateRawFileReaders(
+ const BinaryRow& partition, const
std::vector<std::shared_ptr<DataFileMeta>>& data_files,
+ const std::shared_ptr<arrow::Schema>& read_schema, const
std::shared_ptr<Predicate>& predicate,
+ DeletionVector::Factory dv_factory, const
std::optional<std::vector<Range>>& row_ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const std::map<std::string, std::string>& extra_format_options) const {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<RawFileReaderWithMeta> readers_with_meta,
+ CreateRawFileReadersWithMeta(partition, data_files, read_schema,
predicate, dv_factory,
+ row_ranges, data_file_path_factory,
extra_format_options));
+ std::vector<std::unique_ptr<FileBatchReader>> raw_file_readers;
+ raw_file_readers.reserve(readers_with_meta.size());
+ for (auto& reader_with_meta : readers_with_meta) {
+ raw_file_readers.push_back(std::move(reader_with_meta.reader));
+ }
+ return raw_file_readers;
+}
+
+Result<std::vector<AbstractSplitRead::RawFileReaderWithMeta>>
+AbstractSplitRead::CreateRawFileReadersWithMeta(
const BinaryRow& partition, const
std::vector<std::shared_ptr<DataFileMeta>>& data_files,
const std::shared_ptr<arrow::Schema>& read_schema, const
std::shared_ptr<Predicate>& predicate,
DeletionVector::Factory dv_factory, const
std::optional<std::vector<Range>>& row_ranges,
const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
const std::map<std::string, std::string>& extra_format_options) const {
if (data_files.empty()) {
- return std::vector<std::unique_ptr<FileBatchReader>>();
+ return std::vector<RawFileReaderWithMeta>();
}
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<FieldMappingBuilder> field_mapping_builder,
FieldMappingBuilder::Create(read_schema, context_->GetPartitionKeys(),
predicate));
- std::vector<std::unique_ptr<FileBatchReader>> raw_file_readers;
+ std::vector<RawFileReaderWithMeta> raw_file_readers;
raw_file_readers.reserve(data_files.size());
const uint32_t parallel_num =
static_cast<uint32_t>(std::min<size_t>(kReaderBuildMaxParallelNum,
data_files.size()));
@@ -106,7 +134,7 @@ Result<std::vector<std::unique_ptr<FileBatchReader>>>
AbstractSplitRead::CreateR
CreateRawFileReader(partition, file,
field_mapping_builder.get(), dv_factory,
row_ranges, data_file_path_factory,
extra_format_options));
if (file_reader) {
- raw_file_readers.push_back(std::move(file_reader));
+ raw_file_readers.push_back({file, std::move(file_reader)});
}
}
return std::move(raw_file_readers);
@@ -129,7 +157,9 @@ Result<std::vector<std::unique_ptr<FileBatchReader>>>
AbstractSplitRead::CreateR
}
// Readers keep the file order of the split, and CollectAll preserves the
submit order.
Status first_error;
+ size_t file_index = 0;
for (auto& built : CollectAll(futures)) {
+ const std::shared_ptr<DataFileMeta>& file = data_files[file_index++];
if (!built.ok()) {
if (first_error.ok()) {
first_error = built.status();
@@ -138,13 +168,96 @@ Result<std::vector<std::unique_ptr<FileBatchReader>>>
AbstractSplitRead::CreateR
}
std::unique_ptr<FileBatchReader> file_reader =
std::move(built).value();
if (file_reader) {
- raw_file_readers.push_back(std::move(file_reader));
+ raw_file_readers.push_back({file, std::move(file_reader)});
}
}
PAIMON_RETURN_NOT_OK(first_error);
return std::move(raw_file_readers);
}
+Result<std::unique_ptr<FileBatchReader>>
AbstractSplitRead::ApplyIndexAndDvReaderIfNeeded(
+ std::unique_ptr<FileBatchReader>&& file_reader, const
std::shared_ptr<DataFileMeta>& file,
+ const std::shared_ptr<arrow::Schema>& data_schema,
+ const std::shared_ptr<arrow::Schema>& read_schema, const
std::shared_ptr<Predicate>& predicate,
+ DeletionVector::Factory dv_factory, const
std::optional<std::vector<Range>>& row_ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
+ std::shared_ptr<FileIndexResult> file_index_result;
+ if (options_.FileIndexReadEnabled()) {
+ PAIMON_ASSIGN_OR_RAISE(
+ file_index_result,
+ FileIndexEvaluator::Evaluate(data_schema, predicate,
data_file_path_factory, file,
+ options_.GetFileSystem(), pool_));
+ PAIMON_ASSIGN_OR_RAISE(bool is_remain, file_index_result->IsRemain());
+ if (!is_remain) {
+ return std::unique_ptr<FileBatchReader>();
+ }
+ }
+
+ // prepare selection bitmap for index
+ const RoaringBitmap32* selection = nullptr;
+ if (auto* bitmap_file_index =
dynamic_cast<BitmapIndexResult*>(file_index_result.get())) {
+ PAIMON_ASSIGN_OR_RAISE(selection, bitmap_file_index->GetBitmap());
+ }
+
+ // narrow the selection to the file-local row positions of an indexed split
+ std::optional<RoaringBitmap32> row_ranges_selection;
+ if (row_ranges) {
+ RoaringBitmap32 row_ranges_bitmap;
+ for (const Range& range : row_ranges.value()) {
+ row_ranges_bitmap.AddRange(static_cast<int32_t>(range.from),
+ static_cast<int32_t>(range.to + 1));
+ }
+ row_ranges_selection = selection ? RoaringBitmap32::And(*selection,
row_ranges_bitmap)
+ : std::move(row_ranges_bitmap);
+ selection = &row_ranges_selection.value();
+ }
+
+ // prepare deletion bitmap for deletion vector
+ std::shared_ptr<DeletionVector> deletion_vector;
+ if (dv_factory) {
+ PAIMON_ASSIGN_OR_RAISE(deletion_vector, dv_factory(file->file_name));
+ }
+ const RoaringBitmap32* deletion = nullptr;
+ if (auto* bitmap_dv =
dynamic_cast<BitmapDeletionVector*>(deletion_vector.get())) {
+ deletion = bitmap_dv->GetBitmap();
+ }
+
+ // merge deletion and bitmap index selection
+ std::optional<RoaringBitmap32> actual_selection;
+ if (selection && deletion) {
+ actual_selection = RoaringBitmap32::AndNot(*selection, *deletion);
+ } else if (selection) {
+ actual_selection = *selection;
+ } else if (deletion) {
+ actual_selection = *deletion;
+ PAIMON_ASSIGN_OR_RAISE(uint64_t num_rows,
file_reader->GetNumberOfRows());
+ actual_selection->Flip(0, num_rows);
+ }
+
+ if (actual_selection && actual_selection->IsEmpty()) {
+ return std::unique_ptr<FileBatchReader>();
+ }
+
+ ::ArrowSchema c_read_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema,
&c_read_schema));
+ PAIMON_RETURN_NOT_OK(file_reader->SetReadSchema(&c_read_schema, predicate,
actual_selection));
+
+ std::unique_ptr<FileBatchReader> reader;
+ if (!file_reader->SupportPreciseBitmapSelection() && actual_selection) {
+ reader =
std::make_unique<ApplyBitmapIndexBatchReader>(std::move(file_reader),
+
std::move(actual_selection).value());
+ } else {
+ reader = std::move(file_reader);
+ }
+
+ if (deletion_vector && !deletion && !deletion_vector->IsEmpty()) {
+ // TODO(xinyu.lxy): if deletion vector is bitmap64, use
ApplyBitmapIndexBatchReader to
+ // filter result
+ return Status::NotImplemented("Only support BitmapDeletionVector");
+ }
+ return reader;
+}
+
Result<std::unique_ptr<FileBatchReader>>
AbstractSplitRead::CreateRawFileReader(
const BinaryRow& partition, const std::shared_ptr<DataFileMeta>& file,
const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory
dv_factory,
diff --git a/src/paimon/core/operation/abstract_split_read.h
b/src/paimon/core/operation/abstract_split_read.h
index a52bed66..4e1ef86d 100644
--- a/src/paimon/core/operation/abstract_split_read.h
+++ b/src/paimon/core/operation/abstract_split_read.h
@@ -77,11 +77,27 @@ class AbstractSplitRead : public SplitRead {
const std::map<std::string, std::string>& extra_format_options) const;
protected:
+ struct RawFileReaderWithMeta {
+ std::shared_ptr<DataFileMeta> file;
+ std::unique_ptr<FileBatchReader> reader;
+ };
+
AbstractSplitRead(const std::shared_ptr<FileStorePathFactory>&
path_factory,
const std::shared_ptr<InternalReadContext>& context,
std::unique_ptr<SchemaManager>&& schema_manager,
const std::shared_ptr<MemoryPool>& memory_pool,
const std::shared_ptr<Executor>& executor);
+
+ /// Creates raw readers while preserving the metadata associated with
every reader which was
+ /// not eliminated by a file index or deletion vector.
+ Result<std::vector<RawFileReaderWithMeta>> CreateRawFileReadersWithMeta(
+ const BinaryRow& partition, const
std::vector<std::shared_ptr<DataFileMeta>>& data_files,
+ const std::shared_ptr<arrow::Schema>& read_schema,
+ const std::shared_ptr<Predicate>& predicate, DeletionVector::Factory
dv_factory,
+ const std::optional<std::vector<Range>>& row_ranges,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const std::map<std::string, std::string>& extra_format_options) const;
+
Result<std::unique_ptr<BatchReader>> ApplyPredicateFilterIfNeeded(
std::unique_ptr<BatchReader>&& reader, const
std::shared_ptr<Predicate>& predicate) const;
@@ -93,7 +109,7 @@ class AbstractSplitRead : public SplitRead {
const std::shared_ptr<arrow::Schema>& read_schema,
const std::shared_ptr<Predicate>& predicate, DeletionVector::Factory
dv_factory,
const std::optional<std::vector<Range>>& row_ranges,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory)
const = 0;
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory)
const;
// 1. project write cols to data schema
// 2. add partition fields (if write cols not contain)
diff --git a/src/paimon/core/operation/append_only_file_store_scan.cpp
b/src/paimon/core/operation/append_only_file_store_scan.cpp
index 435ece57..78bf3c7a 100644
--- a/src/paimon/core/operation/append_only_file_store_scan.cpp
+++ b/src/paimon/core/operation/append_only_file_store_scan.cpp
@@ -18,31 +18,18 @@
#include "paimon/core/operation/append_only_file_store_scan.h"
-#include <cassert>
#include <cstdint>
#include <exception>
-#include <map>
-#include <optional>
-#include <set>
-#include <string>
-#include <utility>
-#include <vector>
-#include "arrow/type.h"
#include "fmt/format.h"
#include "paimon/common/predicate/predicate_filter.h"
-#include "paimon/common/types/data_field.h"
#include "paimon/core/core_options.h"
#include "paimon/core/io/data_file_meta.h"
-#include "paimon/core/io/file_index_evaluator.h"
#include "paimon/core/manifest/manifest_entry.h"
#include "paimon/core/schema/schema_manager.h"
#include "paimon/core/schema/table_schema.h"
#include "paimon/core/stats/simple_stats_evolution.h"
#include "paimon/core/stats/simple_stats_evolutions.h"
-#include "paimon/core/utils/field_mapping.h"
-#include "paimon/file_index/file_index_result.h"
-#include "paimon/predicate/predicate_utils.h"
#include "paimon/scan_context.h"
#include "paimon/status.h"
@@ -130,35 +117,7 @@ Result<bool> AppendOnlyFileStoreScan::FilterByStats(const
ManifestEntry& entry)
fmt::format("FilterByStats failed for file {}, with unknown
error", meta->file_name));
}
- if (!core_options_.FileIndexReadEnabled()) {
- return true;
- }
-
- return TestFileIndex(meta, evolution, data_schema);
-}
-
-Result<bool> AppendOnlyFileStoreScan::TestFileIndex(
- const std::shared_ptr<DataFileMeta>& meta,
- const std::shared_ptr<SimpleStatsEvolution>& evolution,
- const std::shared_ptr<TableSchema>& data_schema) const {
- std::shared_ptr<Predicate> data_predicate = predicates_;
- if (data_schema->Id() != table_schema_->Id()) {
- PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<Predicate>>
reconstruct_predicate,
-
FieldMappingBuilder::ReconstructPredicateWithDataFields(
- predicates_,
evolution->GetFieldNameToTableField(),
- evolution->GetFieldIdToDataField()));
-
- if (reconstruct_predicate == std::nullopt) {
- return true;
- }
- data_predicate = reconstruct_predicate.value();
- }
- assert(data_predicate);
- auto data_arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(data_schema->Fields());
- PAIMON_ASSIGN_OR_RAISE(
- std::shared_ptr<FileIndexResult> index_result,
- FileIndexEvaluator::Evaluate(data_arrow_schema, data_predicate, meta,
pool_));
- return index_result->IsRemain();
+ return TestFileIndex(predicates_, meta, evolution, data_schema);
}
} // namespace paimon
diff --git a/src/paimon/core/operation/append_only_file_store_scan.h
b/src/paimon/core/operation/append_only_file_store_scan.h
index ba4feb1f..c7bac6bf 100644
--- a/src/paimon/core/operation/append_only_file_store_scan.h
+++ b/src/paimon/core/operation/append_only_file_store_scan.h
@@ -41,11 +41,9 @@ class ManifestList;
class MemoryPool;
class ScanFilter;
class SchemaManager;
-class SimpleStatsEvolution;
class SimpleStatsEvolutions;
class SnapshotManager;
class TableSchema;
-struct DataFileMeta;
/// `FileStoreScan` for `AppendOnlyFileStore`.
class AppendOnlyFileStoreScan : public FileStoreScan {
@@ -64,10 +62,6 @@ class AppendOnlyFileStoreScan : public FileStoreScan {
Result<bool> FilterByStats(const ManifestEntry& entry) const override;
private:
- Result<bool> TestFileIndex(const std::shared_ptr<DataFileMeta>& meta,
- const std::shared_ptr<SimpleStatsEvolution>&
evolution,
- const std::shared_ptr<TableSchema>&
data_schema) const;
-
AppendOnlyFileStoreScan(const std::shared_ptr<SnapshotManager>&
snapshot_manager,
const std::shared_ptr<SchemaManager>&
schema_manager,
const std::shared_ptr<ManifestList>& manifest_list,
diff --git a/src/paimon/core/operation/file_store_scan.cpp
b/src/paimon/core/operation/file_store_scan.cpp
index 45376b22..fa22ac56 100644
--- a/src/paimon/core/operation/file_store_scan.cpp
+++ b/src/paimon/core/operation/file_store_scan.cpp
@@ -38,6 +38,7 @@
#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/field_type_utils.h"
#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/file_index_evaluator.h"
#include "paimon/core/manifest/file_entry.h"
#include "paimon/core/manifest/file_kind.h"
#include "paimon/core/manifest/manifest_file.h"
@@ -45,12 +46,14 @@
#include "paimon/core/manifest/manifest_list.h"
#include "paimon/core/manifest/snapshot_live_manifest_entries.h"
#include "paimon/core/partition/partition_info.h"
+#include "paimon/core/schema/table_schema.h"
#include "paimon/core/stats/simple_stats.h"
#include "paimon/core/stats/simple_stats_evolution.h"
#include "paimon/core/utils/branch_manager.h"
#include "paimon/core/utils/duration.h"
#include "paimon/core/utils/field_mapping.h"
#include "paimon/core/utils/snapshot_manager.h"
+#include "paimon/file_index/file_index_result.h"
#include "paimon/memory/bytes.h"
#include "paimon/memory/memory_segment.h"
#include "paimon/predicate/literal.h"
@@ -90,6 +93,34 @@ Result<std::shared_ptr<Predicate>>
FileStoreScan::ReconstructPredicateWithNonCas
return PredicateUtils::ExcludePredicateWithFields(predicate,
excluded_field_names);
}
+Result<bool> FileStoreScan::TestFileIndex(const std::shared_ptr<Predicate>&
predicate,
+ const std::shared_ptr<DataFileMeta>&
meta,
+ const
std::shared_ptr<SimpleStatsEvolution>& evolution,
+ const std::shared_ptr<TableSchema>&
data_schema) const {
+ if (!core_options_.FileIndexReadEnabled() || meta->embedded_index ==
nullptr) {
+ return true;
+ }
+
+ std::shared_ptr<Predicate> data_predicate = predicate;
+ if (data_schema->Id() != table_schema_->Id()) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<Predicate>>
reconstructed_predicate,
+
FieldMappingBuilder::ReconstructPredicateWithDataFields(
+ predicate,
evolution->GetFieldNameToTableField(),
+ evolution->GetFieldIdToDataField()));
+ if (!reconstructed_predicate) {
+ return true;
+ }
+ data_predicate = reconstructed_predicate.value();
+ }
+ assert(data_predicate);
+ std::shared_ptr<arrow::Schema> data_arrow_schema =
+ DataField::ConvertDataFieldsToArrowSchema(data_schema->Fields());
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileIndexResult> index_result,
+ FileIndexEvaluator::Evaluate(data_arrow_schema, data_predicate, meta,
pool_));
+ return index_result->IsRemain();
+}
+
std::vector<ManifestEntry> FileStoreScan::RawPlan::Files(const FileKind& kind)
{
std::vector<ManifestEntry> entries = Files();
std::vector<ManifestEntry> filtered_entries;
diff --git a/src/paimon/core/operation/file_store_scan.h
b/src/paimon/core/operation/file_store_scan.h
index 0eef2909..5fe128da 100644
--- a/src/paimon/core/operation/file_store_scan.h
+++ b/src/paimon/core/operation/file_store_scan.h
@@ -73,6 +73,7 @@ class SchemaManager;
class SimpleStatsEvolution;
class SnapshotManager;
class TableSchema;
+struct DataFileMeta;
/// Scan operation which produces a plan.
class FileStoreScan {
@@ -257,6 +258,13 @@ class FileStoreScan {
const std::shared_ptr<Predicate>& predicate,
const std::shared_ptr<SimpleStatsEvolution>& evolution);
+ /// Tests only an index embedded in DataFileMeta. External index files are
evaluated while
+ /// reading the retained files.
+ Result<bool> TestFileIndex(const std::shared_ptr<Predicate>& predicate,
+ const std::shared_ptr<DataFileMeta>& meta,
+ const std::shared_ptr<SimpleStatsEvolution>&
evolution,
+ const std::shared_ptr<TableSchema>&
data_schema) const;
+
private:
Status ReadManifests(std::optional<Snapshot>* snapshot_ptr,
std::vector<ManifestFileMeta>* all_manifests_ptr,
diff --git a/src/paimon/core/operation/key_value_file_store_scan.cpp
b/src/paimon/core/operation/key_value_file_store_scan.cpp
index b89a147a..e541fce3 100644
--- a/src/paimon/core/operation/key_value_file_store_scan.cpp
+++ b/src/paimon/core/operation/key_value_file_store_scan.cpp
@@ -171,9 +171,6 @@ Result<bool>
KeyValueFileStoreScan::FilterByValueFilter(const ManifestEntry& ent
if (!value_filter_) {
return true;
}
- if (entry.File()->embedded_index != nullptr) {
- return Status::NotImplemented("do not support embedded index in
DataFileMeta");
- }
const auto& meta = entry.File();
@@ -201,12 +198,12 @@ Result<bool>
KeyValueFileStoreScan::FilterByValueFilter(const ManifestEntry& ent
SimpleStatsEvolution::EvolutionStats new_stats,
evolution->Evolution(meta->value_stats, meta->row_count,
meta->value_stats_cols));
+ bool predicate_result = false;
try {
PAIMON_ASSIGN_OR_RAISE(
- bool predicate_result,
+ predicate_result,
predicate_filter->Test(schema_, meta->row_count,
*(new_stats.min_values),
*(new_stats.max_values),
*(new_stats.null_counts)));
- return predicate_result;
} catch (const std::exception& e) {
return Status::Invalid(fmt::format("FilterByValueFilter failed for
file {}, with {} error",
meta->file_name, e.what()));
@@ -214,6 +211,11 @@ Result<bool>
KeyValueFileStoreScan::FilterByValueFilter(const ManifestEntry& ent
return Status::Invalid(fmt::format(
"FilterByValueFilter failed for file {}, with unknown error",
meta->file_name));
}
+ if (!predicate_result) {
+ return false;
+ }
+
+ return TestFileIndex(value_filter_, meta, evolution, data_schema);
}
bool KeyValueFileStoreScan::NoOverlapping(const std::vector<ManifestEntry>&
entries) {
diff --git a/src/paimon/core/operation/merge_file_split_read.cpp
b/src/paimon/core/operation/merge_file_split_read.cpp
index 7c6b1724..679c48ca 100644
--- a/src/paimon/core/operation/merge_file_split_read.cpp
+++ b/src/paimon/core/operation/merge_file_split_read.cpp
@@ -26,8 +26,6 @@
#include <set>
#include <utility>
-#include "arrow/c/abi.h"
-#include "arrow/c/bridge.h"
#include "arrow/type.h"
#include "fmt/format.h"
#include "paimon/common/reader/complete_row_kind_batch_reader.h"
@@ -38,8 +36,6 @@
#include "paimon/common/utils/object_utils.h"
#include "paimon/common/utils/scope_guard.h"
#include "paimon/core/core_options.h"
-#include "paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h"
-#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/deletionvectors/deletion_vector.h"
#include "paimon/core/io/async_key_value_projection_reader.h"
#include "paimon/core/io/concat_key_value_record_reader.h"
@@ -69,7 +65,6 @@
#include "paimon/predicate/predicate_utils.h"
#include "paimon/reader/file_batch_reader.h"
#include "paimon/table/source/data_split.h"
-#include "paimon/utils/roaring_bitmap32.h"
namespace paimon {
class BinaryRow;
@@ -482,47 +477,6 @@ MergeFileSplitRead::CreateMergeFunctionWrapper(const
CoreOptions& core_options,
return
std::make_shared<ReducerMergeFunctionWrapper>(std::move(merge_function));
}
-Result<std::unique_ptr<FileBatchReader>>
MergeFileSplitRead::ApplyIndexAndDvReaderIfNeeded(
- std::unique_ptr<FileBatchReader>&& file_reader, const
std::shared_ptr<DataFileMeta>& file,
- const std::shared_ptr<arrow::Schema>& data_schema,
- const std::shared_ptr<arrow::Schema>& read_schema, const
std::shared_ptr<Predicate>& predicate,
- DeletionVector::Factory dv_factory, const
std::optional<std::vector<Range>>& ranges,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
- // merge read does not use index
- std::shared_ptr<DeletionVector> deletion_vector;
- if (dv_factory) {
- PAIMON_ASSIGN_OR_RAISE(deletion_vector, dv_factory(file->file_name));
- }
-
- const RoaringBitmap32* deletion = nullptr;
- if (auto* bitmap_dv =
dynamic_cast<BitmapDeletionVector*>(deletion_vector.get())) {
- deletion = bitmap_dv->GetBitmap();
- }
-
- std::optional<RoaringBitmap32> actual_selection;
- if (deletion) {
- actual_selection = *deletion;
- PAIMON_ASSIGN_OR_RAISE(uint64_t num_rows,
file_reader->GetNumberOfRows());
- actual_selection.value().Flip(0, num_rows);
- }
-
- ::ArrowSchema c_read_schema;
- PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema,
&c_read_schema));
-
- PAIMON_RETURN_NOT_OK(file_reader->SetReadSchema(&c_read_schema, predicate,
actual_selection));
-
- if (!file_reader->SupportPreciseBitmapSelection() && actual_selection) {
- return
std::make_unique<ApplyDeletionVectorBatchReader>(std::move(file_reader),
-
deletion_vector);
- }
- if (deletion_vector && !deletion && !deletion_vector->IsEmpty()) {
- // TODO(xinyu.lxy): if deletion vector is bitmap64, use
ApplyBitmapIndexBatchReader to
- // filter result
- return Status::NotImplemented("Only support BitmapDeletionVector");
- }
- return std::move(file_reader);
-}
-
Result<std::unique_ptr<BatchReader>> MergeFileSplitRead::CreateMergeReader(
const std::shared_ptr<DataSplitImpl>& data_split,
const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) {
@@ -867,19 +821,18 @@ Result<std::unique_ptr<KeyValueRecordReader>>
MergeFileSplitRead::CreateReaderFo
// no overlap in a run
const auto& data_files = sorted_run.Files();
PAIMON_ASSIGN_OR_RAISE(
- std::vector<std::unique_ptr<FileBatchReader>> raw_file_readers,
- CreateRawFileReaders(partition, data_files, read_schema_, predicate,
dv_factory,
- /*row_ranges=*/{}, data_file_path_factory,
- /*extra_format_options=*/{}));
+ std::vector<RawFileReaderWithMeta> raw_file_readers,
+ CreateRawFileReadersWithMeta(partition, data_files, read_schema_,
predicate, dv_factory,
+ /*row_ranges=*/{}, data_file_path_factory,
+ /*extra_format_options=*/{}));
- assert(data_files.size() == raw_file_readers.size());
// KeyValueDataFileRecordReader converts arrow array from format reader to
KeyValue objects
std::vector<std::unique_ptr<KeyValueRecordReader>> file_record_readers;
- file_record_readers.reserve(data_files.size());
- for (size_t i = 0; i < data_files.size(); i++) {
+ file_record_readers.reserve(raw_file_readers.size());
+ for (auto& raw_file_reader : raw_file_readers) {
file_record_readers.push_back(std::make_unique<KeyValueDataFileRecordReader>(
- std::move(raw_file_readers[i]), key_schema_, value_schema_,
data_files[i]->level,
- pool_));
+ std::move(raw_file_reader.reader), key_schema_, value_schema_,
+ raw_file_reader.file->level, pool_));
}
return
std::make_unique<ConcatKeyValueRecordReader>(std::move(file_record_readers));
}
diff --git a/src/paimon/core/operation/merge_file_split_read.h
b/src/paimon/core/operation/merge_file_split_read.h
index 07b5e70b..45f19a46 100644
--- a/src/paimon/core/operation/merge_file_split_read.h
+++ b/src/paimon/core/operation/merge_file_split_read.h
@@ -91,14 +91,6 @@ class MergeFileSplitRead : public AbstractSplitRead {
force_keep_delete_ = force_keep_delete;
}
- Result<std::unique_ptr<FileBatchReader>> ApplyIndexAndDvReaderIfNeeded(
- std::unique_ptr<FileBatchReader>&& file_reader, const
std::shared_ptr<DataFileMeta>& file,
- const std::shared_ptr<arrow::Schema>& data_schema,
- const std::shared_ptr<arrow::Schema>& read_schema,
- const std::shared_ptr<Predicate>& predicate, DeletionVector::Factory
dv_factory,
- const std::optional<std::vector<Range>>& ranges,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory)
const override;
-
Result<std::unique_ptr<SortMergeReader>> CreateSortMergeReaderForSection(
const std::vector<SortedRun>& section, const BinaryRow& partition,
DeletionVector::Factory dv_factory, const std::shared_ptr<Predicate>&
predicate,
diff --git a/src/paimon/core/operation/raw_file_split_read.cpp
b/src/paimon/core/operation/raw_file_split_read.cpp
index 8439ba19..8830b8ad 100644
--- a/src/paimon/core/operation/raw_file_split_read.cpp
+++ b/src/paimon/core/operation/raw_file_split_read.cpp
@@ -23,32 +23,23 @@
#include <utility>
#include <vector>
-#include "arrow/c/abi.h"
-#include "arrow/c/bridge.h"
#include "fmt/format.h"
-#include "paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h"
#include "paimon/common/reader/complete_row_kind_batch_reader.h"
#include "paimon/common/reader/concat_batch_reader.h"
-#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/object_utils.h"
#include "paimon/core/core_options.h"
-#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/deletionvectors/deletion_vector.h"
#include "paimon/core/global_index/indexed_split_impl.h"
#include "paimon/core/io/data_file_meta.h"
-#include "paimon/core/io/file_index_evaluator.h"
#include "paimon/core/operation/internal_read_context.h"
#include "paimon/core/schema/schema_manager.h"
#include "paimon/core/schema/table_schema.h"
#include "paimon/core/table/source/data_split_impl.h"
#include "paimon/core/utils/file_store_path_factory.h"
-#include "paimon/file_index/bitmap_index_result.h"
-#include "paimon/file_index/file_index_result.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/reader/file_batch_reader.h"
#include "paimon/status.h"
#include "paimon/table/source/data_split.h"
-#include "paimon/utils/roaring_bitmap32.h"
namespace paimon {
class DataFilePathFactory;
@@ -189,91 +180,4 @@ Result<bool> RawFileSplitRead::Match(const
std::shared_ptr<Split>& split,
return matched;
}
-Result<std::unique_ptr<FileBatchReader>>
RawFileSplitRead::ApplyIndexAndDvReaderIfNeeded(
- std::unique_ptr<FileBatchReader>&& file_reader, const
std::shared_ptr<DataFileMeta>& file,
- const std::shared_ptr<arrow::Schema>& data_schema,
- const std::shared_ptr<arrow::Schema>& read_schema, const
std::shared_ptr<Predicate>& predicate,
- DeletionVector::Factory dv_factory, const
std::optional<std::vector<Range>>& ranges,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
- std::shared_ptr<FileIndexResult> file_index_result;
- if (options_.FileIndexReadEnabled()) {
- PAIMON_ASSIGN_OR_RAISE(
- file_index_result,
- FileIndexEvaluator::Evaluate(data_schema, predicate,
data_file_path_factory, file,
- options_.GetFileSystem(), pool_));
- PAIMON_ASSIGN_OR_RAISE(bool is_remain, file_index_result->IsRemain());
- if (!is_remain) {
- return std::unique_ptr<FileBatchReader>();
- }
- }
- // prepare selection bitmap for index
- const RoaringBitmap32* selection = nullptr;
- if (auto* bitmap_file_index =
dynamic_cast<BitmapIndexResult*>(file_index_result.get())) {
- PAIMON_ASSIGN_OR_RAISE(selection, bitmap_file_index->GetBitmap());
- }
-
- // narrow the selection to the file-local row positions of an indexed split
- std::optional<RoaringBitmap32> ranges_selection;
- if (ranges != std::nullopt) {
- RoaringBitmap32 ranges_bitmap;
- for (const Range& range : ranges.value()) {
- ranges_bitmap.AddRange(static_cast<int32_t>(range.from),
- static_cast<int32_t>(range.to + 1));
- }
- if (selection != nullptr) {
- ranges_selection = RoaringBitmap32::And(*selection, ranges_bitmap);
- } else {
- ranges_selection = std::move(ranges_bitmap);
- }
- selection = &ranges_selection.value();
- }
-
- // prepare deletion bitmap for deletion vector
- std::shared_ptr<DeletionVector> deletion_vector;
- if (dv_factory) {
- PAIMON_ASSIGN_OR_RAISE(deletion_vector, dv_factory(file->file_name));
- }
- const RoaringBitmap32* deletion = nullptr;
- if (auto* bitmap_dv =
dynamic_cast<BitmapDeletionVector*>(deletion_vector.get())) {
- deletion = bitmap_dv->GetBitmap();
- }
-
- // merge deletion and bitmap index selection
- std::optional<RoaringBitmap32> actual_selection;
- if (selection && deletion) {
- actual_selection = RoaringBitmap32::AndNot(*selection, *deletion);
- } else if (selection) {
- actual_selection = *selection;
- } else if (deletion) {
- actual_selection = *deletion;
- PAIMON_ASSIGN_OR_RAISE(uint64_t num_rows,
file_reader->GetNumberOfRows());
- actual_selection.value().Flip(0, num_rows);
- }
-
- if (actual_selection && actual_selection.value().IsEmpty()) {
- return std::unique_ptr<FileBatchReader>();
- }
-
- ::ArrowSchema c_read_schema;
- PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema,
&c_read_schema));
- PAIMON_RETURN_NOT_OK(file_reader->SetReadSchema(&c_read_schema, predicate,
actual_selection));
-
- std::unique_ptr<FileBatchReader> reader;
- if (!file_reader->SupportPreciseBitmapSelection() && actual_selection) {
- // Some formats (for example blob) return an accurate batch result,
where
- // ApplyBitmapIndexBatchReader is not necessary
- reader =
std::make_unique<ApplyBitmapIndexBatchReader>(std::move(file_reader),
-
std::move(actual_selection).value());
- } else {
- reader = std::move(file_reader);
- }
-
- if (deletion_vector && !deletion && !deletion_vector->IsEmpty()) {
- // TODO(xinyu.lxy): if deletion vector is bitmap64, use
ApplyBitmapIndexBatchReader to
- // filter result
- return Status::NotImplemented("Only support BitmapDeletionVector");
- }
- return std::move(reader);
-}
-
} // namespace paimon
diff --git a/src/paimon/core/operation/raw_file_split_read.h
b/src/paimon/core/operation/raw_file_split_read.h
index 646f24ac..537232f0 100644
--- a/src/paimon/core/operation/raw_file_split_read.h
+++ b/src/paimon/core/operation/raw_file_split_read.h
@@ -83,14 +83,6 @@ class RawFileSplitRead : public AbstractSplitRead {
const std::optional<std::vector<Range>>& local_row_ranges);
Result<bool> Match(const std::shared_ptr<Split>& split, bool
force_keep_delete) const override;
-
- Result<std::unique_ptr<FileBatchReader>> ApplyIndexAndDvReaderIfNeeded(
- std::unique_ptr<FileBatchReader>&& file_reader, const
std::shared_ptr<DataFileMeta>& file,
- const std::shared_ptr<arrow::Schema>& data_schema,
- const std::shared_ptr<arrow::Schema>& read_schema,
- const std::shared_ptr<Predicate>& predicate, DeletionVector::Factory
dv_factory,
- const std::optional<std::vector<Range>>& ranges,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory)
const override;
};
} // namespace paimon
diff --git a/test/inte/realtime_write_inte_test.cpp
b/test/inte/realtime_write_inte_test.cpp
index 569a9d03..ee598ea4 100644
--- a/test/inte/realtime_write_inte_test.cpp
+++ b/test/inte/realtime_write_inte_test.cpp
@@ -2660,6 +2660,88 @@ TEST_F(RealtimeWriteInteTest,
TestAppendScanKeepsDiskSplitsIndependent) {
ASSERT_OK(PrepareAndClose(writer.get()));
}
+TEST_F(RealtimeWriteInteTest,
TestAppendRealtimeReadUsesEmbeddedAndExternalBitmapIndexes) {
+ options_["file-index.bitmap.columns"] = "payload";
+ options_[Options::FILE_INDEX_IN_MANIFEST_THRESHOLD] = "1MB";
+ options_[Options::REALTIME_STORE_STATS_MODE] = "full";
+ CreateTable(/*partition_keys=*/{});
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
+ RealtimeContext::Create());
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> embedded_writer,
+ CreateRealtimeWriter(realtime_context));
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> embedded_batch,
+ MakeBatch(
+ {{1, "embedded-one", "p0"}, {2, "embedded-match", "p0"}, {3,
"embedded-three", "p0"}},
+ /*partitioned=*/false));
+ ASSERT_OK(embedded_writer->Write(std::move(embedded_batch)));
+ ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress> embedded_progress,
+
embedded_writer->PrepareCommitWithProgress(/*commit_identifier=*/0));
+ std::vector<std::shared_ptr<DataFileMeta>> embedded_files =
NewFiles(embedded_progress);
+ ASSERT_EQ(1, embedded_files.size());
+ ASSERT_NE(nullptr, embedded_files[0]->embedded_index);
+ ASSERT_TRUE(embedded_files[0]->extra_files.empty());
+ ASSERT_OK_AND_ASSIGN(int64_t embedded_snapshot_id,
+ Commit(embedded_progress, /*commit_identifier=*/0));
+ ASSERT_OK(embedded_writer->RefreshCommittedSnapshot(embedded_snapshot_id));
+ ASSERT_OK(embedded_writer->Close());
+
+ options_[Options::FILE_INDEX_IN_MANIFEST_THRESHOLD] = "1B";
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> external_writer,
+ CreateRealtimeWriter(realtime_context));
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> external_batch,
+ MakeBatch(
+ {{4, "external-four", "p0"}, {5, "external-match", "p0"}, {6,
"external-six", "p0"}},
+ /*partitioned=*/false));
+ ASSERT_OK(external_writer->Write(std::move(external_batch)));
+ ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress> external_progress,
+
external_writer->PrepareCommitWithProgress(/*commit_identifier=*/1));
+ std::vector<std::shared_ptr<DataFileMeta>> external_files =
NewFiles(external_progress);
+ ASSERT_EQ(1, external_files.size());
+ ASSERT_EQ(nullptr, external_files[0]->embedded_index);
+ ASSERT_EQ(1, external_files[0]->extra_files.size());
+ ASSERT_TRUE(external_files[0]->extra_files[0].has_value());
+ std::string external_index_path =
+ PathUtil::JoinPath(table_path_, "bucket-0/" +
external_files[0]->extra_files[0].value());
+ ASSERT_OK_AND_ASSIGN(bool external_index_exists,
+ dir_->GetFileSystem()->Exists(external_index_path));
+ ASSERT_TRUE(external_index_exists);
+ ASSERT_OK_AND_ASSIGN(int64_t external_snapshot_id,
+ Commit(external_progress, /*commit_identifier=*/1));
+ ASSERT_OK(external_writer->RefreshCommittedSnapshot(external_snapshot_id));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> memory_batch,
+ MakeBatch({{7, "memory-seven", "p0"}},
/*partitioned=*/false));
+ ASSERT_OK(external_writer->Write(std::move(memory_batch)));
+
+ const std::string embedded_match = "embedded-match";
+ const std::string external_match = "external-match";
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> predicate,
+ PredicateBuilder::Or(
+ {PredicateBuilder::Equal(
+ /*field_index=*/1, /*field_name=*/"payload",
FieldType::STRING,
+ Literal(FieldType::STRING, embedded_match.data(),
embedded_match.size())),
+ PredicateBuilder::Equal(
+ /*field_index=*/1, /*field_name=*/"payload",
FieldType::STRING,
+ Literal(FieldType::STRING, external_match.data(),
external_match.size()))}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan,
CreatePlan(realtime_context, predicate));
+ bool has_realtime_split = false;
+ for (const std::shared_ptr<Split>& split : plan->Splits()) {
+ has_realtime_split |= std::dynamic_pointer_cast<RealtimeSplit>(split)
!= nullptr;
+ }
+ ASSERT_TRUE(has_realtime_split);
+ // Residual filtering is disabled. Batch statistics remove the
non-matching memory batch, while
+ // the absence of the other disk rows verifies that both bitmap index
storage forms are used.
+ ASSERT_OK_AND_ASSIGN(std::vector<Row> actual_rows, ReadRows(plan,
realtime_context, predicate,
+
/*enable_predicate_filter=*/false));
+ std::sort(actual_rows.begin(), actual_rows.end());
+ ASSERT_EQ((std::vector<Row>{{2, "embedded-match", "p0"}, {5,
"external-match", "p0"}}),
+ actual_rows);
+ ASSERT_OK(PrepareAndClose(external_writer.get()));
+}
+
TEST_F(RealtimeWriteInteTest, TestCommitOrdersPreparedOffsetRanges) {
CreateTable(/*partition_keys=*/{});
ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> writer,
CreateRealtimeWriter());
@@ -2820,6 +2902,210 @@ TEST_F(RealtimeWriteInteTest,
TestRealtimeWriteAcrossAppendCompaction) {
ASSERT_OK(writer->Close());
}
+TEST_F(RealtimeWriteInteTest,
TestPkDvExternalBitmapCanEliminateEntireDiskFile) {
+ options_[Options::FILE_FORMAT] = "parquet";
+ options_[Options::DELETION_VECTORS_ENABLED] = "true";
+ options_[Options::FILE_INDEX_READ_ENABLED] = "true";
+ options_["file-index.bitmap.columns"] = "payload";
+ options_[Options::FILE_INDEX_IN_MANIFEST_THRESHOLD] = "1B";
+ options_[Options::REALTIME_STORE_STATS_MODE] = "full";
+ CreatePkTable();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
+ RealtimeContext::Create());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> writer,
+ CreateRealtimeWriter(realtime_context));
+
+ // "middle" is inside the file's min/max range but absent from its bitmap
index. Keeping the
+ // index external makes scan planning retain the file and lets the merge
reader eliminate it.
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> disk_batch,
+ MakeBatch({{1, "aaa", "p0"}, {2, "zzz", "p0"}},
/*partitioned=*/false));
+ ASSERT_OK(writer->Write(std::move(disk_batch)));
+ ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress> disk_progress,
+
writer->PrepareCommitWithProgress(/*commit_identifier=*/0));
+ ASSERT_OK_AND_ASSIGN(int64_t disk_snapshot_id, Commit(disk_progress,
/*commit_identifier=*/0));
+ ASSERT_OK(writer->RefreshCommittedSnapshot(disk_snapshot_id));
+
+ ASSERT_OK_AND_ASSIGN(Snapshot compact_snapshot,
CompactAndCommit(/*partition=*/{}, /*bucket=*/0,
+
/*commit_identifier=*/1));
+ ASSERT_OK(writer->RefreshCommittedSnapshot(compact_snapshot.Id()));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> memory_batch,
+ MakeBatch({{3, "middle", "p0"}},
/*partitioned=*/false));
+ ASSERT_OK(writer->Write(std::move(memory_batch)));
+
+ const std::string middle = "middle";
+ const std::shared_ptr<Predicate> predicate = PredicateBuilder::Equal(
+ /*field_index=*/1, /*field_name=*/"payload", FieldType::STRING,
+ Literal(FieldType::STRING, middle.data(), middle.size()));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan,
CreatePlan(realtime_context, predicate));
+ ASSERT_EQ(1, plan->Splits().size());
+ std::shared_ptr<RealtimeSplit> realtime_split =
+ std::dynamic_pointer_cast<RealtimeSplit>(plan->Splits()[0]);
+ ASSERT_NE(nullptr, realtime_split);
+
+ size_t external_high_level_file_count = 0;
+ for (const std::shared_ptr<Split>& split : realtime_split->DiskSplits()) {
+ std::shared_ptr<DataSplitImpl> data_split =
std::dynamic_pointer_cast<DataSplitImpl>(split);
+ ASSERT_NE(nullptr, data_split);
+ for (const std::shared_ptr<DataFileMeta>& file :
data_split->DataFiles()) {
+ if (file->level > 0 && file->embedded_index == nullptr &&
!file->extra_files.empty() &&
+ file->extra_files[0].has_value()) {
+ ++external_high_level_file_count;
+ }
+ }
+ }
+ ASSERT_EQ(1, external_high_level_file_count);
+
+ // The external bitmap removes the only disk reader. The in-memory
matching row must still be
+ // returned, and constructing the empty disk run must not rely on
positional file/reader pairs.
+ ASSERT_OK_AND_ASSIGN(std::vector<Row> actual_rows, ReadRows(plan,
realtime_context, predicate,
+
/*enable_predicate_filter=*/false));
+ ASSERT_EQ((std::vector<Row>{{3, "middle", "p0"}}), actual_rows);
+ ASSERT_OK(PrepareAndClose(writer.get()));
+}
+
+TEST_F(RealtimeWriteInteTest,
TestPkDvRealtimeReadUsesEmbeddedAndExternalBitmapIndexes) {
+ options_[Options::FILE_FORMAT] = "parquet";
+ options_[Options::FILE_COMPRESSION] = "none";
+ options_[Options::DELETION_VECTORS_ENABLED] = "true";
+ options_["file-index.bitmap.columns"] = "payload";
+ options_[Options::FILE_INDEX_IN_MANIFEST_THRESHOLD] = "1MB";
+ options_[Options::REALTIME_STORE_STATS_MODE] = "full";
+ CreatePkTable();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
+ RealtimeContext::Create());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> embedded_writer,
+ CreateRealtimeWriter(realtime_context));
+
+ // Keep the max-level file substantially larger than the later L0 files.
Otherwise the
+ // size-ratio picker may rewrite all levels together instead of producing
a deletion vector.
+ // Put the padding in a non-indexed column so the data file is much larger
than both L0 files
+ // without making the embedded bitmap index itself large.
+ const std::string first_padding(256 * 1024, 'X');
+ const std::string third_padding(256 * 1024, 'Y');
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> base_batch,
+ MakeBatch({{1, "high-one", first_padding},
+ {2, "high-match", "p0"},
+ {3, "high-old-three", third_padding}},
+ /*partitioned=*/false));
+ ASSERT_OK(embedded_writer->Write(std::move(base_batch)));
+ ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress> base_progress,
+
embedded_writer->PrepareCommitWithProgress(/*commit_identifier=*/0));
+ std::vector<std::shared_ptr<DataFileMeta>> base_files =
NewFiles(base_progress);
+ ASSERT_EQ(1, base_files.size());
+ ASSERT_NE(nullptr, base_files[0]->embedded_index);
+ ASSERT_TRUE(base_files[0]->extra_files.empty());
+ ASSERT_OK_AND_ASSIGN(int64_t base_snapshot_id, Commit(base_progress,
/*commit_identifier=*/0));
+ ASSERT_OK(embedded_writer->RefreshCommittedSnapshot(base_snapshot_id));
+
+ ASSERT_OK_AND_ASSIGN(Snapshot full_compact_snapshot,
+ CompactAndCommit(/*partition=*/{}, /*bucket=*/0,
+ /*commit_identifier=*/1));
+
ASSERT_OK(embedded_writer->RefreshCommittedSnapshot(full_compact_snapshot.Id()));
+ ASSERT_OK(embedded_writer->Close());
+
+ options_[Options::FILE_INDEX_IN_MANIFEST_THRESHOLD] = "1B";
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> external_writer,
+ CreateRealtimeWriter(realtime_context));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> external_batch,
+ MakeBatch({{3, "external-current-three", "p0"},
+ {4, "external-four", "p0"},
+ {5, "external-match", "p0"}},
+ /*partitioned=*/false, /*bucket=*/0,
+ {RecordBatch::RowKind::UPDATE_AFTER,
+ RecordBatch::RowKind::INSERT,
RecordBatch::RowKind::INSERT}));
+ ASSERT_OK(external_writer->Write(std::move(external_batch)));
+ ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress> external_progress,
+
external_writer->PrepareCommitWithProgress(/*commit_identifier=*/2));
+ std::vector<std::shared_ptr<DataFileMeta>> external_files =
NewFiles(external_progress);
+ ASSERT_EQ(1, external_files.size());
+ ASSERT_EQ(0, external_files[0]->level);
+ ASSERT_EQ(nullptr, external_files[0]->embedded_index);
+ ASSERT_EQ(1, external_files[0]->extra_files.size());
+ ASSERT_TRUE(external_files[0]->extra_files[0].has_value());
+ std::string external_index_path =
+ PathUtil::JoinPath(table_path_, "bucket-0/" +
external_files[0]->extra_files[0].value());
+ ASSERT_OK_AND_ASSIGN(bool external_index_exists,
+ dir_->GetFileSystem()->Exists(external_index_path));
+ ASSERT_TRUE(external_index_exists);
+ ASSERT_OK_AND_ASSIGN(int64_t external_snapshot_id,
+ Commit(external_progress, /*commit_identifier=*/2));
+ ASSERT_OK(external_writer->RefreshCommittedSnapshot(external_snapshot_id));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> second_external_batch,
+ MakeBatch({{6, "external-six", "p0"}},
/*partitioned=*/false));
+ ASSERT_OK(external_writer->Write(std::move(second_external_batch)));
+ ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress>
second_external_progress,
+
external_writer->PrepareCommitWithProgress(/*commit_identifier=*/3));
+ ASSERT_OK_AND_ASSIGN(int64_t second_external_snapshot_id,
+ Commit(second_external_progress,
/*commit_identifier=*/3));
+
ASSERT_OK(external_writer->RefreshCommittedSnapshot(second_external_snapshot_id));
+
+ ASSERT_OK_AND_ASSIGN(Snapshot dv_compact_snapshot,
+ CompactAndCommit(/*partition=*/{}, /*bucket=*/0,
+ /*commit_identifier=*/4,
+ /*full_compaction=*/false));
+
ASSERT_OK(external_writer->RefreshCommittedSnapshot(dv_compact_snapshot.Id()));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> memory_batch,
+ MakeBatch({{7, "memory-seven", "p0"}},
/*partitioned=*/false));
+ ASSERT_OK(external_writer->Write(std::move(memory_batch)));
+
+ const std::string high_match = "high-match";
+ const std::string external_match = "external-match";
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> predicate,
+ PredicateBuilder::Or(
+ {PredicateBuilder::Equal(
+ /*field_index=*/1, /*field_name=*/"payload",
FieldType::STRING,
+ Literal(FieldType::STRING, high_match.data(),
high_match.size())),
+ PredicateBuilder::Equal(
+ /*field_index=*/1, /*field_name=*/"payload",
FieldType::STRING,
+ Literal(FieldType::STRING, external_match.data(),
external_match.size()))}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan,
CreatePlan(realtime_context, predicate));
+ ASSERT_EQ(1, plan->Splits().size());
+ std::shared_ptr<RealtimeSplit> realtime_split =
+ std::dynamic_pointer_cast<RealtimeSplit>(plan->Splits()[0]);
+ ASSERT_NE(nullptr, realtime_split);
+ ASSERT_OK_AND_ASSIGN(DiskFileLayout layout,
InspectDiskFiles(realtime_split->DiskSplits()));
+ ASSERT_TRUE(layout.has_high_level_file);
+ ASSERT_TRUE(layout.has_high_level_deletion_vector);
+ bool has_embedded_high_level_index_with_dv = false;
+ bool has_external_high_level_index = false;
+ for (const std::shared_ptr<Split>& split : realtime_split->DiskSplits()) {
+ std::shared_ptr<DataSplitImpl> data_split =
std::dynamic_pointer_cast<DataSplitImpl>(split);
+ ASSERT_NE(nullptr, data_split);
+ const std::vector<std::shared_ptr<DataFileMeta>>& files =
data_split->DataFiles();
+ const std::vector<std::optional<DeletionFile>>& deletion_files =
+ data_split->DeletionFiles();
+ for (size_t i = 0; i < files.size(); ++i) {
+ has_embedded_high_level_index_with_dv |=
+ files[i]->level > 0 && files[i]->embedded_index != nullptr &&
+ !deletion_files.empty() && deletion_files[i].has_value();
+ if (files[i]->level > 0 && files[i]->embedded_index == nullptr &&
+ !files[i]->extra_files.empty() &&
files[i]->extra_files[0].has_value()) {
+ has_external_high_level_index = true;
+ std::string index_path =
+ PathUtil::JoinPath(data_split->BucketPath(),
files[i]->extra_files[0].value());
+ ASSERT_OK_AND_ASSIGN(bool index_exists,
dir_->GetFileSystem()->Exists(index_path));
+ ASSERT_TRUE(index_exists);
+ }
+ }
+ }
+ ASSERT_TRUE(has_embedded_high_level_index_with_dv);
+ ASSERT_TRUE(has_external_high_level_index);
+
+ // Only primary-key predicates are sent to L0 and memory in PK MOR reads.
The non-key predicate
+ // still reaches both high-level bitmap indexes, while the unfiltered
memory row remains.
+ ASSERT_OK_AND_ASSIGN(std::vector<Row> actual_rows, ReadRows(plan,
realtime_context, predicate,
+
/*enable_predicate_filter=*/false));
+ std::sort(actual_rows.begin(), actual_rows.end());
+ ASSERT_EQ((std::vector<Row>{
+ {2, "high-match", "p0"}, {5, "external-match", "p0"}, {7,
"memory-seven", "p0"}}),
+ actual_rows);
+ ASSERT_OK(PrepareAndClose(external_writer.get()));
+}
+
TEST_F(RealtimeWriteInteTest, TestPkDvPredicateAcrossHighLevelLevel0AndMemory)
{
options_[Options::FILE_FORMAT] = "parquet";
options_[Options::DELETION_VECTORS_ENABLED] = "true";
diff --git a/test/inte/scan_and_read_inte_test.cpp
b/test/inte/scan_and_read_inte_test.cpp
index 96f528bf..a8e3a6ed 100644
--- a/test/inte/scan_and_read_inte_test.cpp
+++ b/test/inte/scan_and_read_inte_test.cpp
@@ -3123,6 +3123,167 @@ TEST_P(ScanAndReadInteTest,
TestScanAndReadWithDisableIndex) {
ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString();
}
+TEST(ScanAndReadFileIndexInteTest, TestPkDvEmbeddedBitmapFiltersScanAndRead) {
+ std::unique_ptr<UniqueTestDirectory> dir =
UniqueTestDirectory::Create("local");
+ ASSERT_NE(nullptr, dir);
+ const std::string table_path = PathUtil::JoinPath(dir->Str(),
"foo.db/bar");
+ const arrow::FieldVector fields = {arrow::field("id", arrow::int32()),
+ arrow::field("indexed_value",
arrow::int32()),
+ arrow::field("payload", arrow::utf8())};
+ const std::shared_ptr<arrow::Schema> schema = arrow::schema(fields);
+ const std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::BUCKET, "1"},
+ {Options::BUCKET_KEY, "id"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::COMMIT_FORCE_COMPACT, "true"},
+ {Options::FILE_INDEX_READ_ENABLED, "true"},
+ {Options::FILE_INDEX_IN_MANIFEST_THRESHOLD, "1MB"},
+ {"file-index.bitmap.columns", "indexed_value"},
+ };
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<TestHelper> helper,
+ TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{},
/*primary_keys=*/{"id"},
+ options, /*is_streaming_mode=*/true,
/*ignore_if_exists=*/false,
+ PathUtil::JoinPath(dir->Str(), "tmp")));
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields),
+ R"([[1, 10, "one"], [2, 20, "two"], [3,
30, "three"]])",
+ /*partition_map=*/{}, /*bucket=*/0,
/*row_kinds=*/{}));
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<std::shared_ptr<CommitMessage>> commit_messages,
+ helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0,
std::nullopt));
+ ASSERT_FALSE(commit_messages.empty());
+
+ // 15 falls inside the file's [10, 30] min/max range, so statistics alone
must retain the
+ // file. Disabling file-index reads verifies that baseline explicitly.
+ const std::shared_ptr<Predicate> absent_predicate =
PredicateBuilder::Equal(
+ /*field_index=*/1, /*field_name=*/"indexed_value", FieldType::INT,
Literal(15));
+ ScanContextBuilder stats_only_scan_builder(table_path);
+ stats_only_scan_builder.SetOptions(options)
+ .AddOption(Options::FILE_INDEX_READ_ENABLED, "false")
+ .SetPredicate(absent_predicate);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> stats_only_scan_context,
+ stats_only_scan_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> stats_only_scan,
+
TableScan::Create(std::move(stats_only_scan_context)));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> stats_only_plan,
stats_only_scan->CreatePlan());
+ ASSERT_FALSE(stats_only_plan->Splits().empty());
+
+ // The embedded bitmap knows that 15 is absent and eliminates the file
during scan planning.
+ ScanContextBuilder indexed_scan_builder(table_path);
+ indexed_scan_builder.SetOptions(options).SetPredicate(absent_predicate);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> indexed_scan_context,
+ indexed_scan_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> indexed_scan,
+ TableScan::Create(std::move(indexed_scan_context)));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> indexed_plan,
indexed_scan->CreatePlan());
+ ASSERT_TRUE(indexed_plan->Splits().empty());
+
+ // For a present value, scan retains the embedded-index file and read
applies its precise
+ // bitmap. Residual predicate filtering stays disabled so only file-index
selection can remove
+ // the other rows.
+ const std::shared_ptr<Predicate> present_predicate =
PredicateBuilder::Equal(
+ /*field_index=*/1, /*field_name=*/"indexed_value", FieldType::INT,
Literal(20));
+ ScanContextBuilder retained_scan_builder(table_path);
+ retained_scan_builder.SetOptions(options).SetPredicate(present_predicate);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> retained_scan_context,
+ retained_scan_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> retained_scan,
+ TableScan::Create(std::move(retained_scan_context)));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> retained_plan,
retained_scan->CreatePlan());
+ ASSERT_FALSE(retained_plan->Splits().empty());
+
+ ReadContextBuilder read_context_builder(table_path);
+ read_context_builder.SetOptions(options)
+ .SetPredicate(present_predicate)
+ .EnablePredicateFilter(false);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context,
read_context_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read,
+ TableRead::Create(std::move(read_context)));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> batch_reader,
+ table_read->CreateReader(retained_plan->Splits()));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual,
+
ReadResultCollector::CollectResult(std::move(batch_reader)));
+ const std::shared_ptr<arrow::DataType> result_type = arrow::struct_(
+ {arrow::field("_VALUE_KIND", arrow::int8()), arrow::field("id",
arrow::int32()),
+ arrow::field("indexed_value", arrow::int32()),
arrow::field("payload", arrow::utf8())});
+ const std::shared_ptr<arrow::ChunkedArray> expected =
std::make_shared<arrow::ChunkedArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(result_type, R"([[0, 2, 20,
"two"]])")
+ .ValueOrDie());
+ ASSERT_TRUE(expected->Equals(actual)) << actual->ToString();
+}
+
+TEST(ScanAndReadFileIndexInteTest,
TestPkDvReconstructsFileIndexPredicateAfterSchemaEvolution) {
+ std::unique_ptr<UniqueTestDirectory> dir =
UniqueTestDirectory::Create("local");
+ ASSERT_NE(nullptr, dir);
+ const std::string table_path = PathUtil::JoinPath(dir->Str(),
"foo.db/bar");
+ const arrow::FieldVector fields = {arrow::field("id", arrow::int32()),
+ arrow::field("indexed_value",
arrow::int32()),
+ arrow::field("payload", arrow::utf8())};
+ const std::shared_ptr<arrow::Schema> schema = arrow::schema(fields);
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::BUCKET, "1"},
+ {Options::BUCKET_KEY, "id"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::COMMIT_FORCE_COMPACT, "true"},
+ {Options::FILE_INDEX_READ_ENABLED, "true"},
+ {Options::FILE_INDEX_IN_MANIFEST_THRESHOLD, "1MB"},
+ {"file-index.bitmap.columns", "indexed_value"},
+ };
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<TestHelper> helper,
+ TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{},
/*primary_keys=*/{"id"},
+ options, /*is_streaming_mode=*/true,
/*ignore_if_exists=*/false,
+ PathUtil::JoinPath(dir->Str(), "tmp")));
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields),
+ R"([[1, 10, "one"], [2, 20, "two"], [3,
30, "three"]])",
+ /*partition_map=*/{}, /*bucket=*/0,
/*row_kinds=*/{}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(batch),
/*commit_identifier=*/0, std::nullopt));
+ helper.reset();
+
+ // Keep field IDs and types unchanged while renaming the indexed value
field. The old file and
+ // its embedded bitmap still use "indexed_value", whereas scans use
"renamed_value".
+ options["file-index.bitmap.columns"] = "renamed_value";
+ const std::vector<DataField> evolved_fields = {
+ DataField(0, arrow::field("id", arrow::int32(), /*nullable=*/false)),
+ DataField(1, arrow::field("renamed_value", arrow::int32())),
+ DataField(2, arrow::field("payload", arrow::utf8()))};
+ ASSERT_OK(TestHelper::WriteNextSchema(dir->GetFileSystem(), table_path,
evolved_fields,
+ /*highest_field_id=*/2, options));
+
+ const std::shared_ptr<Predicate> predicate = PredicateBuilder::Equal(
+ /*field_index=*/1, /*field_name=*/"renamed_value", FieldType::INT,
Literal(15));
+
+ // The evolved min/max range is still [10, 30], so statistics alone retain
the old file.
+ ScanContextBuilder stats_only_scan_builder(table_path);
+ stats_only_scan_builder.SetOptions(options)
+ .AddOption(Options::FILE_INDEX_READ_ENABLED, "false")
+ .SetPredicate(predicate);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> stats_only_scan_context,
+ stats_only_scan_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> stats_only_scan,
+
TableScan::Create(std::move(stats_only_scan_context)));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> stats_only_plan,
stats_only_scan->CreatePlan());
+ ASSERT_FALSE(stats_only_plan->Splits().empty());
+
+ // Reconstructing the predicate to schema-0 lets its embedded bitmap prove
that 15 is absent.
+ ScanContextBuilder indexed_scan_builder(table_path);
+ indexed_scan_builder.SetOptions(options).SetPredicate(predicate);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> indexed_scan_context,
+ indexed_scan_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> indexed_scan,
+ TableScan::Create(std::move(indexed_scan_context)));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> indexed_plan,
indexed_scan->CreatePlan());
+ ASSERT_TRUE(indexed_plan->Splits().empty());
+}
+
TEST_P(ScanAndReadInteTest, TestPkDvTableIndexInDataAndWithExternalPath) {
auto file_format = FileFormat();
std::string table_path =
