github-actions[bot] commented on code in PR #65369:
URL: https://github.com/apache/doris/pull/65369#discussion_r3549581323
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -597,42 +780,500 @@ Status
ParquetScanScheduler::skip_current_row_group_rows(int64_t rows) {
return Status::OK();
}
+namespace {
+
+struct PredicateConjunctSchedule {
+ std::map<size_t, VExprContextSPtrs> single_column_conjuncts;
+ VExprContextSPtrs remaining_conjuncts;
+};
+
+PredicateConjunctSchedule build_predicate_conjunct_schedule(
+ const format::FileScanRequest& request) {
+ std::unordered_set<size_t> predicate_block_positions;
+ predicate_block_positions.reserve(request.predicate_columns.size());
+ for (const auto& col : request.predicate_columns) {
+ const auto position_it = request.local_positions.find(col.column_id());
+ DORIS_CHECK(position_it != request.local_positions.end());
+ predicate_block_positions.insert(position_it->second.value());
+ }
+
+ PredicateConjunctSchedule schedule;
+ for (const auto& conjunct : request.conjuncts) {
+ DORIS_CHECK(conjunct != nullptr);
+ DORIS_CHECK(conjunct->root() != nullptr);
+ if (!conjunct->root()->is_deterministic()) {
+ // Round-by-round filtering can compact later predicate columns
before evaluating
+ // remaining expressions. Stateful functions such as random(1)
must see the same full
+ // batch they saw before this optimization, so any
non-deterministic conjunct disables
+ // the per-column schedule for the whole batch.
+ schedule.remaining_conjuncts = request.conjuncts;
+ schedule.single_column_conjuncts.clear();
+ return schedule;
+ }
+ std::set<int> referenced_positions;
+ conjunct->root()->collect_slot_column_ids(referenced_positions);
+ if (referenced_positions.size() != 1) {
+ schedule.remaining_conjuncts.push_back(conjunct);
+ continue;
+ }
+ const auto block_position =
static_cast<size_t>(*referenced_positions.begin());
+ if (!predicate_block_positions.contains(block_position)) {
+ schedule.remaining_conjuncts.push_back(conjunct);
+ continue;
+ }
+ schedule.single_column_conjuncts[block_position].push_back(conjunct);
+ }
+ return schedule;
+}
+
+bool can_evaluate_all_with_dictionary(const VExprContextSPtrs& conjuncts) {
+ if (conjuncts.empty()) {
+ return false;
+ }
+ return std::ranges::all_of(conjuncts, [](const auto& conjunct) {
+ return conjunct != nullptr && conjunct->root() != nullptr &&
+ conjunct->root()->can_evaluate_dictionary_filter();
+ });
+}
+
+void collect_dictionary_residual_exprs(const VExprContextSPtr& owner_context,
const VExprSPtr& expr,
+ DictionaryResidualConjuncts*
residual_conjuncts) {
+ DORIS_CHECK(owner_context != nullptr);
+ DORIS_CHECK(expr != nullptr);
+ DORIS_CHECK(residual_conjuncts != nullptr);
+
+ // The dictionary interface is exact for a single dictionary-aware
predicate and for OR only
+ // when every child is dictionary-aware. AND is different: VCompoundPred
evaluates only the
+ // dictionary-aware children as a prefilter. Split AND recursively so the
already-applied
+ // dictionary children are not executed again on materialized rows, while
non-dictionary
+ // residual children still keep the original row-level semantics.
+ const auto* compound_pred = dynamic_cast<const VCompoundPred*>(expr.get());
+ if (compound_pred != nullptr && compound_pred->op() ==
TExprOpcode::COMPOUND_AND) {
+ for (const auto& child : expr->children()) {
+ collect_dictionary_residual_exprs(owner_context, child,
residual_conjuncts);
+ }
+ return;
+ }
+
+ if (!expr->can_evaluate_dictionary_filter()) {
Review Comment:
This still drops residual work for nested compound predicates.
`collect_dictionary_residual_exprs()` only descends through `AND`; for any
other node it assumes `can_evaluate_dictionary_filter()` means the whole
expression is dictionary-equivalent. But `VCompoundPred` makes `OR` return true
when each child can dictionary-filter, while an `AND` child returns true when
only one child can dictionary-filter. A predicate like `(value = 'az') OR
(value = 'za' AND length(value) = 1)` therefore has a dictionary-evaluable
root: the dictionary bitmap keeps both `az` and `za`, and no residual is
recorded for `length(value) = 1`. When `read_round_by_round()` sees
`used_dictionary_filter`, it runs that empty residual list, so `za` rows
survive even though the original row predicate is false. Please treat OR as
dictionary-exact only when every descendant is fully dictionary-equivalent, or
keep the whole compound expression as a row-level residual when any branch is
only a prefilter.
##########
be/test/format_v2/parquet/parquet_reader_test.cpp:
##########
@@ -1637,6 +1801,183 @@ TEST_F(NewParquetReaderTest,
PredicateFiltersRowGroupsByDictionary) {
EXPECT_EQ(values, std::vector<std::string>({"lm"}));
}
+TEST_F(NewParquetReaderTest, DictionaryPredicateFiltersRowsInsideRowGroup) {
+ write_single_row_group_dictionary_filter_parquet_file(_file_path);
+ auto parquet_file_reader =
::parquet::ParquetFileReader::OpenFile(_file_path, false);
+ ASSERT_EQ(parquet_file_reader->metadata()->num_row_groups(), 1);
+ auto row_group = parquet_file_reader->metadata()->RowGroup(0);
+ ASSERT_NE(row_group, nullptr);
+ ASSERT_TRUE(row_group->ColumnChunk(1)->has_dictionary_page());
+
+ RuntimeProfile profile("new_parquet_reader_dictionary_filter_profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(1)};
+ request->non_predicate_columns = {field_projection(0)};
+ request->conjuncts.push_back(create_string_in_conjunct(1, {"az", "za"}));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> values;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = nullable_nested_column<ColumnInt32>(block, 0);
+ const auto& value_column = nullable_nested_column<ColumnString>(block,
1);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ values.push_back(value_column.get_data_at(row).to_string());
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({2, 5}));
+ EXPECT_EQ(values, std::vector<std::string>({"az", "za"}));
+ EXPECT_EQ(profile.get_counter("RowsFilteredByConjunct")->value(), 4);
+ EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 4);
+ EXPECT_EQ(profile.get_counter("DictFilterCandidateColumns")->value(), 1);
+ EXPECT_EQ(profile.get_counter("DictFilterColumns")->value(), 1);
+ EXPECT_EQ(profile.get_counter("DictFilterUnsupportedColumns")->value(), 0);
+ EXPECT_EQ(profile.get_counter("DictFilterReadFailures")->value(), 0);
+ ASSERT_NE(profile.get_counter("DictFilterExprRewriteTime"), nullptr);
+ ASSERT_NE(profile.get_counter("DictFilterReadDictTime"), nullptr);
+ ASSERT_NE(profile.get_counter("DictFilterBuildTime"), nullptr);
+ EXPECT_EQ(profile.get_counter("SelectedRows")->value(), 2);
+ EXPECT_GE(profile.get_counter("ReaderSelectRows")->value(), 8);
+}
+
+TEST_F(NewParquetReaderTest, DictionaryPredicateWorksWithoutRuntimeProfile) {
+ write_single_row_group_dictionary_filter_parquet_file(_file_path);
+
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(1)};
+ request->non_predicate_columns = {field_projection(0)};
+ request->conjuncts.push_back(create_string_in_conjunct(1, {"az", "za"}));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> values;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = nullable_nested_column<ColumnInt32>(block, 0);
+ const auto& value_column = nullable_nested_column<ColumnString>(block,
1);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ values.push_back(value_column.get_data_at(row).to_string());
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({2, 5}));
+ EXPECT_EQ(values, std::vector<std::string>({"az", "za"}));
+}
+
+TEST_F(NewParquetReaderTest,
DictionaryPredicateSkipsRemainingPredicateColumnsWhenEmpty) {
+ write_single_row_group_dictionary_filter_parquet_file(_file_path);
+
+ RuntimeProfile
profile("new_parquet_reader_dictionary_filter_empty_profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(1), field_projection(0)};
+ request->conjuncts.push_back(create_string_in_conjunct(1,
{"not_present"}));
+ request->conjuncts.push_back(create_int32_greater_than_conjunct(0, 0));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ bool eof = false;
+ size_t total_rows = 0;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ total_rows += rows;
+ }
+
+ EXPECT_EQ(total_rows, 0);
+ EXPECT_EQ(profile.get_counter("RowsFilteredByConjunct")->value(), 6);
+ EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 6);
+ EXPECT_EQ(profile.get_counter("DictFilterCandidateColumns")->value(), 1);
+ EXPECT_EQ(profile.get_counter("DictFilterColumns")->value(), 1);
+ EXPECT_EQ(profile.get_counter("DictFilterUnsupportedColumns")->value(), 0);
+ EXPECT_EQ(profile.get_counter("DictFilterReadFailures")->value(), 0);
+ EXPECT_EQ(profile.get_counter("SelectedRows")->value(), 0);
+ // The first dictionary predicate column is read once to produce a compact
row filter. The
+ // second predicate column is skipped after the selection becomes empty,
which verifies the
+ // StarRocks-style round-by-round policy: only rows surviving previous
predicates are read.
+ EXPECT_EQ(profile.get_counter("ReaderSelectRows")->value(), 6);
+ EXPECT_EQ(profile.get_counter("ReaderSkipRows")->value(), 6);
+}
+
+TEST_F(NewParquetReaderTest,
DictionaryPredicateRunsResidualConjunctOnSurvivors) {
+ write_single_row_group_dictionary_filter_parquet_file(_file_path);
+
+ RuntimeProfile
profile("new_parquet_reader_dictionary_prefilter_residual_profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(1)};
+ request->non_predicate_columns = {field_projection(0)};
+ request->conjuncts.push_back(
+ create_string_dictionary_and_residual_conjunct(1, {"az", "za"},
"za"));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> values;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = nullable_nested_column<ColumnInt32>(block, 0);
+ const auto& value_column = nullable_nested_column<ColumnString>(block,
1);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ values.push_back(value_column.get_data_at(row).to_string());
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({5}));
+ EXPECT_EQ(values, std::vector<std::string>({"za"}));
+ EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 4);
+ EXPECT_EQ(profile.get_counter("RowsFilteredByConjunct")->value(), 1);
Review Comment:
This expectation does not match the scanner counter path. The dictionary
filter already removes 4 of the 6 rows, and `read_filter_columns()` adds those
4 to `conjunct_filtered_rows`; then the residual `"za"` check removes one of
the two survivors and adds one more. `read_current_row_group_batch()` publishes
that accumulator as `RowsFilteredByConjunct`, so this test should see 5, not 1.
Either update this expectation to match the current counter semantics, or stop
adding dictionary-prefiltered rows to `conjunct_filtered_rows` and adjust the
other dictionary profile assertions consistently.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -597,42 +780,500 @@ Status
ParquetScanScheduler::skip_current_row_group_rows(int64_t rows) {
return Status::OK();
}
+namespace {
+
+struct PredicateConjunctSchedule {
+ std::map<size_t, VExprContextSPtrs> single_column_conjuncts;
+ VExprContextSPtrs remaining_conjuncts;
+};
+
+PredicateConjunctSchedule build_predicate_conjunct_schedule(
+ const format::FileScanRequest& request) {
+ std::unordered_set<size_t> predicate_block_positions;
+ predicate_block_positions.reserve(request.predicate_columns.size());
+ for (const auto& col : request.predicate_columns) {
+ const auto position_it = request.local_positions.find(col.column_id());
+ DORIS_CHECK(position_it != request.local_positions.end());
+ predicate_block_positions.insert(position_it->second.value());
+ }
+
+ PredicateConjunctSchedule schedule;
+ for (const auto& conjunct : request.conjuncts) {
+ DORIS_CHECK(conjunct != nullptr);
+ DORIS_CHECK(conjunct->root() != nullptr);
+ if (!conjunct->root()->is_deterministic()) {
Review Comment:
This guard still lets deterministic error-preserving predicates move onto
the compacted survivor path. For example, `assert_true` is deterministic, but
FE marks it as `NoneMovableFunction`, and the BE implementation throws when any
evaluated row is false or null. Before this PR, every conjunct was executed
with `rows=batch_rows`, so `a > 0 AND assert_true(b > 0, 'bad')` still checked
`b` for rows rejected by `a > 0`. Now the later single-column conjunct can be
scheduled here, `read_round_by_round()` reads/evaluates it only after
`selected_rows` was shrunk by the earlier predicate, and those rejected rows no
longer throw. Please keep error-preserving/non-movable functions on the old
full-batch path, or add an execution-safety property that is stronger than
determinism before scheduling a conjunct round by round.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
