This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 21146692 perf(predicate): avoid materializing all columns for field
bounds checks (#318)
21146692 is described below
commit 21146692aa6ca228f8a29e26f6a71edf9c1bede1
Author: gripleaf <[email protected]>
AuthorDate: Thu Sep 10 20:55:31 2026 +0800
perf(predicate): avoid materializing all columns for field bounds checks
(#318)
---
src/paimon/common/predicate/leaf_predicate_impl.h | 4 +-
src/paimon/common/predicate/predicate_test.cpp | 20 ++++
test/inte/write_and_read_inte_test.cpp | 108 ++++++++++++++++++++++
3 files changed, 130 insertions(+), 2 deletions(-)
diff --git a/src/paimon/common/predicate/leaf_predicate_impl.h
b/src/paimon/common/predicate/leaf_predicate_impl.h
index 0d298409..a301e19c 100644
--- a/src/paimon/common/predicate/leaf_predicate_impl.h
+++ b/src/paimon/common/predicate/leaf_predicate_impl.h
@@ -44,10 +44,10 @@ class LeafPredicateImpl : public LeafPredicate, public
PredicateFilter {
Result<std::vector<char>> Test(const arrow::Array& array,
arrow::MemoryPool* pool) const override {
const auto& struct_array = checked_cast<const
arrow::StructArray&>(array);
- if (field_index_ >=
static_cast<int32_t>(struct_array.fields().size())) {
+ if (field_index_ >= struct_array.num_fields()) {
return Status::Invalid(
fmt::format("field index {} exceed field count {} in struct
array", field_index_,
- struct_array.fields().size()));
+ struct_array.num_fields()));
}
const auto& field_array = struct_array.field(field_index_);
return leaf_function_.Test(*field_array, literals_, pool);
diff --git a/src/paimon/common/predicate/predicate_test.cpp
b/src/paimon/common/predicate/predicate_test.cpp
index 332fa276..97fef15d 100644
--- a/src/paimon/common/predicate/predicate_test.cpp
+++ b/src/paimon/common/predicate/predicate_test.cpp
@@ -172,6 +172,26 @@ class PredicateTest : public ::testing::Test {
}
};
+TEST_F(PredicateTest, TestLeafFieldBoundsOnWideSlicedBatch) {
+ auto values =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::int64(),
"[9,1,null,2,9]").ValueOrDie();
+ arrow::ArrayVector columns(64, values);
+ std::vector<std::string> names;
+ for (int32_t i = 0; i < 64; ++i) {
+ names.push_back(fmt::format("field_{}", i));
+ }
+ auto batch = arrow::StructArray::Make(columns, names).ValueOrDie();
+ auto sliced = batch->Slice(1, 3);
+ auto pool = arrow::default_memory_pool();
+ auto predicate = std::dynamic_pointer_cast<PredicateFilter>(
+ PredicateBuilder::Equal(63, "field_63", FieldType::BIGINT,
Literal(int64_t{2})));
+ ASSERT_OK_AND_ASSIGN(auto full, predicate->Test(*sliced, pool));
+ ASSERT_EQ(full, std::vector<char>({0, 0, 1}));
+ auto invalid = std::dynamic_pointer_cast<PredicateFilter>(
+ PredicateBuilder::Equal(64, "missing", FieldType::BIGINT,
Literal(int64_t{2})));
+ ASSERT_NOK_WITH_MSG(invalid->Test(*sliced, pool), "field index 64 exceed
field count 64");
+}
+
TEST_F(PredicateTest, TestInvalidFieldIndex) {
auto bigint_type = arrow::int64();
auto predicate_base = PredicateBuilder::Equal(/*field_index=*/2,
/*field_name=*/"f0",
diff --git a/test/inte/write_and_read_inte_test.cpp
b/test/inte/write_and_read_inte_test.cpp
index cca4f786..7d184bb8 100644
--- a/test/inte/write_and_read_inte_test.cpp
+++ b/test/inte/write_and_read_inte_test.cpp
@@ -357,6 +357,114 @@ TEST_P(WriteAndReadInteTest, TestAppendSimple) {
ASSERT_TRUE(success);
}
+TEST_P(WriteAndReadInteTest, TestAppendReadWithNestedPredicateAcrossBatches) {
+ auto [file_format, file_system] = GetParam();
+ arrow::FieldVector fields = {
+ arrow::field("id", arrow::int32()), arrow::field("key",
arrow::int64()),
+ arrow::field("value", arrow::utf8()), arrow::field("payload",
arrow::binary())};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, file_format},
+ {Options::FILE_SYSTEM, file_system},
+ {Options::BUCKET, "-1"},
+ {Options::TARGET_FILE_SIZE, "1048576"},
+ };
+ if (file_system == "jindo") {
+ options = AddOptionsForJindo(options);
+ }
+ ASSERT_OK_AND_ASSIGN(auto helper,
+ TestHelper::Create(test_dir_, arrow::schema(fields),
/*partition_keys=*/{},
+ /*primary_keys=*/{}, options,
+ /*is_streaming_mode=*/false));
+ // For the read batch size of 4 configured below, consecutive groups of
four
+ // input rows have 0, 2, 4, 0, 2, and 4 matches for the AND predicate.
+ // Input rows at positions 4 and 6 are identical and must both survive.
+ const std::string data_json = R"([
+ [0, 0, "a", "zero"], [1, null, "b", null],
+ [2, 0, null, "two"], [3, 0, "c", "three"],
+ [4, 1, "b", "keep"], [5, 0, "a", "drop"],
+ [4, 1, "b", "keep"], [7, 1, null, "null"],
+ [8, 1, "a", "eight"], [9, 1, "b", null],
+ [10, 1, "a", "ten"], [11, 1, "b", "eleven"],
+ [12, 0, "a", "twelve"], [13, 0, "b", "thirteen"],
+ [14, null, "a", "fourteen"], [15, 1, "c", "fifteen"],
+ [16, 1, "b", "sixteen"], [17, 0, "b", "seventeen"],
+ [18, 1, "a", "eighteen"], [19, 1, null, "nineteen"],
+ [20, 1, "a", "twenty"], [21, 1, "a", "twenty-one"],
+ [22, 1, "b", "twenty-two"], [23, 1, "b", "twenty-three"]
+ ])";
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields),
data_json,
+ /*partition_map=*/{},
/*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0,
+
/*expected_commit_messages=*/std::nullopt));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> splits,
+ helper->NewScan(StartupMode::LatestFull(),
/*snapshot_id=*/std::nullopt,
+ /*is_streaming=*/false));
+ ASSERT_FALSE(splits.empty());
+
+ // Predicate indices refer to the projected read schema: payload, value,
id, key.
+ auto key = PredicateBuilder::Equal(3, "key", FieldType::BIGINT,
Literal(int64_t{1}));
+ auto a =
+ PredicateBuilder::Equal(1, "value", FieldType::STRING,
Literal(FieldType::STRING, "a", 1));
+ auto b =
+ PredicateBuilder::Equal(1, "value", FieldType::STRING,
Literal(FieldType::STRING, "b", 1));
+ ASSERT_OK_AND_ASSIGN(auto alternatives, PredicateBuilder::Or({a, b}));
+ ASSERT_OK_AND_ASSIGN(auto conjunction, PredicateBuilder::And({key,
alternatives}));
+ ASSERT_OK_AND_ASSIGN(auto both, PredicateBuilder::And({key, a}));
+ ASSERT_OK_AND_ASSIGN(auto disjunction, PredicateBuilder::Or({both, b}));
+ const std::vector<std::pair<std::shared_ptr<Predicate>,
std::vector<int64_t>>> cases = {
+ {conjunction, {4, 6, 8, 9, 10, 11, 16, 18, 20, 21, 22, 23}},
+ {disjunction, {1, 4, 6, 8, 9, 10, 11, 13, 16, 17, 18, 20, 21, 22, 23}},
+ };
+
+ // Build the oracle from written rows and explicit expected positions,
independently
+ // of predicate evaluation. Reordering both predicate columns also tests
name binding.
+ auto input = checked_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
data_json).ValueOrDie());
+ auto row_kinds = arrow::MakeArrayFromScalar(arrow::Int8Scalar(0),
input->length()).ValueOrDie();
+ auto projected =
+ arrow::StructArray::Make(
+ {row_kinds, input->field(3), input->field(2), input->field(0),
input->field(1)},
+ std::vector<std::string>{"_VALUE_KIND", "payload", "value", "id",
"key"})
+ .ValueOrDie();
+ const auto unfiltered = std::make_shared<arrow::ChunkedArray>(projected);
+ const std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
+ for (const auto& [predicate, expected_positions] : cases) {
+ SCOPED_TRACE(predicate->ToString());
+ arrow::ArrayVector expected_rows;
+ for (int64_t position : expected_positions) {
+ expected_rows.push_back(projected->Slice(position, 1));
+ }
+ const auto filtered =
std::make_shared<arrow::ChunkedArray>(expected_rows);
+ for (int32_t batch_size : {1, 4, 7}) {
+ SCOPED_TRACE(batch_size);
+ for (bool enable_filter : {false, true}) {
+ SCOPED_TRACE(enable_filter);
+ ReadContextBuilder builder(table_path);
+ builder.SetOptions(options)
+ .SetReadFieldNames({"payload", "value", "id", "key"})
+ .SetPredicate(predicate)
+ .EnablePredicateFilter(enable_filter)
+ .EnablePrefetch(false)
+ .EnableLateMaterializing(false)
+ .AddOption(Options::READ_BATCH_SIZE,
std::to_string(batch_size));
+ ASSERT_OK_AND_ASSIGN(auto context, builder.Finish());
+ ASSERT_OK_AND_ASSIGN(auto table_read,
TableRead::Create(std::move(context)));
+ ASSERT_OK_AND_ASSIGN(auto reader,
table_read->CreateReader(splits));
+ ASSERT_OK_AND_ASSIGN(auto actual,
ReadResultCollector::CollectResult(reader.get()));
+ ASSERT_TRUE(actual);
+ // The same predicate without precise filtering must retain
all input rows,
+ // proving that format pushdown did not remove the empty or
partial batches.
+ const auto& expected = enable_filter ? filtered : unfiltered;
+ ASSERT_TRUE(expected->Equals(actual)) << actual->ToString();
+ ASSERT_OK_AND_ASSIGN(auto eof, reader->NextBatch());
+ ASSERT_TRUE(BatchReader::IsEofBatch(eof));
+ reader->Close();
+ }
+ }
+ }
+}
+
TEST_P(WriteAndReadInteTest, TestAppendVector) {
auto [file_format, file_system] = GetParam();
if (file_format != "parquet") {
