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 7ff18554 feat(read): support file index and predicate pushdown for
data evolution (#215)
7ff18554 is described below
commit 7ff18554271c354601d8f30d7ae32db98e11b02c
Author: lxy <[email protected]>
AuthorDate: Wed Aug 19 19:48:34 2026 +0800
feat(read): support file index and predicate pushdown for data evolution
(#215)
---
.../core/operation/data_evolution_split_read.cpp | 139 ++++++++++++-
.../core/operation/data_evolution_split_read.h | 22 ++-
.../operation/data_evolution_split_read_test.cpp | 27 +++
test/inte/data_evolution_table_test.cpp | 214 +++++++++++++++++----
4 files changed, 358 insertions(+), 44 deletions(-)
diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp
b/src/paimon/core/operation/data_evolution_split_read.cpp
index 52a1bb51..9e71fe47 100644
--- a/src/paimon/core/operation/data_evolution_split_read.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read.cpp
@@ -22,6 +22,7 @@
#include <cassert>
#include <limits>
#include <map>
+#include <set>
#include <string>
#include <string_view>
#include <thread>
@@ -53,8 +54,13 @@
#include "paimon/core/core_options.h"
#include "paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h"
#include "paimon/core/global_index/indexed_split_impl.h"
+#include "paimon/core/io/file_index_evaluator.h"
#include "paimon/core/utils/blob_view_lookup.h"
#include "paimon/core/utils/data_evolution_utils.h"
+#include "paimon/core/utils/field_mapping.h"
+#include "paimon/file_index/bitmap_index_result.h"
+#include "paimon/file_index/file_index_result.h"
+#include "paimon/predicate/predicate_utils.h"
namespace paimon {
namespace {
@@ -389,12 +395,22 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::InnerCreateReader(
path_factory_->CreateDataFilePathFactory(split_impl->Partition(),
split_impl->Bucket()));
auto metas = split_impl->DataFiles();
DeletionVector::Factory split_dv_factory =
CreateSplitDvFactory(*split_impl);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Predicate> push_down_predicate,
+ CreatePushDownPredicate(context_->GetPredicate(),
raw_read_schema_));
PAIMON_ASSIGN_OR_RAISE(std::vector<std::vector<std::shared_ptr<DataFileMeta>>>
split_by_row_id,
MergeRangesAndSort(std::move(metas)));
std::vector<std::unique_ptr<BatchReader>> sub_readers;
for (const std::vector<std::shared_ptr<DataFileMeta>>& need_merge_files :
split_by_row_id) {
+ if (need_merge_files.size() > 1) {
+ PAIMON_ASSIGN_OR_RAISE(
+ bool skip_group,
+ SkipByFileIndex(push_down_predicate, need_merge_files,
data_file_path_factory));
+ if (skip_group) {
+ continue;
+ }
+ }
PAIMON_ASSIGN_OR_RAISE(std::optional<GroupDeletionVector> group_dv,
ReadGroupDeletionVector(need_merge_files,
split_dv_factory));
PAIMON_ASSIGN_OR_RAISE(DeletionVector::Factory group_dv_factory,
@@ -404,10 +420,15 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::InnerCreateReader(
PAIMON_ASSIGN_OR_RAISE(
std::vector<std::unique_ptr<FileBatchReader>> raw_file_readers,
CreateRawFileReaders(split_impl->Partition(),
need_merge_files, raw_read_schema_,
- /*predicate=*/nullptr, group_dv_factory,
row_ranges,
+ push_down_predicate, group_dv_factory,
row_ranges,
data_file_path_factory,
/*extra_format_options=*/{}));
- assert(raw_file_readers.size() == 1);
+ if (raw_file_readers.empty()) {
+ continue;
+ }
+ if (raw_file_readers.size() != 1) {
+ return Status::Invalid("Single-file data evolution group
created multiple readers");
+ }
sub_readers.push_back(std::move(raw_file_readers[0]));
} else {
PAIMON_ASSIGN_OR_RAISE(
@@ -424,17 +445,110 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::InnerCreateReader(
return
std::make_unique<CompleteRowKindBatchReader>(std::move(batch_reader), pool_);
}
+Result<std::shared_ptr<Predicate>>
DataEvolutionSplitRead::CreatePushDownPredicate(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::shared_ptr<arrow::Schema>& read_schema) {
+ std::map<std::string, int32_t> picked_field_name_to_idx;
+ for (int32_t i = 0; i < read_schema->num_fields(); ++i) {
+ const std::string& field_name = read_schema->field(i)->name();
+ if (!SpecialFields::IsSystemField(field_name)) {
+ picked_field_name_to_idx.emplace(field_name, i);
+ }
+ }
+ return PredicateUtils::CreatePickedFieldFilter(predicate,
picked_field_name_to_idx);
+}
+
+Result<bool> DataEvolutionSplitRead::SkipByFileIndex(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
+ if (!options_.FileIndexReadEnabled() || !predicate) {
+ return false;
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<FieldMappingBuilder> field_mapping_builder,
+ FieldMappingBuilder::Create(raw_read_schema_,
context_->GetPartitionKeys(), predicate));
+ std::set<int32_t> claimed_field_ids;
+ for (const auto& file : files) {
+ // Blob and vector-store files may cover only part of the row range,
so their indexes
+ // cannot prove that the complete merged group misses the predicate.
+ if (!DataEvolutionUtils::IsNormalFile(file->file_name)) {
+ continue;
+ }
+
+ std::shared_ptr<TableSchema> data_schema = context_->GetTableSchema();
+ if (file->schema_id != data_schema->Id()) {
+ PAIMON_ASSIGN_OR_RAISE(data_schema,
schema_manager_->ReadSchema(file->schema_id));
+ }
+ std::vector<DataField> written_fields;
+ if (file->write_cols) {
+ std::vector<std::string> data_write_cols;
+ data_write_cols.reserve(file->write_cols->size());
+ for (const auto& write_col : file->write_cols.value()) {
+ if (!SpecialFields::IsSystemField(write_col)) {
+ data_write_cols.push_back(write_col);
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE(written_fields,
data_schema->GetFields(data_write_cols));
+ } else {
+ written_fields = data_schema->Fields();
+ }
+
+ std::set<std::string> overwritten_field_names;
+ for (const auto& field : written_fields) {
+ if (!claimed_field_ids.insert(field.Id()).second) {
+ overwritten_field_names.insert(field.Name());
+ }
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FieldMapping> field_mapping,
+
field_mapping_builder->CreateFieldMapping(written_fields));
+ std::shared_ptr<Predicate> data_predicate =
+ field_mapping->non_partition_info.non_partition_filter;
+ if (!overwritten_field_names.empty()) {
+ PAIMON_ASSIGN_OR_RAISE(data_predicate,
PredicateUtils::ExcludePredicateWithFields(
+ data_predicate,
overwritten_field_names));
+ }
+ if (!data_predicate) {
+ continue;
+ }
+
+ auto written_schema =
DataField::ConvertDataFieldsToArrowSchema(written_fields);
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileIndexResult> index_result,
+ FileIndexEvaluator::Evaluate(written_schema, data_predicate,
data_file_path_factory,
+ file, options_.GetFileSystem(),
pool_));
+ PAIMON_ASSIGN_OR_RAISE(bool is_remain, index_result->IsRemain());
+ if (!is_remain) {
+ return true;
+ }
+ }
+ return false;
+}
+
Result<std::unique_ptr<FileBatchReader>>
DataEvolutionSplitRead::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 {
- if (predicate) {
- assert(false);
- // as DataEvolutionSplitRead will skip predicate
- return Status::Invalid("DataEvolutionSplitRead do not support
predicate");
+ 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>();
+ }
+ }
+ const RoaringBitmap32* index_selection = nullptr;
+ if (auto* bitmap_index =
dynamic_cast<BitmapIndexResult*>(file_index_result.get())) {
+ PAIMON_ASSIGN_OR_RAISE(index_selection, bitmap_index->GetBitmap());
}
+
// the factory is per row range group and already returns a view taking
file-local positions.
// Unlike RawFileSplitRead the vector is not folded into the format
reader's selection: it is
// no BitmapDeletionVector, and the blob fallback path's gap segments have
no format reader.
@@ -444,10 +558,19 @@ Result<std::unique_ptr<FileBatchReader>>
DataEvolutionSplitRead::ApplyIndexAndDv
}
PAIMON_ASSIGN_OR_RAISE(std::optional<RoaringBitmap32> selection_row_ids,
file->ToFileSelection(row_ranges));
+ if (index_selection) {
+ if (selection_row_ids) {
+ selection_row_ids.value() &= *index_selection;
+ } else {
+ selection_row_ids = *index_selection;
+ }
+ }
+ if (selection_row_ids && selection_row_ids->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=*/nullptr,
selection_row_ids));
+ PAIMON_RETURN_NOT_OK(file_reader->SetReadSchema(&c_read_schema, predicate,
selection_row_ids));
std::unique_ptr<FileBatchReader> reader;
if (!file_reader->SupportPreciseBitmapSelection() && selection_row_ids) {
diff --git a/src/paimon/core/operation/data_evolution_split_read.h
b/src/paimon/core/operation/data_evolution_split_read.h
index 983ca29d..94a59fa0 100644
--- a/src/paimon/core/operation/data_evolution_split_read.h
+++ b/src/paimon/core/operation/data_evolution_split_read.h
@@ -68,9 +68,10 @@ struct DeletionFile;
/// ->(DelegatingPrefetchReader)->(PrefetchFileBatchReader)->FormatReader
///
///
-/// A union `SplitRead` to read multiple inner files to merge columns, note
that this class
-/// does not support filtering push down: a predicate would have to be
evaluated consistently
-/// across the files being merged, which is not implemented here.
+/// A union `SplitRead` to read multiple inner files to merge columns. A
single-file row range
+/// group gets both file-index and format-level predicate pushdown. A merged
group only uses file
+/// indexes to skip the whole group: filtering its child readers independently
would break their
+/// positional alignment.
///
/// Deletion vectors are supported: a row range group's vector is maintained
against the
/// group's anchor file (DataEvolutionUtils::RetrieveAnchorFile), so its
positions are
@@ -172,6 +173,21 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
const std::shared_ptr<DataSplit>& data_split,
const std::optional<std::vector<Range>>& row_ranges) const;
+ /// Keeps top-level conjuncts whose fields all belong to `read_schema`,
excluding conjuncts
+ /// over system fields. The returned predicate is for pushdown only; the
original predicate is
+ /// still evaluated as a residual filter when requested by the read
context.
+ static Result<std::shared_ptr<Predicate>> CreatePushDownPredicate(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::shared_ptr<arrow::Schema>& read_schema);
+
+ /// Returns true when file indexes prove that no row in a merged row range
group can match.
+ /// Only normal files are considered, and an older copy of a field is
excluded after a newer
+ /// file has claimed the same field id.
+ Result<bool> SkipByFileIndex(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::vector<std::shared_ptr<DataFileMeta>>& files,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory)
const;
+
/// Builds the deletion vector factory over the split's deletion files,
keyed by data file
/// name. Only anchor files carry one. Returns a null factory when the
split has none.
DeletionVector::Factory CreateSplitDvFactory(const DataSplitImpl&
split_impl) const;
diff --git a/src/paimon/core/operation/data_evolution_split_read_test.cpp
b/src/paimon/core/operation/data_evolution_split_read_test.cpp
index 809933e9..03bc95b6 100644
--- a/src/paimon/core/operation/data_evolution_split_read_test.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read_test.cpp
@@ -25,6 +25,7 @@
#include "gtest/gtest.h"
#include "paimon/common/data/binary_row.h"
+#include "paimon/common/table/special_fields.h"
#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/io/data_file_meta.h"
#include "paimon/core/manifest/file_source.h"
@@ -37,6 +38,8 @@
#include "paimon/executor.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/memory/memory_pool.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/predicate/predicate_builder.h"
#include "paimon/read_context.h"
#include "paimon/status.h"
#include "paimon/testing/utils/testharness.h"
@@ -102,6 +105,30 @@ class DataEvolutionSplitReadTest : public ::testing::Test {
std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
};
+TEST_F(DataEvolutionSplitReadTest, TestCreatePushDownPredicate) {
+ auto f0_predicate =
+ PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0",
FieldType::INT, Literal(1));
+ auto f1_predicate =
+ PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"f1",
FieldType::INT, Literal(2));
+ auto row_id_predicate = PredicateBuilder::Equal(
+ /*field_index=*/2, SpecialFields::RowId().Name(), FieldType::BIGINT,
Literal(3l));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> predicate,
+ PredicateBuilder::And({f0_predicate, f1_predicate,
row_id_predicate}));
+
+ auto read_schema = DataField::ConvertDataFieldsToArrowSchema(
+ {DataField(0, arrow::field("f0", arrow::int32())),
SpecialFields::RowId()});
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> push_down,
+
DataEvolutionSplitRead::CreatePushDownPredicate(predicate, read_schema));
+ ASSERT_TRUE(push_down);
+ ASSERT_EQ(*push_down, *f0_predicate);
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> or_predicate,
+ PredicateBuilder::Or({f0_predicate, f1_predicate}));
+ ASSERT_OK_AND_ASSIGN(
+ push_down,
DataEvolutionSplitRead::CreatePushDownPredicate(or_predicate, read_schema));
+ ASSERT_FALSE(push_down);
+}
+
TEST_F(DataEvolutionSplitReadTest, TestAddSingleBlobEntry) {
auto blob_entry =
CreateBlobFile("blob1", /*first_row_id=*/0, /*row_count=*/100,
diff --git a/test/inte/data_evolution_table_test.cpp
b/test/inte/data_evolution_table_test.cpp
index 6232c1e7..13fede80 100644
--- a/test/inte/data_evolution_table_test.cpp
+++ b/test/inte/data_evolution_table_test.cpp
@@ -400,10 +400,11 @@ class DataEvolutionTableTest : public ::testing::Test,
const std::shared_ptr<arrow::StructArray>&
expected_array,
const std::shared_ptr<Predicate>& predicate = nullptr,
const std::vector<Range>& row_ranges = {},
- bool check_scan_plan_when_empty_result = true) const {
+ bool check_scan_plan_when_empty_result = true,
+ bool apply_predicate_to_scan = true) const {
// scan
ScanContextBuilder scan_context_builder(table_path);
- scan_context_builder.SetPredicate(predicate);
+ scan_context_builder.SetPredicate(apply_predicate_to_scan ? predicate
: nullptr);
if (!row_ranges.empty()) {
auto global_index_result =
BitmapGlobalIndexResult::FromRanges(row_ranges);
scan_context_builder.SetGlobalIndexResult(global_index_result);
@@ -1880,7 +1881,7 @@ TEST_P(DataEvolutionTableTest, TestScanAndReadWithIndex) {
expected_array));
}
{
- // first 4 records read with data evolution, ignore index
+ // The old file's f2 index does not contain 102, but the newer file
owns f2.
auto predicate = PredicateBuilder::Equal(/*field_index=*/2,
/*field_name=*/"f2",
FieldType::INT, Literal(102));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
@@ -1892,51 +1893,45 @@ TEST_P(DataEvolutionTableTest,
TestScanAndReadWithIndex) {
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path,
arrow::schema(arrow_data_type->fields())->field_names(),
- expected_array, predicate));
+ expected_array, predicate,
+ /*row_ranges=*/{},
+ /*check_scan_plan_when_empty_result=*/true,
+ /*apply_predicate_to_scan=*/false));
}
{
- // f2 has bitmap index, but data evolution scan and read ignore index
+ // The bitmap proves that neither row range group contains f2 = 103.
auto predicate = PredicateBuilder::Equal(/*field_index=*/2,
/*field_name=*/"f2",
FieldType::INT, Literal(103));
- auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
- arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
- ["Lily", 2, 102, 2.1],
- ["Alice", 4, 104, 3.1],
- ["Bob", 6, 106, 4.1],
- ["David", 8, 108, 5.1]
- ])")
- .ValueOrDie());
ASSERT_OK(ScanAndRead(table_path,
arrow::schema(arrow_data_type->fields())->field_names(),
- expected_array, predicate));
+ /*expected_array=*/nullptr, predicate,
+ /*row_ranges=*/{},
+ /*check_scan_plan_when_empty_result=*/false,
+ /*apply_predicate_to_scan=*/false));
}
{
- // f2 has bitmap index, data evolution scan will ignore index => not
empty plan
- // data evolution split read will also ignore index => not empty read
batch
+ // Scan planning keeps the split, but reader-side indexes skip both
row range groups.
auto predicate = PredicateBuilder::Equal(/*field_index=*/2,
/*field_name=*/"f2",
FieldType::INT, Literal(203));
- auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
- arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
- [null, null, 202, 6.1],
- [null, null, 204, 7.1]
- ])")
- .ValueOrDie());
ASSERT_OK(ScanAndRead(table_path,
arrow::schema(arrow_data_type->fields())->field_names(),
- expected_array, predicate,
+ /*expected_array=*/nullptr, predicate,
/*row_ranges=*/{},
- /*check_scan_plan_when_empty_result=*/true));
+ /*check_scan_plan_when_empty_result=*/false,
+ /*apply_predicate_to_scan=*/false));
}
{
- // f2 has bitmap index, data evolution split read will ignore index
+ // A single-file group applies the exact bitmap row selection.
auto predicate = PredicateBuilder::Equal(/*field_index=*/2,
/*field_name=*/"f2",
FieldType::INT, Literal(202));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
- [null, null, 202, 6.1],
- [null, null, 204, 7.1]
+ [null, null, 202, 6.1]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path,
arrow::schema(arrow_data_type->fields())->field_names(),
- expected_array, predicate));
+ expected_array, predicate,
+ /*row_ranges=*/{},
+ /*check_scan_plan_when_empty_result=*/true,
+ /*apply_predicate_to_scan=*/false));
}
{
auto predicate =
@@ -1953,7 +1948,7 @@ TEST_P(DataEvolutionTableTest, TestScanAndReadWithIndex) {
{
// test row id with predicate
std::vector<Range> row_ranges = {Range(0l, 2l)};
- // row id = {0, 1, 2}, while data evolution split read will ignore
index
+ // A merged group keeps all selected row ids to preserve column
alignment.
auto predicate = PredicateBuilder::Equal(/*field_index=*/2,
/*field_name=*/"f2",
FieldType::INT, Literal(106));
CheckScanResult(table_path, /*predicate=*/predicate,
/*row_ranges=*/row_ranges,
@@ -1967,26 +1962,179 @@ TEST_P(DataEvolutionTableTest,
TestScanAndReadWithIndex) {
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path,
arrow::schema(arrow_data_type->fields())->field_names(),
expected_array, predicate,
- /*row_ranges=*/row_ranges));
+ /*row_ranges=*/row_ranges,
+ /*check_scan_plan_when_empty_result=*/true,
+ /*apply_predicate_to_scan=*/false));
}
{
// test row id with predicate
std::vector<Range> row_ranges = {Range(4l, 5l)};
- // row id = {4, 5}, data evolution split read will ignore bitmap index
+ // The single-file bitmap selection is intersected with the row-id
selection.
auto predicate = PredicateBuilder::Equal(/*field_index=*/2,
/*field_name=*/"f2",
FieldType::INT, Literal(204));
CheckScanResult(table_path, /*predicate=*/predicate,
/*row_ranges=*/row_ranges,
/*expected_first_row_ids=*/{4},
/*expected_row_counts=*/{2});
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow_data_type, R"([
- [null, null, 202, 6.1],
[null, null, 204, 7.1]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path,
arrow::schema(arrow_data_type->fields())->field_names(),
expected_array, predicate,
- /*row_ranges=*/row_ranges));
+ /*row_ranges=*/row_ranges,
+ /*check_scan_plan_when_empty_result=*/true,
+ /*apply_predicate_to_scan=*/false));
+ }
+}
+
+TEST_P(DataEvolutionTableTest, TestDataEvolutionPredicatePushDownBoundaries) {
+ auto file_format = FileFormat();
+ if (file_format == "avro") {
+ return;
+ }
+ std::string table_path = paimon::test::GetDataDir() + file_format +
+
"/data_evolution_with_index.db/data_evolution_with_index";
+
+ {
+ // A file without f0 must not interpret the predicate as f0 = null.
+ auto predicate = PredicateBuilder::Equal(
+ /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING,
+ Literal(FieldType::STRING, "Lily", 4));
+ auto read_type =
+ arrow::struct_({arrow::field("f0", arrow::utf8()),
arrow::field("f2", arrow::int32())});
+ auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(read_type, R"([
+ ["Lily", 102],
+ ["Alice", 104],
+ ["Bob", 106],
+ ["David", 108],
+ [null, 202],
+ [null, 204]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"f0", "f2"}, expected_array,
predicate,
+ /*row_ranges=*/{},
+ /*check_scan_plan_when_empty_result=*/true,
+ /*apply_predicate_to_scan=*/false));
}
+ {
+ // System fields are completed after reading and cannot be pushed into
data files.
+ auto predicate = PredicateBuilder::Equal(/*field_index=*/1,
/*field_name=*/"_ROW_ID",
+ FieldType::BIGINT,
Literal(99l));
+ auto read_type =
+ arrow::struct_({arrow::field("f2", arrow::int32()),
SpecialFields::RowId().field_});
+ auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(read_type, R"([
+ [102, 0],
+ [104, 1],
+ [106, 2],
+ [108, 3],
+ [202, 4],
+ [204, 5]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"f2", "_ROW_ID"}, expected_array,
predicate,
+ /*row_ranges=*/{},
+ /*check_scan_plan_when_empty_result=*/true,
+ /*apply_predicate_to_scan=*/false));
+ }
+ {
+ // Dropping a system-field conjunct must not drop a pushable data
conjunct.
+ auto data_predicate = PredicateBuilder::Equal(
+ /*field_index=*/0, /*field_name=*/"f2", FieldType::INT,
Literal(103));
+ auto system_predicate = PredicateBuilder::Equal(
+ /*field_index=*/1, /*field_name=*/"_ROW_ID", FieldType::BIGINT,
Literal(0l));
+ ASSERT_OK_AND_ASSIGN(auto predicate,
+ PredicateBuilder::And({data_predicate,
system_predicate}));
+ ASSERT_OK(ScanAndRead(table_path, {"f2", "_ROW_ID"},
/*expected_array=*/nullptr, predicate,
+ /*row_ranges=*/{},
+ /*check_scan_plan_when_empty_result=*/false,
+ /*apply_predicate_to_scan=*/false));
+ }
+ {
+ // Bitmap positions compose with row ranges without changing the
physical row id.
+ auto predicate = PredicateBuilder::Equal(/*field_index=*/0,
/*field_name=*/"f2",
+ FieldType::INT, Literal(204));
+ auto read_type =
+ arrow::struct_({arrow::field("f2", arrow::int32()),
SpecialFields::RowId().field_});
+ auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(read_type, R"([
+ [204, 5]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"f2", "_ROW_ID"}, expected_array,
predicate,
+ /*row_ranges=*/{Range(4l, 5l)},
+ /*check_scan_plan_when_empty_result=*/true,
+ /*apply_predicate_to_scan=*/false));
+ }
+ {
+ // The bitmap selects row id 5 while the global-index selection keeps
only row id 4.
+ auto predicate = PredicateBuilder::Equal(/*field_index=*/0,
/*field_name=*/"f2",
+ FieldType::INT, Literal(204));
+ ASSERT_OK(ScanAndRead(table_path, {"f2", "_ROW_ID"},
/*expected_array=*/nullptr, predicate,
+ /*row_ranges=*/{Range(4l, 4l)},
+ /*check_scan_plan_when_empty_result=*/false,
+ /*apply_predicate_to_scan=*/false));
+ }
+ {
+ // The predicate keeps row ids {4, 5}; the global-index selection
keeps {0, 1, 2, 3, 4}.
+ auto equal_202 = PredicateBuilder::Equal(/*field_index=*/0,
/*field_name=*/"f2",
+ FieldType::INT, Literal(202));
+ auto equal_204 = PredicateBuilder::Equal(/*field_index=*/0,
/*field_name=*/"f2",
+ FieldType::INT, Literal(204));
+ ASSERT_OK_AND_ASSIGN(auto predicate, PredicateBuilder::Or({equal_202,
equal_204}));
+ auto read_type =
+ arrow::struct_({arrow::field("f2", arrow::int32()),
SpecialFields::RowId().field_});
+ auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(read_type, R"([
+ [202, 4]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"f2", "_ROW_ID"}, expected_array,
predicate,
+ /*row_ranges=*/{Range(0l, 4l)},
+ /*check_scan_plan_when_empty_result=*/true,
+ /*apply_predicate_to_scan=*/false));
+ }
+}
+
+TEST_P(DataEvolutionTableTest, TestFormatPredicatePushDownWithoutFileIndex) {
+ if (FileFormat() == "avro") {
+ return;
+ }
+
+ CreateDataEvolutionTable(
+ /*deletion_vectors_enabled=*/false,
{{Options::FILE_INDEX_READ_ENABLED, "false"},
+ {Options::WRITE_BATCH_SIZE, "1"},
+ {"parquet.page.size", "1"},
+ {"parquet.enable-dictionary",
"false"},
+
{"parquet.write.enable-page-index", "true"},
+
{"parquet.read.enable-page-index-filter", "true"},
+ {"orc.stripe.size", "1"},
+ {"orc.row.index.stride", "1"}});
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+ auto input = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [1, "a", "x"],
+ [2, "b", "y"],
+ [3, "c", "z"],
+ [4, "d", "w"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(auto commit_messages, WriteArray(table_path, {"f0",
"f1", "f2"}, input));
+ ASSERT_OK(Commit(table_path, commit_messages));
+
+ auto predicate =
+ PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0",
FieldType::INT, Literal(3));
+ auto expected = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [3, "c", "z"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2"}, expected, predicate,
+ /*row_ranges=*/{},
+ /*check_scan_plan_when_empty_result=*/true,
+ /*apply_predicate_to_scan=*/true));
}
TEST_P(DataEvolutionTableTest, TestPredicate) {
