wangyong9999 commented on code in PR #301:
URL: https://github.com/apache/paimon-cpp/pull/301#discussion_r3964266123
##########
src/paimon/common/predicate/leaf_predicate_impl.h:
##########
@@ -44,15 +45,49 @@ 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);
}
+ Result<std::vector<char>> TestSelected(const arrow::Array& array,
+ const std::vector<int64_t>&
selection,
+ arrow::MemoryPool* pool) const
override {
+ const auto& struct_array = checked_cast<const
arrow::StructArray&>(array);
+ if (field_index_ >= struct_array.num_fields()) {
+ return Status::Invalid(
+ fmt::format("field index {} exceed field count {} in struct
array", field_index_,
+ 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
+ // 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 ||
Review Comment:
The TIMESTAMP and DECIMAL128 arms guard a per-row failure that cannot
happen. `GetLiteralFromTimestampArray` and `GetLiteralFromDecimalArray` return
a plain `std::vector<Literal>` with no failure path, `Decimal`'s constructor
does not validate, and `Literal::CompareTo` on two same-type non-null
TIMESTAMP/DECIMAL values returns a value rather than a Status. The only per-row
Status a comparison leaf can raise is the type mismatch, and the literal loop
just below already routes that to the eager path.
So these two arms only cost: any AND/OR with a timestamp or decimal child
evaluates that child over the whole batch even when the candidate set is down
to a handful of rows. Suggest dropping both and keeping the LIKE arm, whose
escape validation really is reached per value.
##########
src/paimon/common/predicate/predicate_test.cpp:
##########
@@ -1149,6 +1250,245 @@ 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<std::string>{"key", "value"})
+ .ValueOrDie();
+ auto sliced = batch->Slice(1, 5);
+ auto key = std::make_shared<RecordingEqualPredicate>(0, "key",
FieldType::BIGINT,
+
Literal(int64_t{1}));
+ auto a = std::make_shared<RecordingEqualPredicate>(1, "value",
field_type,
+ Literal(field_type,
"a", 1));
+ auto b = std::make_shared<RecordingEqualPredicate>(1, "value",
field_type,
+ Literal(field_type,
"b", 1));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> alternatives,
PredicateBuilder::Or({a, b}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> predicate,
+ PredicateBuilder::And({key, alternatives}));
+ auto filter = std::dynamic_pointer_cast<PredicateFilter>(predicate);
+ arrow::ProxyMemoryPool pool(arrow::default_memory_pool());
+ ASSERT_OK_AND_ASSIGN(std::vector<char> actual, filter->Test(*sliced,
&pool));
+ ASSERT_EQ(actual, std::vector<char>({1, 0, 1, 0, 0}));
+ ASSERT_EQ(key->selections, std::vector<std::vector<int64_t>>({{0, 1,
2, 3, 4}}));
+ ASSERT_EQ(a->selections, std::vector<std::vector<int64_t>>({{0, 2,
3}}));
+ ASSERT_EQ(b->selections, std::vector<std::vector<int64_t>>({{2, 3}}));
+ ASSERT_GT(pool.max_memory(), 0);
+ ASSERT_EQ(pool.bytes_allocated(), 0);
+
+ ASSERT_OK_AND_ASSIGN(std::vector<char> selected,
+ filter->TestSelected(*sliced, {2, 4}, &pool));
+ ASSERT_EQ(selected, std::vector<char>({1, 0}));
+ ASSERT_OK_AND_ASSIGN(std::vector<char> empty,
filter->Test(*sliced->Slice(0, 0), &pool));
+ ASSERT_TRUE(empty.empty());
+ }
+}
+
+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}));
+ ASSERT_OK_AND_ASSIGN(auto selected, predicate->TestSelected(*sliced, {1,
2}, pool));
+ ASSERT_EQ(selected, std::vector<char>({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");
+ 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<std::string>{"key"}).ValueOrDie();
+ auto invalid = PredicateBuilder::Equal(1, "missing", FieldType::BIGINT,
Literal(int64_t{1}));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> conjunction,
+ PredicateBuilder::And(
+ {PredicateBuilder::Equal(0, "key", FieldType::BIGINT,
Literal(int64_t{9})), invalid}));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> disjunction,
+ PredicateBuilder::Or({PredicateBuilder::IsNotNull(0, "key",
FieldType::BIGINT), invalid}));
+ for (const auto& predicate : {conjunction, disjunction}) {
+ auto filter = std::dynamic_pointer_cast<PredicateFilter>(predicate);
+ ASSERT_NOK_WITH_MSG(filter->Test(*batch,
arrow::default_memory_pool()), "field index");
+ }
+}
+
+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 strings =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::utf8(),
R"(["a","b"])").ValueOrDie();
+ auto batch = arrow::StructArray::Make({keys, unsupported, strings},
+ std::vector<std::string>{"key",
"unsupported", "text"})
+ .ValueOrDie();
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> invalid_like,
+ PredicateBuilder::Like(2, "text", FieldType::STRING,
Literal(FieldType::STRING, "\\q", 2)));
+ std::vector<std::shared_ptr<Predicate>> invalid_predicates = {
+ std::make_shared<LeafPredicateImpl>(Equal::Instance(), 0, "key",
FieldType::BIGINT,
+ std::vector<Literal>{}),
+ 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<Predicate> conjunction,
+ PredicateBuilder::And(
+ {PredicateBuilder::Equal(0, "key", FieldType::BIGINT,
Literal(int64_t{9})),
+ invalid}));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> disjunction,
+ PredicateBuilder::Or(
+ {PredicateBuilder::IsNotNull(0, "key", FieldType::BIGINT),
invalid}));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> 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<PredicateFilter>(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<std::string>{"value"}).ValueOrDie();
+ auto a = std::make_shared<RecordingEqualPredicate>(0, "value",
FieldType::STRING,
+
Literal(FieldType::STRING, "a", 1));
+ auto b = std::make_shared<RecordingEqualPredicate>(0, "value",
FieldType::STRING,
+
Literal(FieldType::STRING, "b", 1));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> predicate,
PredicateBuilder::Or({a, b}));
+ auto filter = std::dynamic_pointer_cast<PredicateFilter>(predicate);
+ ASSERT_OK_AND_ASSIGN(std::vector<char> actual,
+ filter->Test(*batch, arrow::default_memory_pool()));
+ ASSERT_EQ(actual, std::vector<char>({1, 1, 0, 1, 1}));
+ ASSERT_EQ(b->selections, std::vector<std::vector<int64_t>>({{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<std::string>{"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<Predicate> inner_or,
PredicateBuilder::Or({a, b}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> inner_and,
+ PredicateBuilder::And({key_pred, a}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> root_and,
+ PredicateBuilder::And({key_pred, inner_or}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> root_or,
+ PredicateBuilder::Or({inner_and, b}));
+ for (const auto& predicate : {root_and, root_or}) {
+ auto filter =
std::dynamic_pointer_cast<PredicateFilter>(predicate);
+ for (int64_t offset = 0; offset <= batch->length(); ++offset) {
+ auto slice = batch->Slice(offset);
+ ASSERT_OK_AND_ASSIGN(std::vector<char> expected,
+ TestEager(predicate, *slice,
arrow::default_memory_pool()));
+ ASSERT_OK_AND_ASSIGN(std::vector<char> actual,
+ filter->Test(*slice,
arrow::default_memory_pool()));
+ ASSERT_EQ(actual, expected);
+ }
+ }
+ }
+}
+
+// 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) {
Review Comment:
This benchmark never runs in CI, prints to std::cout, and pulls `<chrono>`
and `<iostream>` into the unit-test binary. It is also the only caller of
`TestEager`'s `literal_equal` and `box_all_fields` parameters, so the shared
correctness helper carries two extra modes purely to feed it.
The repo already has a `benchmark/` target for this. Suggest moving it
there, or dropping it and keeping the numbers in the PR description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
