From 2a33fdd37b872c0fd7ae99f5d058432088ecc46f Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Tue, 8 Sep 2026 20:03:08 +0800 Subject: [PATCH 1/6] perf(predicate): evaluate compound filters on candidate rows --- src/paimon/common/predicate/and.h | 21 +- .../common/predicate/compound_function.h | 55 ++++ .../predicate/compound_predicate_impl.h | 6 + .../common/predicate/leaf_predicate_impl.h | 35 +++ src/paimon/common/predicate/or.h | 21 +- .../common/predicate/predicate_filter.h | 41 +++ .../common/predicate/predicate_test.cpp | 238 ++++++++++++++++++ 7 files changed, 391 insertions(+), 26 deletions(-) diff --git a/src/paimon/common/predicate/and.h b/src/paimon/common/predicate/and.h index f0da67d74..a73a95d29 100644 --- a/src/paimon/common/predicate/and.h +++ b/src/paimon/common/predicate/and.h @@ -52,19 +52,14 @@ class And : public CompoundFunction { Result> Test(const arrow::Array& array, const std::vector>& children, arrow::MemoryPool* pool) const override { - std::vector is_valid(array.length(), true); - for (const auto& child : children) { - auto child_filter = std::dynamic_pointer_cast(child); - if (!child_filter) { - return Status::Invalid( - fmt::format("child filter {} does not support Test", child->ToString())); - } - PAIMON_ASSIGN_OR_RAISE(std::vector child_valid, child_filter->Test(array, pool)); - for (size_t i = 0; i < is_valid.size(); i++) { - is_valid[i] = (is_valid[i] & child_valid[i]); - } - } - return is_valid; + return TestWithSelection(array, children, pool, /*is_and=*/true); + } + + Result> TestSelected(const arrow::Array& array, + const std::vector>& children, + const std::vector& selection, + arrow::MemoryPool* pool) const override { + return TestWithSelection(array, children, pool, /*is_and=*/true, &selection); } Result Test(const std::shared_ptr& schema, const InternalRow& row, diff --git a/src/paimon/common/predicate/compound_function.h b/src/paimon/common/predicate/compound_function.h index e32a3188b..9a0f16371 100644 --- a/src/paimon/common/predicate/compound_function.h +++ b/src/paimon/common/predicate/compound_function.h @@ -18,13 +18,17 @@ #pragma once +#include #include +#include #include #include "arrow/array/array_base.h" #include "arrow/type_fwd.h" +#include "fmt/format.h" #include "paimon/common/data/internal_array.h" #include "paimon/common/data/internal_row.h" +#include "paimon/common/predicate/predicate_filter.h" #include "paimon/predicate/function.h" #include "paimon/predicate/predicate.h" #include "paimon/result.h" @@ -46,6 +50,57 @@ class CompoundFunction : public Function { const InternalArray& null_counts, const std::vector>& children) const = 0; + virtual Result> TestSelected( + const arrow::Array& array, const std::vector>& children, + const std::vector& selection, arrow::MemoryPool* pool) const = 0; + virtual const CompoundFunction& Negate() const = 0; + + protected: + static Result> TestWithSelection( + const arrow::Array& array, const std::vector>& children, + arrow::MemoryPool* pool, bool is_and, + const std::vector* input_selection = nullptr) { + std::vector selection; + if (input_selection) { + selection = *input_selection; + } else { + selection.resize(array.length()); + std::iota(selection.begin(), selection.end(), int64_t{0}); + } + std::vector results(selection.size(), is_and); + std::vector positions(selection.size()); + std::iota(positions.begin(), positions.end(), size_t{0}); + for (const auto& child : children) { + auto filter = std::dynamic_pointer_cast(child); + if (!filter) { + return Status::Invalid( + fmt::format("child filter {} does not support Test", child->ToString())); + } + PAIMON_ASSIGN_OR_RAISE(std::vector child_results, + filter->TestSelected(array, selection, pool)); + if (child_results.size() != selection.size()) { + return Status::Invalid("predicate result size does not match selection size"); + } + // Avoid rewriting indices and results when every candidate needs the next child. + if (std::all_of(child_results.begin(), child_results.end(), + [is_and](char matched) { return (matched != 0) == is_and; })) { + continue; + } + size_t remaining = 0; + for (size_t i = 0; i < selection.size(); ++i) { + const int64_t row = selection[i]; + const bool matched = child_results[i] != 0; + results[positions[i]] = matched; + // AND continues only on matches; OR continues only on non-matches. + if (matched == is_and) { + positions[remaining] = positions[i]; + selection[remaining++] = row; + } + } + selection.resize(remaining); + } + return results; + } }; } // namespace paimon diff --git a/src/paimon/common/predicate/compound_predicate_impl.h b/src/paimon/common/predicate/compound_predicate_impl.h index 7ab5b613b..5da7766ef 100644 --- a/src/paimon/common/predicate/compound_predicate_impl.h +++ b/src/paimon/common/predicate/compound_predicate_impl.h @@ -37,6 +37,12 @@ class CompoundPredicateImpl : public CompoundPredicate, public PredicateFilter { return compound_function_.Test(array, children_, pool); } + Result> TestSelected(const arrow::Array& array, + const std::vector& selection, + arrow::MemoryPool* pool) const override { + return compound_function_.TestSelected(array, children_, selection, pool); + } + Result Test(const std::shared_ptr& schema, const InternalRow& row) const override { return compound_function_.Test(schema, row, children_); diff --git a/src/paimon/common/predicate/leaf_predicate_impl.h b/src/paimon/common/predicate/leaf_predicate_impl.h index 0d2984095..8707645d0 100644 --- a/src/paimon/common/predicate/leaf_predicate_impl.h +++ b/src/paimon/common/predicate/leaf_predicate_impl.h @@ -28,6 +28,7 @@ #include "paimon/common/predicate/literal_converter.h" #include "paimon/common/predicate/predicate_filter.h" #include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/field_type_utils.h" #include "paimon/predicate/leaf_predicate.h" namespace paimon { class LeafPredicateImpl : public LeafPredicate, public PredicateFilter { @@ -53,6 +54,40 @@ class LeafPredicateImpl : public LeafPredicate, public PredicateFilter { return leaf_function_.Test(*field_array, literals_, pool); } + Result> TestSelected(const arrow::Array& array, + const std::vector& selection, + arrow::MemoryPool* pool) const override { + const auto& struct_array = checked_cast(array); + if (field_index_ >= static_cast(struct_array.fields().size())) { + return Status::Invalid( + fmt::format("field index {} exceed field count {} in struct array", field_index_, + struct_array.fields().size())); + } + const auto& field_array = struct_array.field(field_index_); + // Preserve eager conversion/comparison errors, even when the offending row is not + // selected. Dictionary and unsupported physical types also retain the original path. + // LIKE validates escapes while examining values; temporal/decimal conversions may fail + // on rows outside the candidate set as well. + if (leaf_function_.GetType() == Function::Type::LIKE || + field_array->type_id() == arrow::Type::TIMESTAMP || + field_array->type_id() == arrow::Type::DECIMAL128) { + return PredicateFilter::TestSelected(array, selection, pool); + } + auto field_type = FieldTypeUtils::ConvertToFieldType(field_array->type_id()); + if (!field_type.ok()) { + return PredicateFilter::TestSelected(array, selection, pool); + } + for (const auto& literal : literals_) { + if (!literal.IsNull() && literal.GetType() != field_type.value()) { + return PredicateFilter::TestSelected(array, selection, pool); + } + } + // Gather only the predicate column, not unrelated payload columns. + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr selected, + SelectRows(*field_array, selection, pool)); + return leaf_function_.Test(*selected, literals_, pool); + } + Result Test(const std::shared_ptr& schema, const InternalRow& row) const override { if (field_index_ >= row.GetFieldCount()) { diff --git a/src/paimon/common/predicate/or.h b/src/paimon/common/predicate/or.h index f74046407..88825ecc4 100644 --- a/src/paimon/common/predicate/or.h +++ b/src/paimon/common/predicate/or.h @@ -48,19 +48,14 @@ class Or : public CompoundFunction { Result> Test(const arrow::Array& array, const std::vector>& children, arrow::MemoryPool* pool) const override { - std::vector is_valid(array.length(), false); - for (const auto& child : children) { - auto child_filter = std::dynamic_pointer_cast(child); - if (!child_filter) { - return Status::Invalid( - fmt::format("child filter {} does not support Test", child->ToString())); - } - PAIMON_ASSIGN_OR_RAISE(std::vector child_valid, child_filter->Test(array, pool)); - for (size_t i = 0; i < is_valid.size(); i++) { - is_valid[i] = (is_valid[i] | child_valid[i]); - } - } - return is_valid; + return TestWithSelection(array, children, pool, /*is_and=*/false); + } + + Result> TestSelected(const arrow::Array& array, + const std::vector>& children, + const std::vector& selection, + arrow::MemoryPool* pool) const override { + return TestWithSelection(array, children, pool, /*is_and=*/false, &selection); } Result Test(const std::shared_ptr& schema, const InternalRow& row, diff --git a/src/paimon/common/predicate/predicate_filter.h b/src/paimon/common/predicate/predicate_filter.h index d14b47476..348a425bd 100644 --- a/src/paimon/common/predicate/predicate_filter.h +++ b/src/paimon/common/predicate/predicate_filter.h @@ -22,8 +22,10 @@ #include #include "arrow/api.h" +#include "arrow/compute/api.h" #include "paimon/common/data/internal_array.h" #include "paimon/common/data/internal_row.h" +#include "paimon/common/utils/arrow/status_utils.h" #include "paimon/predicate/predicate.h" namespace paimon { @@ -34,10 +36,49 @@ class PredicateFilter : virtual public Predicate { /// null virtual Result> Test(const arrow::Array& array, arrow::MemoryPool* pool) const = 0; + /// Evaluate only the specified batch-local rows, returning results in selection order. + /// The selection must contain valid, strictly increasing indices. The default retains eager + /// evaluation, including errors on unselected rows. Implementations may specialize this only + /// when skipping rows does not bypass validation. + virtual Result> TestSelected(const arrow::Array& array, + const std::vector& selection, + arrow::MemoryPool* pool) const { + PAIMON_ASSIGN_OR_RAISE(std::vector all_results, Test(array, pool)); + if (all_results.size() != static_cast(array.length())) { + return Status::Invalid("predicate result size does not match array length"); + } + std::vector results; + results.reserve(selection.size()); + for (int64_t row : selection) { + results.push_back(all_results[row]); + } + return results; + } virtual Result Test(const std::shared_ptr& schema, const InternalRow& row) const = 0; virtual Result Test(const std::shared_ptr& schema, int64_t row_count, const InternalRow& min_values, const InternalRow& max_values, const InternalArray& null_counts) const = 0; + + protected: + static Result> SelectRows(const arrow::Array& array, + const std::vector& selection, + arrow::MemoryPool* pool) { + if (selection.empty()) { + return array.Slice(0, 0); + } + if (selection.size() == static_cast(array.length())) { + return arrow::MakeArray(array.data()); + } + arrow::Int64Builder builder(pool); + PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.AppendValues(selection)); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr indices, builder.Finish()); + arrow::compute::ExecContext context(pool); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + arrow::Datum selected, + arrow::compute::Take(arrow::Datum(array), arrow::Datum(indices), + arrow::compute::TakeOptions::Defaults(), &context)); + return selected.make_array(); + } }; } // namespace paimon diff --git a/src/paimon/common/predicate/predicate_test.cpp b/src/paimon/common/predicate/predicate_test.cpp index 332fa276d..26b9684ac 100644 --- a/src/paimon/common/predicate/predicate_test.cpp +++ b/src/paimon/common/predicate/predicate_test.cpp @@ -18,8 +18,10 @@ #include "paimon/predicate/predicate.h" +#include #include #include +#include #include #include #include @@ -35,10 +37,12 @@ #include "paimon/common/data/binary_array.h" #include "paimon/common/data/binary_row.h" #include "paimon/common/data/binary_row_writer.h" +#include "paimon/common/predicate/equal.h" #include "paimon/common/predicate/leaf_predicate_impl.h" #include "paimon/common/predicate/predicate_filter.h" #include "paimon/defs.h" #include "paimon/memory/memory_pool.h" +#include "paimon/predicate/compound_predicate.h" #include "paimon/predicate/function.h" #include "paimon/predicate/literal.h" #include "paimon/predicate/predicate_builder.h" @@ -51,6 +55,42 @@ class Array; } // namespace arrow namespace paimon::test { +namespace { +// Reference implementation of the eager batch evaluation used before candidate selection. +Result> TestEager(const std::shared_ptr& predicate, + const arrow::Array& array, arrow::MemoryPool* pool) { + auto compound = std::dynamic_pointer_cast(predicate); + if (!compound) { + return std::dynamic_pointer_cast(predicate)->Test(array, pool); + } + const bool is_and = predicate->GetFunction().GetType() == Function::Type::AND; + std::vector result(array.length(), is_and); + for (const auto& child : compound->Children()) { + PAIMON_ASSIGN_OR_RAISE(std::vector matches, TestEager(child, array, pool)); + for (size_t i = 0; i < result.size(); ++i) { + result[i] = is_and ? (result[i] & matches[i]) : (result[i] | matches[i]); + } + } + return result; +} + +class RecordingEqualPredicate : public LeafPredicateImpl { + public: + RecordingEqualPredicate(int32_t index, const std::string& name, FieldType type, + const Literal& literal) + : LeafPredicateImpl(Equal::Instance(), index, name, type, {literal}) {} + + Result> TestSelected(const arrow::Array& array, + const std::vector& selection, + arrow::MemoryPool* pool) const override { + selections.push_back(selection); + return LeafPredicateImpl::TestSelected(array, selection, pool); + } + + mutable std::vector> selections; +}; +} // namespace + class PredicateTest : public ::testing::Test { public: void SetUp() override {} @@ -1149,6 +1189,204 @@ TEST_F(PredicateTest, TestOr) { StatsCheck(*predicate, 3ll, {FieldStats(6ll, 7ll, 0ll), FieldStats(8ll, 10ll, 0ll)})); } +TEST_F(PredicateTest, TestNestedCandidateSelectionWithSliceAndNulls) { + for (const auto& type : {arrow::utf8(), arrow::binary()}) { + const FieldType field_type = + type->id() == arrow::Type::STRING ? FieldType::STRING : FieldType::BINARY; + auto keys = arrow::ipc::internal::json::ArrayFromJSON(arrow::int64(), "[9,1,0,1,1,null,9]") + .ValueOrDie(); + auto values = + arrow::ipc::internal::json::ArrayFromJSON(type, R"(["pad","a","b","b",null,"a","pad"])") + .ValueOrDie(); + auto batch = + arrow::StructArray::Make({keys, values}, std::vector{"key", "value"}) + .ValueOrDie(); + auto sliced = batch->Slice(1, 5); + auto key = std::make_shared(0, "key", FieldType::BIGINT, + Literal(int64_t{1})); + auto a = std::make_shared(1, "value", field_type, + Literal(field_type, "a", 1)); + auto b = std::make_shared(1, "value", field_type, + Literal(field_type, "b", 1)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr alternatives, PredicateBuilder::Or({a, b})); + ASSERT_OK_AND_ASSIGN(std::shared_ptr predicate, + PredicateBuilder::And({key, alternatives})); + auto filter = std::dynamic_pointer_cast(predicate); + arrow::ProxyMemoryPool pool(arrow::default_memory_pool()); + ASSERT_OK_AND_ASSIGN(std::vector actual, filter->Test(*sliced, &pool)); + ASSERT_EQ(actual, std::vector({1, 0, 1, 0, 0})); + ASSERT_EQ(key->selections, std::vector>({{0, 1, 2, 3, 4}})); + ASSERT_EQ(a->selections, std::vector>({{0, 2, 3}})); + ASSERT_EQ(b->selections, std::vector>({{2, 3}})); + ASSERT_GT(pool.max_memory(), 0); + ASSERT_EQ(pool.bytes_allocated(), 0); + + ASSERT_OK_AND_ASSIGN(std::vector selected, + filter->TestSelected(*sliced, {2, 4}, &pool)); + ASSERT_EQ(selected, std::vector({1, 0})); + ASSERT_OK_AND_ASSIGN(std::vector empty, filter->Test(*sliced->Slice(0, 0), &pool)); + ASSERT_TRUE(empty.empty()); + } +} + +TEST_F(PredicateTest, TestEmptyCandidatesStillValidateFields) { + auto keys = arrow::ipc::internal::json::ArrayFromJSON(arrow::int64(), "[1,2]").ValueOrDie(); + auto batch = arrow::StructArray::Make({keys}, std::vector{"key"}).ValueOrDie(); + auto invalid = PredicateBuilder::Equal(1, "missing", FieldType::BIGINT, Literal(int64_t{1})); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr conjunction, + PredicateBuilder::And( + {PredicateBuilder::Equal(0, "key", FieldType::BIGINT, Literal(int64_t{9})), invalid})); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr disjunction, + PredicateBuilder::Or({PredicateBuilder::IsNotNull(0, "key", FieldType::BIGINT), invalid})); + for (const auto& predicate : {conjunction, disjunction}) { + auto filter = std::dynamic_pointer_cast(predicate); + ASSERT_NOK_WITH_MSG(filter->Test(*batch, arrow::default_memory_pool()), "field index"); + } +} + +TEST_F(PredicateTest, TestEmptyCandidatesStillValidateLiteralsAndTypes) { + auto keys = arrow::ipc::internal::json::ArrayFromJSON(arrow::int64(), "[1,2]").ValueOrDie(); + auto unsupported = + arrow::ipc::internal::json::ArrayFromJSON(arrow::uint64(), "[1,2]").ValueOrDie(); + auto batch = arrow::StructArray::Make({keys, unsupported}, + std::vector{"key", "unsupported"}) + .ValueOrDie(); + std::vector> invalid_predicates = { + std::make_shared(Equal::Instance(), 0, "key", FieldType::BIGINT, + std::vector{}), + PredicateBuilder::Equal(1, "unsupported", FieldType::BIGINT, Literal(int64_t{1}))}; + for (const auto& invalid : invalid_predicates) { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr conjunction, + PredicateBuilder::And( + {PredicateBuilder::Equal(0, "key", FieldType::BIGINT, Literal(int64_t{9})), + invalid})); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr disjunction, + PredicateBuilder::Or( + {PredicateBuilder::IsNotNull(0, "key", FieldType::BIGINT), invalid})); + for (const auto& predicate : {conjunction, disjunction}) { + auto filter = std::dynamic_pointer_cast(predicate); + auto expected = TestEager(predicate, *batch, arrow::default_memory_pool()); + auto actual = filter->Test(*batch, arrow::default_memory_pool()); + ASSERT_FALSE(expected.ok()); + ASSERT_FALSE(actual.ok()); + ASSERT_EQ(actual.status().ToString(), expected.status().ToString()); + } + } +} + +TEST_F(PredicateTest, TestCandidateSelectionDictionaryColumn) { + auto indices = + arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[0,1,null,0,1]").ValueOrDie(); + auto dictionary = + arrow::ipc::internal::json::ArrayFromJSON(arrow::utf8(), R"(["a","b"])").ValueOrDie(); + auto values = arrow::DictionaryArray::FromArrays(indices, dictionary).ValueOrDie(); + auto batch = arrow::StructArray::Make({values}, std::vector{"value"}).ValueOrDie(); + auto a = std::make_shared(0, "value", FieldType::STRING, + Literal(FieldType::STRING, "a", 1)); + auto b = std::make_shared(0, "value", FieldType::STRING, + Literal(FieldType::STRING, "b", 1)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr predicate, PredicateBuilder::Or({a, b})); + auto filter = std::dynamic_pointer_cast(predicate); + ASSERT_OK_AND_ASSIGN(std::vector actual, + filter->Test(*batch, arrow::default_memory_pool())); + ASSERT_EQ(actual, std::vector({1, 1, 0, 1, 1})); + ASSERT_EQ(b->selections, std::vector>({{1, 2, 4}})); +} + +TEST_F(PredicateTest, TestCandidateSelectionMatchesEagerEvaluation) { + auto keys = arrow::ipc::internal::json::ArrayFromJSON(arrow::int64(), "[1,2,null,1,2,1,3,null]") + .ValueOrDie(); + auto values = arrow::ipc::internal::json::ArrayFromJSON( + arrow::utf8(), R"(["a","b",null,"b","c",null,"a","c"])") + .ValueOrDie(); + auto batch = arrow::StructArray::Make({keys, values}, std::vector{"key", "value"}) + .ValueOrDie(); + for (int64_t key = 0; key <= 3; ++key) { + auto key_pred = PredicateBuilder::Equal(0, "key", FieldType::BIGINT, Literal(key)); + 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(std::shared_ptr inner_or, PredicateBuilder::Or({a, b})); + ASSERT_OK_AND_ASSIGN(std::shared_ptr inner_and, + PredicateBuilder::And({key_pred, a})); + ASSERT_OK_AND_ASSIGN(std::shared_ptr root_and, + PredicateBuilder::And({key_pred, inner_or})); + ASSERT_OK_AND_ASSIGN(std::shared_ptr root_or, + PredicateBuilder::Or({inner_and, b})); + for (const auto& predicate : {root_and, root_or}) { + auto filter = std::dynamic_pointer_cast(predicate); + for (int64_t offset = 0; offset <= batch->length(); ++offset) { + auto slice = batch->Slice(offset); + ASSERT_OK_AND_ASSIGN(std::vector expected, + TestEager(predicate, *slice, arrow::default_memory_pool())); + ASSERT_OK_AND_ASSIGN(std::vector actual, + filter->Test(*slice, arrow::default_memory_pool())); + ASSERT_EQ(actual, expected); + } + } + } +} + +// Opt-in microbenchmark: no timing thresholds in CI. The reference and candidate paths use +// identical predicates, data and allocator, alternating execution order between rounds. +TEST_F(PredicateTest, DISABLED_BenchmarkCandidateSelection) { + constexpr int64_t kRows = 8192; + constexpr int32_t kIterations = 10; + for (int64_t key_period : {int64_t{8192}, int64_t{16}, int64_t{1}}) { + arrow::Int64Builder key_builder; + arrow::StringBuilder value_builder; + for (int64_t i = 0; i < kRows; ++i) { + ASSERT_TRUE(key_builder.Append(i % key_period).ok()); + ASSERT_TRUE(value_builder.Append("q15").ok()); + } + auto keys = key_builder.Finish().ValueOrDie(); + auto values = value_builder.Finish().ValueOrDie(); + auto batch = + arrow::StructArray::Make({keys, values}, std::vector{"key", "value"}) + .ValueOrDie(); + std::vector> alternatives; + for (int32_t i = 0; i < 16; ++i) { + const std::string value = fmt::format("q{}", i); + alternatives.push_back( + PredicateBuilder::Equal(1, "value", FieldType::STRING, + Literal(FieldType::STRING, value.data(), value.size()))); + } + ASSERT_OK_AND_ASSIGN(std::shared_ptr inner, PredicateBuilder::Or(alternatives)); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr predicate, + PredicateBuilder::And( + {PredicateBuilder::Equal(0, "key", FieldType::BIGINT, Literal(int64_t{0})), + inner})); + auto filter = std::dynamic_pointer_cast(predicate); + ASSERT_OK_AND_ASSIGN(std::vector expected, + TestEager(predicate, *batch, arrow::default_memory_pool())); + for (int32_t round = 0; round < 6; ++round) { + for (int32_t pass = 0; pass < 2; ++pass) { + const bool eager = (round + pass) % 2 == 0; + const auto start = std::chrono::steady_clock::now(); + for (int32_t i = 0; i < kIterations; ++i) { + ASSERT_OK_AND_ASSIGN( + std::vector actual, + eager ? TestEager(predicate, *batch, arrow::default_memory_pool()) + : filter->Test(*batch, arrow::default_memory_pool())); + ASSERT_EQ(actual, expected); + } + const double elapsed = std::chrono::duration( + std::chrono::steady_clock::now() - start) + .count() / + kIterations; + std::cout << "candidate_benchmark period=" << key_period << " round=" << round + << " eager=" << eager << " ms=" << elapsed << std::endl; + } + } + } +} + TEST_F(PredicateTest, TestBetween) { auto bigint_type = arrow::int64(); auto predicate_base = PredicateBuilder::Between(/*field_index=*/0, /*field_name=*/"f0", From c63ee6248cc278744c1aec79848ea8f5c5387373 Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Tue, 8 Sep 2026 20:03:30 +0800 Subject: [PATCH 2/6] perf(predicate): avoid boxing all columns for field bounds checks --- .../common/predicate/leaf_predicate_impl.h | 8 +++---- .../common/predicate/predicate_test.cpp | 24 +++++++++++++++++++ 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/src/paimon/common/predicate/leaf_predicate_impl.h b/src/paimon/common/predicate/leaf_predicate_impl.h index 8707645d0..d119d2995 100644 --- a/src/paimon/common/predicate/leaf_predicate_impl.h +++ b/src/paimon/common/predicate/leaf_predicate_impl.h @@ -45,10 +45,10 @@ class LeafPredicateImpl : public LeafPredicate, public PredicateFilter { Result> Test(const arrow::Array& array, arrow::MemoryPool* pool) const override { const auto& struct_array = checked_cast(array); - if (field_index_ >= static_cast(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); @@ -58,10 +58,10 @@ class LeafPredicateImpl : public LeafPredicate, public PredicateFilter { const std::vector& selection, arrow::MemoryPool* pool) const override { const auto& struct_array = checked_cast(array); - if (field_index_ >= static_cast(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_); // Preserve eager conversion/comparison errors, even when the offending row is not diff --git a/src/paimon/common/predicate/predicate_test.cpp b/src/paimon/common/predicate/predicate_test.cpp index 26b9684ac..4c18f4af9 100644 --- a/src/paimon/common/predicate/predicate_test.cpp +++ b/src/paimon/common/predicate/predicate_test.cpp @@ -1229,6 +1229,30 @@ TEST_F(PredicateTest, TestNestedCandidateSelectionWithSliceAndNulls) { } } +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 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( + 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({0, 0, 1})); + ASSERT_OK_AND_ASSIGN(auto selected, predicate->TestSelected(*sliced, {1, 2}, pool)); + ASSERT_EQ(selected, std::vector({0, 1})); + auto invalid = std::dynamic_pointer_cast( + 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"); + ASSERT_NOK_WITH_MSG(invalid->TestSelected(*sliced, {}, pool), + "field index 64 exceed field count 64"); +} + TEST_F(PredicateTest, TestEmptyCandidatesStillValidateFields) { auto keys = arrow::ipc::internal::json::ArrayFromJSON(arrow::int64(), "[1,2]").ValueOrDie(); auto batch = arrow::StructArray::Make({keys}, std::vector{"key"}).ValueOrDie(); From 73ddb8bffcd9c931871a08bd7adc486fbfffed76 Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Tue, 8 Sep 2026 20:12:12 +0800 Subject: [PATCH 3/6] perf(predicate): compare typed equality batches without boxing literals --- src/paimon/common/predicate/equal.cpp | 94 ++++++++++ src/paimon/common/predicate/equal.h | 5 + .../common/predicate/predicate_test.cpp | 175 ++++++++++++++++-- 3 files changed, 258 insertions(+), 16 deletions(-) diff --git a/src/paimon/common/predicate/equal.cpp b/src/paimon/common/predicate/equal.cpp index 4260808a4..935150ed9 100644 --- a/src/paimon/common/predicate/equal.cpp +++ b/src/paimon/common/predicate/equal.cpp @@ -18,11 +18,105 @@ #include "paimon/common/predicate/equal.h" +#include + +#include "arrow/array/array_binary.h" +#include "arrow/array/array_primitive.h" #include "paimon/common/predicate/not_equal.h" +#include "paimon/common/utils/checked_cast.h" namespace paimon { class LeafFunction; +namespace { +template +std::vector TestEqualValues(const arrow::Array& array, Matches matches) { + std::vector result(array.length(), false); + if (array.null_count() == 0) { + for (int64_t i = 0; i < array.length(); ++i) { + result[i] = matches(i); + } + } else { + for (int64_t i = 0; i < array.length(); ++i) { + result[i] = !array.IsNull(i) && matches(i); + } + } + return result; +} + +template +std::vector TestEqualPrimitive(const arrow::Array& array, const Literal& literal) { + const auto& values = checked_cast(array); + const ValueType expected = literal.GetValue(); + return TestEqualValues(array, [&](int64_t i) { return values.Value(i) == expected; }); +} + +template +std::vector TestEqualBinary(const arrow::Array& array, const Literal& literal) { + const auto& values = checked_cast(array); + const std::string expected = literal.GetValue(); + return TestEqualValues(array, [&](int64_t i) { + const auto value = values.GetView(i); + return std::string_view(value.data(), value.size()) == expected; + }); +} +} // namespace + +Result> Equal::Test(const arrow::Array& array, + const std::vector& literals, + arrow::MemoryPool* pool) const { + if (!literals.empty() && !literals[0].IsNull()) { + const Literal& literal = literals[0]; + // Only bypass Literal conversion for types with identical equality semantics. Keep the + // existing conversion and validation for all other types, including mismatched literals. + switch (array.type_id()) { + case arrow::Type::BOOL: + if (literal.GetType() == FieldType::BOOLEAN) { + return TestEqualPrimitive(array, literal); + } + break; + case arrow::Type::INT8: + if (literal.GetType() == FieldType::TINYINT) { + return TestEqualPrimitive(array, literal); + } + break; + case arrow::Type::INT16: + if (literal.GetType() == FieldType::SMALLINT) { + return TestEqualPrimitive(array, literal); + } + break; + case arrow::Type::INT32: + if (literal.GetType() == FieldType::INT) { + return TestEqualPrimitive(array, literal); + } + break; + case arrow::Type::INT64: + if (literal.GetType() == FieldType::BIGINT) { + return TestEqualPrimitive(array, literal); + } + break; + case arrow::Type::DATE32: + if (literal.GetType() == FieldType::DATE) { + return TestEqualPrimitive(array, literal); + } + break; + case arrow::Type::STRING: + if (literal.GetType() == FieldType::STRING) { + return TestEqualBinary(array, literal); + } + break; + case arrow::Type::BINARY: + if (literal.GetType() == FieldType::BINARY) { + return TestEqualBinary(array, literal); + } + break; + default: + break; + } + } + return NullFalseLeafBinaryFunction::Test(array, literals, pool); +} + const LeafFunction* Equal::Negate() const { return &NotEqual::Instance(); } diff --git a/src/paimon/common/predicate/equal.h b/src/paimon/common/predicate/equal.h index 2c9b12cf3..640f1a1d9 100644 --- a/src/paimon/common/predicate/equal.h +++ b/src/paimon/common/predicate/equal.h @@ -33,6 +33,11 @@ class LeafFunction; /// A `NullFalseLeafBinaryFunction` to eval equal. class Equal : public NullFalseLeafBinaryFunction { public: + using NullFalseLeafBinaryFunction::Test; + + Result> Test(const arrow::Array& array, const std::vector& literals, + arrow::MemoryPool* pool) const override; + static const Equal& Instance() { static const Equal instance = Equal(); return instance; diff --git a/src/paimon/common/predicate/predicate_test.cpp b/src/paimon/common/predicate/predicate_test.cpp index 4c18f4af9..378367b7a 100644 --- a/src/paimon/common/predicate/predicate_test.cpp +++ b/src/paimon/common/predicate/predicate_test.cpp @@ -56,17 +56,78 @@ class Array; namespace paimon::test { namespace { +// Retain the Literal-based batch path as an independent reference for typed equality. +class LiteralBatchEqual : public NullFalseLeafBinaryFunction { + public: + using NullFalseLeafBinaryFunction::Test; + Result Test(const Literal& value, const Literal& literal) const override { + return Equal::Instance().Test(value, literal); + } + Result Test(int64_t rows, const Literal& min, const Literal& max, + const std::optional& nulls, const Literal& literal) const override { + return Equal::Instance().Test(rows, min, max, nulls, literal); + } + Type GetType() const override { + return Type::EQUAL; + } + std::string ToString() const override { + return "LiteralBatchEqual"; + } + const LeafFunction* Negate() const override { + return Equal::Instance().Negate(); + } +}; + +void CheckEqualBatch(const std::shared_ptr& array, + const std::vector& literals) { + const LiteralBatchEqual reference; + const LeafFunction& candidate = Equal::Instance(); + for (int64_t offset = 0; offset <= array->length(); ++offset) { + for (int64_t length = 0; length <= array->length() - offset; ++length) { + const auto slice = array->Slice(offset, length); + auto expected = reference.Test(*slice, literals, arrow::default_memory_pool()); + auto actual = candidate.Test(*slice, literals, arrow::default_memory_pool()); + ASSERT_EQ(actual.ok(), expected.ok()); + if (expected.ok()) { + ASSERT_EQ(actual.value(), expected.value()); + } else { + ASSERT_EQ(actual.status().ToString(), expected.status().ToString()); + } + } + } +} + // Reference implementation of the eager batch evaluation used before candidate selection. Result> TestEager(const std::shared_ptr& predicate, - const arrow::Array& array, arrow::MemoryPool* pool) { + const arrow::Array& array, arrow::MemoryPool* pool, + bool literal_equal = false, bool box_all_fields = false) { auto compound = std::dynamic_pointer_cast(predicate); if (!compound) { + if (literal_equal || box_all_fields) { + auto leaf = std::dynamic_pointer_cast(predicate); + if (!leaf || leaf->GetFunction().GetType() != Function::Type::EQUAL) { + return Status::Invalid("benchmark reference requires equality leaves"); + } + const auto& batch = checked_cast(array); + const int32_t field_count = + box_all_fields ? static_cast(batch.fields().size()) : batch.num_fields(); + if (leaf->FieldIndex() >= field_count) { + return Status::Invalid("benchmark field index out of bounds"); + } + const auto& field = batch.field(leaf->FieldIndex()); + const LiteralBatchEqual reference; + const LeafFunction& legacy = reference; + const LeafFunction& typed = Equal::Instance(); + const LeafFunction& function = literal_equal ? legacy : typed; + return function.Test(*field, leaf->Literals(), pool); + } return std::dynamic_pointer_cast(predicate)->Test(array, pool); } const bool is_and = predicate->GetFunction().GetType() == Function::Type::AND; std::vector result(array.length(), is_and); for (const auto& child : compound->Children()) { - PAIMON_ASSIGN_OR_RAISE(std::vector matches, TestEager(child, array, pool)); + PAIMON_ASSIGN_OR_RAISE(std::vector matches, + TestEager(child, array, pool, literal_equal, box_all_fields)); for (size_t i = 0; i < result.size(); ++i) { result[i] = is_and ? (result[i] & matches[i]) : (result[i] | matches[i]); } @@ -1270,17 +1331,25 @@ TEST_F(PredicateTest, TestEmptyCandidatesStillValidateFields) { } } -TEST_F(PredicateTest, TestEmptyCandidatesStillValidateLiteralsAndTypes) { +TEST_F(PredicateTest, TestCandidateSelectionPreservesEvaluationErrors) { auto keys = arrow::ipc::internal::json::ArrayFromJSON(arrow::int64(), "[1,2]").ValueOrDie(); auto unsupported = arrow::ipc::internal::json::ArrayFromJSON(arrow::uint64(), "[1,2]").ValueOrDie(); - auto batch = arrow::StructArray::Make({keys, unsupported}, - std::vector{"key", "unsupported"}) + auto strings = + arrow::ipc::internal::json::ArrayFromJSON(arrow::utf8(), R"(["a","b"])").ValueOrDie(); + auto batch = arrow::StructArray::Make({keys, unsupported, strings}, + std::vector{"key", "unsupported", "text"}) .ValueOrDie(); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr invalid_like, + PredicateBuilder::Like(2, "text", FieldType::STRING, Literal(FieldType::STRING, "\\q", 2))); std::vector> invalid_predicates = { std::make_shared(Equal::Instance(), 0, "key", FieldType::BIGINT, std::vector{}), - PredicateBuilder::Equal(1, "unsupported", FieldType::BIGINT, Literal(int64_t{1}))}; + PredicateBuilder::Equal(1, "unsupported", FieldType::BIGINT, Literal(int64_t{1})), + PredicateBuilder::Equal(0, "key", FieldType::BIGINT, + Literal(FieldType::STRING, "wrong-type", 10)), + invalid_like}; for (const auto& invalid : invalid_predicates) { ASSERT_OK_AND_ASSIGN( std::shared_ptr conjunction, @@ -1291,7 +1360,12 @@ TEST_F(PredicateTest, TestEmptyCandidatesStillValidateLiteralsAndTypes) { std::shared_ptr disjunction, PredicateBuilder::Or( {PredicateBuilder::IsNotNull(0, "key", FieldType::BIGINT), invalid})); - for (const auto& predicate : {conjunction, disjunction}) { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr partial, + PredicateBuilder::And( + {PredicateBuilder::Equal(0, "key", FieldType::BIGINT, Literal(int64_t{2})), + invalid})); + for (const auto& predicate : {conjunction, disjunction, partial}) { auto filter = std::dynamic_pointer_cast(predicate); auto expected = TestEager(predicate, *batch, arrow::default_memory_pool()); auto actual = filter->Test(*batch, arrow::default_memory_pool()); @@ -1356,9 +1430,10 @@ TEST_F(PredicateTest, TestCandidateSelectionMatchesEagerEvaluation) { } } -// Opt-in microbenchmark: no timing thresholds in CI. The reference and candidate paths use -// identical predicates, data and allocator, alternating execution order between rounds. -TEST_F(PredicateTest, DISABLED_BenchmarkCandidateSelection) { +// Opt-in microbenchmark: no timing thresholds in CI. Compare the original eager Literal path, +// bounds checks alone, bounds checks plus typed equality, and all three optimizations together. +// These are incremental variants, not additive or independently attributable speedups. +TEST_F(PredicateTest, DISABLED_BenchmarkBatchEvaluation) { constexpr int64_t kRows = 8192; constexpr int32_t kIterations = 10; for (int64_t key_period : {int64_t{8192}, int64_t{16}, int64_t{1}}) { @@ -1390,22 +1465,25 @@ TEST_F(PredicateTest, DISABLED_BenchmarkCandidateSelection) { ASSERT_OK_AND_ASSIGN(std::vector expected, TestEager(predicate, *batch, arrow::default_memory_pool())); for (int32_t round = 0; round < 6; ++round) { - for (int32_t pass = 0; pass < 2; ++pass) { - const bool eager = (round + pass) % 2 == 0; + for (int32_t pass = 0; pass < 4; ++pass) { + const int32_t variant = round % 2 == 0 ? pass : 3 - pass; const auto start = std::chrono::steady_clock::now(); for (int32_t i = 0; i < kIterations; ++i) { ASSERT_OK_AND_ASSIGN( std::vector actual, - eager ? TestEager(predicate, *batch, arrow::default_memory_pool()) - : filter->Test(*batch, arrow::default_memory_pool())); + variant < 3 ? TestEager(predicate, *batch, arrow::default_memory_pool(), + /*literal_equal=*/variant < 2, + /*box_all_fields=*/variant == 0) + : filter->Test(*batch, arrow::default_memory_pool())); ASSERT_EQ(actual, expected); } const double elapsed = std::chrono::duration( std::chrono::steady_clock::now() - start) .count() / kIterations; - std::cout << "candidate_benchmark period=" << key_period << " round=" << round - << " eager=" << eager << " ms=" << elapsed << std::endl; + std::cout << "batch_evaluation_benchmark period=" << key_period + << " round=" << round << " variant=" << variant << " ms=" << elapsed + << std::endl; } } } @@ -2119,6 +2197,71 @@ TEST_F(PredicateTest, TestPredicateToString) { } } +TEST_F(PredicateTest, TestTypedEqualBatchMatchesLiteralEvaluation) { + const std::vector, std::string>> cases = { + {arrow::boolean(), "[true, null, false, true]"}, + {arrow::int8(), "[-128, null, 0, 127]"}, + {arrow::int16(), "[-32768, null, 0, 32767]"}, + {arrow::int32(), "[-2147483648, null, 0, 2147483647]"}, + {arrow::int64(), "[-9223372036854775808, null, 0, 9223372036854775807]"}, + {arrow::date32(), "[-100, null, 0, 100]"}, + {arrow::utf8(), R"(["", null, "a\u0000b", "a\u0000c", "é"])"}, + {arrow::binary(), R"(["", null, "a\u0000b", "a\u0000c", "é"])"}, + {arrow::float64(), "[-0.0, null, 0.0, 1.5]"}, + {arrow::timestamp(arrow::TimeUnit::MICRO), "[-1, null, 0, 1001]"}, + {arrow::decimal128(10, 2), R"(["-1.25", null, "0.00", "1.25"])"}}; + for (const auto& [type, json] : cases) { + auto array = arrow::ipc::internal::json::ArrayFromJSON(type, json).ValueOrDie(); + ASSERT_OK_AND_ASSIGN(auto values, + LiteralConverter::ConvertLiteralsFromArray(*array, false)); + CheckEqualBatch(array, {}); + for (const Literal& value : values) { + CheckEqualBatch(array, {value}); + // The binary function has historically used only the first literal. + CheckEqualBatch(array, {value, Literal(int64_t{999})}); + } + CheckEqualBatch(array, {Literal(FieldType::STRING, "mismatch", 8)}); + CheckEqualBatch(array, {Literal(int32_t{123})}); + } +} + +TEST_F(PredicateTest, TestTypedEqualFallback) { + arrow::DoubleBuilder builder; + ASSERT_TRUE(builder + .AppendValues({std::numeric_limits::quiet_NaN(), + std::numeric_limits::infinity(), -0.0, 0.0}) + .ok()); + auto doubles = builder.Finish().ValueOrDie(); + CheckEqualBatch(doubles, {Literal(std::numeric_limits::quiet_NaN())}); + CheckEqualBatch(doubles, {Literal(-0.0)}); + auto indices = arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[0, 1, null, 2, 0]") + .ValueOrDie(); + auto dictionary = + arrow::ipc::internal::json::ArrayFromJSON(arrow::utf8(), R"(["a", "b", null])") + .ValueOrDie(); + auto encoded = arrow::DictionaryArray::FromArrays(indices, dictionary).ValueOrDie(); + CheckEqualBatch(encoded, {Literal(FieldType::STRING, "a", 1)}); + auto unsupported = + arrow::ipc::internal::json::ArrayFromJSON(arrow::uint32(), "[1, null, 2]").ValueOrDie(); + CheckEqualBatch(unsupported, {Literal(int32_t{1})}); + CheckEqualBatch(unsupported, {Literal(FieldType::INT)}); +} + +TEST_F(PredicateTest, TestTypedEqualBinaryAndBitmapOffsets) { + arrow::BinaryBuilder builder; + const std::string bytes("\x00\x80\xff", 3); + for (int32_t i = 0; i < 19; ++i) { + if (i % 3 == 0) { + ASSERT_TRUE(builder.AppendNull().ok()); + } else { + ASSERT_TRUE(builder.Append(i % 2 == 0 ? bytes : std::string()).ok()); + } + } + auto array = builder.Finish().ValueOrDie(); + CheckEqualBatch(array, {Literal(FieldType::BINARY, bytes.data(), bytes.size())}); + CheckEqualBatch(array, {Literal(FieldType::BINARY, "", 0)}); +} + TEST_F(PredicateTest, TestBuildAndOr) { { // literals cannot be empty From c0a24672f8ab13cb91013905a306601d297e912f Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Tue, 8 Sep 2026 21:15:18 +0800 Subject: [PATCH 4/6] test(predicate): cover nested batch filtering end to end --- test/inte/write_and_read_inte_test.cpp | 107 +++++++++++++++++++++++++ 1 file changed, 107 insertions(+) diff --git a/test/inte/write_and_read_inte_test.cpp b/test/inte/write_and_read_inte_test.cpp index 7ce6f2e62..6bcc33692 100644 --- a/test/inte/write_and_read_inte_test.cpp +++ b/test/inte/write_and_read_inte_test.cpp @@ -357,6 +357,113 @@ 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 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)); + // With four rows per batch, the AND predicate sees empty, partial, full, empty, + // partial, and full matches. Rows 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 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> 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::vector>> 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::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{"_VALUE_KIND", "payload", "value", "id", "key"}) + .ValueOrDie(); + const auto unfiltered = std::make_shared(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(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") { From da745daf8c9f1c15b79d457305178392679aa88f Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Wed, 9 Sep 2026 14:36:25 +0800 Subject: [PATCH 5/6] perf(predicate): evaluate temporal and decimal candidates --- benchmark/CMakeLists.txt | 8 + benchmark/predicate_batch_benchmark.cpp | 190 ++++++++++++++++++ .../common/predicate/leaf_predicate_impl.h | 7 +- .../common/predicate/predicate_test.cpp | 153 +++++++------- 4 files changed, 279 insertions(+), 79 deletions(-) create mode 100644 benchmark/predicate_batch_benchmark.cpp diff --git a/benchmark/CMakeLists.txt b/benchmark/CMakeLists.txt index 0d4dc28cf..13a519083 100644 --- a/benchmark/CMakeLists.txt +++ b/benchmark/CMakeLists.txt @@ -64,6 +64,14 @@ if(PAIMON_BUILD_BENCHMARKS) EXTRA_INCLUDES ${CMAKE_SOURCE_DIR}) + add_paimon_benchmark(predicate_batch_benchmark + SOURCES + predicate_batch_benchmark.cpp + STATIC_LINK_LIBS + paimon_shared + test_utils_static + ${PAIMON_BENCHMARK_LINK_TOOLCHAIN}) + add_paimon_benchmark(parquet_format_benchmark SOURCES parquet_format_benchmark.cpp diff --git a/benchmark/predicate_batch_benchmark.cpp b/benchmark/predicate_batch_benchmark.cpp new file mode 100644 index 000000000..bf0c77122 --- /dev/null +++ b/benchmark/predicate_batch_benchmark.cpp @@ -0,0 +1,190 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +// Incremental batch-predicate variants: original Literal conversion, cheap field bounds, +// typed equality, and candidate selection. Run with --benchmark_repetitions=6 to report +// aggregate timings. These are microbenchmarks, not end-to-end read latencies. + +#include +#include +#include +#include +#include + +#include "arrow/api.h" +#include "benchmark/benchmark.h" +#include "fmt/format.h" +#include "paimon/common/predicate/equal.h" +#include "paimon/common/predicate/leaf_predicate_impl.h" +#include "paimon/common/predicate/predicate_filter.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/predicate/compound_predicate.h" +#include "paimon/predicate/predicate_builder.h" + +namespace paimon { +namespace { +class LiteralBatchEqual : public NullFalseLeafBinaryFunction { + public: + using NullFalseLeafBinaryFunction::Test; + Result Test(const Literal& value, const Literal& literal) const override { + return Equal::Instance().Test(value, literal); + } + Result Test(int64_t rows, const Literal& min, const Literal& max, + const std::optional& nulls, const Literal& literal) const override { + return Equal::Instance().Test(rows, min, max, nulls, literal); + } + Type GetType() const override { + return Type::EQUAL; + } + std::string ToString() const override { + return "LiteralBatchEqual"; + } + const LeafFunction* Negate() const override { + return Equal::Instance().Negate(); + } +}; + +Result> TestEager(const std::shared_ptr& predicate, + const arrow::Array& array, arrow::MemoryPool* pool, + bool literal_equal = false, bool box_all_fields = false) { + auto compound = std::dynamic_pointer_cast(predicate); + if (!compound) { + if (literal_equal || box_all_fields) { + auto leaf = std::dynamic_pointer_cast(predicate); + if (!leaf || leaf->GetFunction().GetType() != Function::Type::EQUAL) { + return Status::Invalid("benchmark reference requires equality leaves"); + } + const auto& batch = checked_cast(array); + const int32_t field_count = + box_all_fields ? static_cast(batch.fields().size()) : batch.num_fields(); + if (leaf->FieldIndex() >= field_count) { + return Status::Invalid("benchmark field index out of bounds"); + } + const auto& field = batch.field(leaf->FieldIndex()); + const LiteralBatchEqual reference; + const LeafFunction& legacy = reference; + const LeafFunction& typed = Equal::Instance(); + const LeafFunction& function = literal_equal ? legacy : typed; + return function.Test(*field, leaf->Literals(), pool); + } + return std::dynamic_pointer_cast(predicate)->Test(array, pool); + } + const bool is_and = predicate->GetFunction().GetType() == Function::Type::AND; + std::vector result(array.length(), is_and); + for (const auto& child : compound->Children()) { + PAIMON_ASSIGN_OR_RAISE(std::vector matches, + TestEager(child, array, pool, literal_equal, box_all_fields)); + for (size_t i = 0; i < result.size(); ++i) { + result[i] = is_and ? (result[i] & matches[i]) : (result[i] | matches[i]); + } + } + return result; +} + +struct BatchInput { + std::shared_ptr array; + std::shared_ptr predicate; +}; + +Result MakeInput(int64_t period) { + constexpr int64_t kRows = 8192; + arrow::Int64Builder keys; + arrow::StringBuilder values; + for (int64_t i = 0; i < kRows; ++i) { + PAIMON_RETURN_NOT_OK_FROM_ARROW(keys.Append(i % period)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(values.Append("q15")); + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr key_array, keys.Finish()); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr value_array, values.Finish()); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr array, + arrow::StructArray::Make({key_array, value_array}, + std::vector{"key", "value"})); + std::vector> alternatives; + for (int32_t i = 0; i < 16; ++i) { + const std::string value = fmt::format("q{}", i); + alternatives.push_back(PredicateBuilder::Equal( + 1, "value", FieldType::STRING, Literal(FieldType::STRING, value.data(), value.size()))); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr inner, PredicateBuilder::Or(alternatives)); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr predicate, + PredicateBuilder::And( + {PredicateBuilder::Equal(0, "key", FieldType::BIGINT, Literal(int64_t{0})), inner})); + return BatchInput{array, predicate}; +} + +void BatchEvaluation(benchmark::State& state, int32_t variant) { + auto input_result = MakeInput(state.range(0)); + if (!input_result.ok()) { + state.SkipWithError(input_result.status().ToString().c_str()); + return; + } + const auto& input = input_result.value(); + auto pool = arrow::default_memory_pool(); + auto reference = TestEager(input.predicate, *input.array, pool, + /*literal_equal=*/true, /*box_all_fields=*/true); + if (!reference.ok()) { + state.SkipWithError(reference.status().ToString().c_str()); + return; + } + auto filter = std::dynamic_pointer_cast(input.predicate); + for (auto _ : state) { + auto actual = + variant < 3 ? TestEager(input.predicate, *input.array, pool, + /*literal_equal=*/variant < 2, /*box_all_fields=*/variant == 0) + : filter->Test(*input.array, pool); + if (!actual.ok()) { + state.SkipWithError(actual.status().ToString().c_str()); + break; + } + benchmark::DoNotOptimize(actual.value().data()); + if (actual.value() != reference.value()) { + state.SkipWithError("predicate result differs from eager Literal evaluation"); + break; + } + } + state.SetItemsProcessed(state.iterations() * input.array->length()); +} + +BENCHMARK_CAPTURE(BatchEvaluation, Original, 0) + ->Arg(8192) + ->Arg(16) + ->Arg(1) + ->Unit(benchmark::kMillisecond); +BENCHMARK_CAPTURE(BatchEvaluation, FieldBounds, 1) + ->Arg(8192) + ->Arg(16) + ->Arg(1) + ->Unit(benchmark::kMillisecond); +BENCHMARK_CAPTURE(BatchEvaluation, TypedEquality, 2) + ->Arg(8192) + ->Arg(16) + ->Arg(1) + ->Unit(benchmark::kMillisecond); +BENCHMARK_CAPTURE(BatchEvaluation, CandidateSelection, 3) + ->Arg(8192) + ->Arg(16) + ->Arg(1) + ->Unit(benchmark::kMillisecond); +} // namespace +} // namespace paimon + +BENCHMARK_MAIN(); diff --git a/src/paimon/common/predicate/leaf_predicate_impl.h b/src/paimon/common/predicate/leaf_predicate_impl.h index d119d2995..91a531f48 100644 --- a/src/paimon/common/predicate/leaf_predicate_impl.h +++ b/src/paimon/common/predicate/leaf_predicate_impl.h @@ -66,11 +66,8 @@ class LeafPredicateImpl : public LeafPredicate, public PredicateFilter { const auto& field_array = struct_array.field(field_index_); // Preserve eager conversion/comparison errors, even when the offending row is not // selected. Dictionary and unsupported physical types also retain the original path. - // LIKE validates escapes while examining values; temporal/decimal conversions may fail - // on rows outside the candidate set as well. - if (leaf_function_.GetType() == Function::Type::LIKE || - field_array->type_id() == arrow::Type::TIMESTAMP || - field_array->type_id() == arrow::Type::DECIMAL128) { + // LIKE validates escapes while examining values, including unselected rows. + if (leaf_function_.GetType() == Function::Type::LIKE) { return PredicateFilter::TestSelected(array, selection, pool); } auto field_type = FieldTypeUtils::ConvertToFieldType(field_array->type_id()); diff --git a/src/paimon/common/predicate/predicate_test.cpp b/src/paimon/common/predicate/predicate_test.cpp index 378367b7a..9d0dff616 100644 --- a/src/paimon/common/predicate/predicate_test.cpp +++ b/src/paimon/common/predicate/predicate_test.cpp @@ -18,10 +18,8 @@ #include "paimon/predicate/predicate.h" -#include #include #include -#include #include #include #include @@ -99,35 +97,15 @@ void CheckEqualBatch(const std::shared_ptr& array, // Reference implementation of the eager batch evaluation used before candidate selection. Result> TestEager(const std::shared_ptr& predicate, - const arrow::Array& array, arrow::MemoryPool* pool, - bool literal_equal = false, bool box_all_fields = false) { + const arrow::Array& array, arrow::MemoryPool* pool) { auto compound = std::dynamic_pointer_cast(predicate); if (!compound) { - if (literal_equal || box_all_fields) { - auto leaf = std::dynamic_pointer_cast(predicate); - if (!leaf || leaf->GetFunction().GetType() != Function::Type::EQUAL) { - return Status::Invalid("benchmark reference requires equality leaves"); - } - const auto& batch = checked_cast(array); - const int32_t field_count = - box_all_fields ? static_cast(batch.fields().size()) : batch.num_fields(); - if (leaf->FieldIndex() >= field_count) { - return Status::Invalid("benchmark field index out of bounds"); - } - const auto& field = batch.field(leaf->FieldIndex()); - const LiteralBatchEqual reference; - const LeafFunction& legacy = reference; - const LeafFunction& typed = Equal::Instance(); - const LeafFunction& function = literal_equal ? legacy : typed; - return function.Test(*field, leaf->Literals(), pool); - } return std::dynamic_pointer_cast(predicate)->Test(array, pool); } const bool is_and = predicate->GetFunction().GetType() == Function::Type::AND; std::vector result(array.length(), is_and); for (const auto& child : compound->Children()) { - PAIMON_ASSIGN_OR_RAISE(std::vector matches, - TestEager(child, array, pool, literal_equal, box_all_fields)); + PAIMON_ASSIGN_OR_RAISE(std::vector matches, TestEager(child, array, pool)); for (size_t i = 0; i < result.size(); ++i) { result[i] = is_and ? (result[i] & matches[i]) : (result[i] | matches[i]); } @@ -135,6 +113,19 @@ Result> TestEager(const std::shared_ptr& predicate, return result; } +class RecordingBatchEqual : public LiteralBatchEqual { + public: + using LiteralBatchEqual::Test; + + Result> Test(const arrow::Array& array, const std::vector& literals, + arrow::MemoryPool* pool) const override { + batch_lengths.push_back(array.length()); + return LiteralBatchEqual::Test(array, literals, pool); + } + + mutable std::vector batch_lengths; +}; + class RecordingEqualPredicate : public LeafPredicateImpl { public: RecordingEqualPredicate(int32_t index, const std::string& name, FieldType type, @@ -1430,60 +1421,74 @@ TEST_F(PredicateTest, TestCandidateSelectionMatchesEagerEvaluation) { } } -// Opt-in microbenchmark: no timing thresholds in CI. Compare the original eager Literal path, -// bounds checks alone, bounds checks plus typed equality, and all three optimizations together. -// These are incremental variants, not additive or independently attributable speedups. -TEST_F(PredicateTest, DISABLED_BenchmarkBatchEvaluation) { - constexpr int64_t kRows = 8192; - constexpr int32_t kIterations = 10; - for (int64_t key_period : {int64_t{8192}, int64_t{16}, int64_t{1}}) { - arrow::Int64Builder key_builder; - arrow::StringBuilder value_builder; - for (int64_t i = 0; i < kRows; ++i) { - ASSERT_TRUE(key_builder.Append(i % key_period).ok()); - ASSERT_TRUE(value_builder.Append("q15").ok()); - } - auto keys = key_builder.Finish().ValueOrDie(); - auto values = value_builder.Finish().ValueOrDie(); +TEST_F(PredicateTest, TestTemporalAndDecimalCandidateSelection) { + const std::vector, std::string>> cases = { + {arrow::timestamp(arrow::TimeUnit::SECOND), "[-1001,null,0,1001,2002]"}, + {arrow::timestamp(arrow::TimeUnit::MILLI), "[-1001,null,0,1001,2002]"}, + {arrow::timestamp(arrow::TimeUnit::MICRO), "[-1001,null,0,1001,2002]"}, + {arrow::timestamp(arrow::TimeUnit::NANO), "[-1001,null,0,1001,2002]"}, + {arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"), "[-1001,null,0,1001,2002]"}, + {arrow::decimal128(10, 2), R"(["-1.25",null,"0.00","1.25","2.50"])"}, + {arrow::decimal128(15, 5), R"(["-1.25001",null,"0.00000","1.25001","2.50002"])"}}; + auto pool = arrow::default_memory_pool(); + for (const auto& [type, json] : cases) { + SCOPED_TRACE(type->ToString()); + auto values = arrow::ipc::internal::json::ArrayFromJSON(type, json).ValueOrDie(); + auto keys = + arrow::ipc::internal::json::ArrayFromJSON(arrow::int64(), "[1,0,1,0,1]").ValueOrDie(); auto batch = arrow::StructArray::Make({keys, values}, std::vector{"key", "value"}) .ValueOrDie(); - std::vector> alternatives; - for (int32_t i = 0; i < 16; ++i) { - const std::string value = fmt::format("q{}", i); - alternatives.push_back( - PredicateBuilder::Equal(1, "value", FieldType::STRING, - Literal(FieldType::STRING, value.data(), value.size()))); - } - ASSERT_OK_AND_ASSIGN(std::shared_ptr inner, PredicateBuilder::Or(alternatives)); - ASSERT_OK_AND_ASSIGN( - std::shared_ptr predicate, - PredicateBuilder::And( - {PredicateBuilder::Equal(0, "key", FieldType::BIGINT, Literal(int64_t{0})), - inner})); - auto filter = std::dynamic_pointer_cast(predicate); - ASSERT_OK_AND_ASSIGN(std::vector expected, - TestEager(predicate, *batch, arrow::default_memory_pool())); - for (int32_t round = 0; round < 6; ++round) { - for (int32_t pass = 0; pass < 4; ++pass) { - const int32_t variant = round % 2 == 0 ? pass : 3 - pass; - const auto start = std::chrono::steady_clock::now(); - for (int32_t i = 0; i < kIterations; ++i) { - ASSERT_OK_AND_ASSIGN( - std::vector actual, - variant < 3 ? TestEager(predicate, *batch, arrow::default_memory_pool(), - /*literal_equal=*/variant < 2, - /*box_all_fields=*/variant == 0) - : filter->Test(*batch, arrow::default_memory_pool())); - ASSERT_EQ(actual, expected); + ASSERT_OK_AND_ASSIGN(auto literals, + LiteralConverter::ConvertLiteralsFromArray(*values, false)); + const FieldType field_type = literals[0].GetType(); + // Observe the batch received by the leaf function: equal results alone would also + // pass with the old eager fallback and would not protect candidate-only evaluation. + RecordingBatchEqual recording; + LeafPredicateImpl recorded(recording, 1, "value", field_type, {literals[3]}); + auto slice = batch->Slice(1, 4); + ASSERT_OK_AND_ASSIGN(auto selected, recorded.TestSelected(*slice, {0, 2}, pool)); + ASSERT_EQ(selected, std::vector({0, 1})); + ASSERT_OK_AND_ASSIGN(auto empty, recorded.TestSelected(*slice, {}, pool)); + ASSERT_TRUE(empty.empty()); + ASSERT_EQ(recording.batch_lengths, std::vector({2, 0})); + + // Include a literal whose timestamp precision or decimal scale differs from the column. + literals.push_back(field_type == FieldType::TIMESTAMP ? Literal(Timestamp(1, 123456)) + : Literal(Decimal(12, 3, 1251))); + for (const Literal& literal : literals) { + const std::vector> leaves = { + PredicateBuilder::Equal(1, "value", field_type, literal), + PredicateBuilder::LessThan(1, "value", field_type, literal), + PredicateBuilder::In(1, "value", field_type, {literal, literals[3]}), + PredicateBuilder::NotIn(1, "value", field_type, {literal, literals[3]})}; + for (const auto& leaf : leaves) { + SCOPED_TRACE(leaf->ToString()); + auto key = + PredicateBuilder::Equal(0, "key", FieldType::BIGINT, Literal(int64_t{1})); + ASSERT_OK_AND_ASSIGN(auto conjunction, PredicateBuilder::And({key, leaf})); + ASSERT_OK_AND_ASSIGN(auto disjunction, PredicateBuilder::Or({key, leaf})); + for (const auto& predicate : {conjunction, disjunction}) { + auto filter = std::dynamic_pointer_cast(predicate); + for (int64_t offset = 0; offset <= batch->length(); ++offset) { + auto input = batch->Slice(offset); + ASSERT_OK_AND_ASSIGN(auto expected, TestEager(predicate, *input, pool)); + ASSERT_OK_AND_ASSIGN(auto actual, filter->Test(*input, pool)); + ASSERT_EQ(actual, expected); + for (const std::vector& selection : + {std::vector{}, input->length() > 1 + ? std::vector{1} + : std::vector{}}) { + ASSERT_OK_AND_ASSIGN(auto subset, + filter->TestSelected(*input, selection, pool)); + std::vector expected_subset; + for (int64_t row : selection) { + expected_subset.push_back(expected[row]); + } + ASSERT_EQ(subset, expected_subset); + } + } } - const double elapsed = std::chrono::duration( - std::chrono::steady_clock::now() - start) - .count() / - kIterations; - std::cout << "batch_evaluation_benchmark period=" << key_period - << " round=" << round << " variant=" << variant << " ms=" << elapsed - << std::endl; } } } From dbef2ae7a7b3b58dac4ee87ed417faac795957d4 Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Thu, 10 Sep 2026 10:21:37 +0800 Subject: [PATCH 6/6] style(predicate): satisfy clang-tidy type deduction checks --- src/paimon/common/predicate/equal.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/paimon/common/predicate/equal.cpp b/src/paimon/common/predicate/equal.cpp index 935150ed9..ea46216b9 100644 --- a/src/paimon/common/predicate/equal.cpp +++ b/src/paimon/common/predicate/equal.cpp @@ -47,14 +47,14 @@ std::vector TestEqualValues(const arrow::Array& array, Matches matches) { template std::vector TestEqualPrimitive(const arrow::Array& array, const Literal& literal) { const auto& values = checked_cast(array); - const ValueType expected = literal.GetValue(); + const auto expected = literal.GetValue(); return TestEqualValues(array, [&](int64_t i) { return values.Value(i) == expected; }); } template std::vector TestEqualBinary(const arrow::Array& array, const Literal& literal) { const auto& values = checked_cast(array); - const std::string expected = literal.GetValue(); + const auto expected = literal.GetValue(); return TestEqualValues(array, [&](int64_t i) { const auto value = values.GetView(i); return std::string_view(value.data(), value.size()) == expected;