github-actions[bot] commented on code in PR #65500:
URL: https://github.com/apache/doris/pull/65500#discussion_r3566415702
##########
be/test/format_v2/table_reader_test.cpp:
##########
@@ -1215,6 +1244,140 @@ TEST(TableReaderTest,
PrepareSplitPrunesPartitionRuntimeFilter) {
EXPECT_FALSE(reader.current_split_pruned());
}
+TEST(TableReaderTest,
PrepareSplitDoesNotEvaluateNonDeterministicPartitionPredicate) {
+ std::vector<ColumnDefinition> projected_columns;
+ auto partition_column = make_table_column(0, "part",
std::make_shared<DataTypeInt32>());
+ partition_column.is_partition_key = true;
+ projected_columns.push_back(std::move(partition_column));
+ set_name_identifiers(&projected_columns);
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RuntimeProfile profile("scanner");
+ TableReader reader;
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ })
+ .ok());
+
+ bool predicate_executed = false;
+ auto predicate =
std::make_shared<NonDeterministicPartitionPredicate>(&predicate_executed);
+ predicate->add_child(table_int32_slot_ref(0, 0, "part"));
+ SplitReadOptions split;
+ split.current_range.__set_path("unused-nondeterministic-file");
+ split.partition_values.emplace("part", Field::create_field<TYPE_INT>(7));
+ split.partition_prune_conjuncts.push_back(
+
VExprContext::create_shared(runtime_filter_wrapper_expr(std::move(predicate))));
+ split.partition_prune_conjuncts.push_back(VExprContext::create_shared(
+ runtime_filter_wrapper_expr(table_int32_greater_than_expr(0, 0,
10))));
+
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ EXPECT_FALSE(predicate_executed);
+ EXPECT_FALSE(reader.current_split_pruned());
+ ASSERT_NE(profile.get_counter("RuntimeFilterPartitionPrunedRangeNum"),
nullptr);
+
EXPECT_EQ(profile.get_counter("RuntimeFilterPartitionPrunedRangeNum")->value(),
0);
+}
+
+TEST(TableReaderTest, ConstantPruningStopsAtUnsafePredicate) {
+ std::vector<ColumnDefinition> projected_columns;
+ auto partition_column = make_table_column(0, "part",
std::make_shared<DataTypeInt32>());
+ partition_column.is_partition_key = true;
+ projected_columns.push_back(std::move(partition_column));
+ set_name_identifiers(&projected_columns);
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ bool predicate_executed = false;
+ auto unsafe_predicate =
+
std::make_shared<NonDeterministicPartitionPredicate>(&predicate_executed);
+ unsafe_predicate->add_child(table_int32_slot_ref(0, 0, "part"));
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ FakeTableReader reader({}, fake_state);
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts =
+ {
+ prepared_conjunct(&state,
unsafe_predicate),
+ prepared_conjunct(&state,
+
table_int32_greater_than_expr(
+
0, 0, 10)),
+ },
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ })
+ .ok());
+
+ SplitReadOptions split;
+ split.current_range.__set_path("fake-table-reader-input");
+ split.partition_values.emplace("part", Field::create_field<TYPE_INT>(7));
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_FALSE(predicate_executed);
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(fake_state->open_count, 1);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST(TableReaderTest, ConstantPruningStopsAtUnsafeSlotlessPredicate) {
+ std::vector<ColumnDefinition> projected_columns;
+ auto partition_column = make_table_column(0, "part",
std::make_shared<DataTypeInt32>());
+ partition_column.is_partition_key = true;
+ projected_columns.push_back(std::move(partition_column));
+ set_name_identifiers(&projected_columns);
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ bool predicate_executed = false;
+ auto unsafe_slotless_predicate =
+
std::make_shared<NonDeterministicPartitionPredicate>(&predicate_executed);
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ FakeTableReader reader({}, fake_state);
+ ASSERT_TRUE(
+ reader
+ .init({
+ .projected_columns = projected_columns,
+ .conjuncts =
+ {
+ prepared_conjunct(&state,
unsafe_slotless_predicate),
+ prepared_conjunct(&state,
table_int32_greater_than_expr(
+
0, 0, 10)),
+ },
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ })
+ .ok());
+
+ SplitReadOptions split;
+ split.current_range.__set_path("fake-table-reader-input");
+ split.partition_values.emplace("part", Field::create_field<TYPE_INT>(7));
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ // The unsafe expression must reach normal row-level evaluation; it must
not be hidden by the
+ // later false constant predicate pruning the split first.
Review Comment:
Nothing in this test path can set `predicate_executed`. The flag is only set
from `NonDeterministicPartitionPredicate::execute_column_impl()`, but the
slotless predicate is not converted into a `TableFilter`,
`FakeFileReader::get_block()` just fills requested columns and does not execute
request conjuncts, and this TableReader unit never runs
`Scanner::_filter_output_block()` over the original `_conjuncts`;
`finalize_chunk()` only materializes columns. As written, this `EXPECT_TRUE`
should fail and still would not prove row-level evaluation is preserved. Please
limit this test to the split-not-pruned assertion and add a real
scanner/file-reader-level regression for the unsafe slotless predicate.
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -118,17 +121,66 @@ bool set_decoded_field(const ParquetColumnSchema&
column_schema, DecodedValueKin
return read_decoded_field(column_schema, view, field, timezone).ok();
}
+int64_t floor_timestamp_seconds(int64_t value, ParquetTimeUnit time_unit) {
+ int64_t units_per_second = 1;
+ switch (time_unit) {
+ case ParquetTimeUnit::MILLIS:
+ units_per_second = 1000;
+ break;
+ case ParquetTimeUnit::MICROS:
+ units_per_second = 1000000;
+ break;
+ case ParquetTimeUnit::NANOS:
+ units_per_second = 1000000000;
+ break;
+ default:
+ DORIS_CHECK(false);
+ }
+ return format::floor_epoch_seconds(value, units_per_second);
+}
+
+bool timestamp_min_max_is_safe(const ParquetColumnSchema& column_schema,
int64_t min_value,
+ int64_t max_value, const cctz::time_zone*
timezone) {
+ if (!column_schema.type_descriptor.is_timestamp ||
+ !column_schema.type_descriptor.timestamp_is_adjusted_to_utc ||
timezone == nullptr ||
+ remove_nullable(column_schema.type)->get_primitive_type() ==
TYPE_TIMESTAMPTZ) {
+ // TIMESTAMPTZ keeps the original UTC ordering, so local civil-time
rollback does not make
+ // its converted min/max non-monotonic.
+ return true;
+ }
+ return format::utc_timestamp_range_is_monotonic(
+ floor_timestamp_seconds(min_value,
column_schema.type_descriptor.time_unit),
+ floor_timestamp_seconds(max_value,
column_schema.type_descriptor.time_unit), *timezone);
+}
Review Comment:
This validation still lets inverted Parquet stats become trusted ZoneMaps.
`valid_min_max()` rejects NaN floating bounds, but every non-floating native
type returns true even when min > max; BYTE_ARRAY and FIXED_LEN_BYTE_ARRAY
pruning bypass this helper through the string min/max paths and also mark
decoded inverted bounds usable. With corrupt stats such as min=10/max=1, a
predicate like `col = 5` can return `kNoMatch` and drop a row that is actually
in the group/page; native MIN/MAX aggregate pushdown also consumes the same
`TransformColumnStatistics()` result and can return those corrupt bounds.
Please validate every decoded min/max pair used for pruning, leaving
`has_min_max=false` when min > max, and fall back from aggregate pushdown for
inverted native stats.
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -523,9 +523,23 @@ Status TableReader::init(TableReadOptions&& options) {
Status TableReader::_build_table_filters_from_conjuncts() {
_table_filters.clear();
+ _constant_pruning_safe_filter_count = 0;
+ bool in_safe_prefix = true;
for (const auto& conjunct : _conjuncts) {
+ DORIS_CHECK(conjunct != nullptr);
+ DORIS_CHECK(conjunct->root() != nullptr);
+ // `_table_filters` omits expressions without slot references, but
such an expression still
+ // occupies a position in the row-level conjunct order. Record how
many localized filters
+ // precede the first unsafe original conjunct so constant pruning
cannot jump over a
+ // slotless non-deterministic/error-preserving barrier.
+ if (in_safe_prefix && !_is_safe_to_pre_execute(conjunct)) {
+ in_safe_prefix = false;
+ }
RETURN_IF_ERROR(
build_table_filters_from_conjunct(conjunct, _runtime_state,
&_table_filters));
+ if (in_safe_prefix) {
Review Comment:
The safe-prefix count only protects `_evaluate_constant_filters()`, but the
full `_table_filters` vector still goes into `create_scan_request()` before
that. A slotless unsafe conjunct is omitted by
`build_table_filters_from_conjunct()` because it has no `GlobalIndex`, so an
order like `[assert_true(false), file_col = 0]` still localizes `file_col = 0`
into `FileScanRequest::conjuncts`. Parquet/ORC execute those file-local
conjuncts before `Scanner::_filter_output_block()`, and `filter_block()`
returns immediately on empty blocks, so the unsafe slotless predicate can be
skipped after the later file predicate filters the batch. Please carry the same
original-order barrier into file-local filter localization, or add
non-executable barrier entries, and cover a slotless unsafe predicate before a
false file-local predicate.
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -118,17 +121,66 @@ bool set_decoded_field(const ParquetColumnSchema&
column_schema, DecodedValueKin
return read_decoded_field(column_schema, view, field, timezone).ok();
}
+int64_t floor_timestamp_seconds(int64_t value, ParquetTimeUnit time_unit) {
+ int64_t units_per_second = 1;
+ switch (time_unit) {
+ case ParquetTimeUnit::MILLIS:
+ units_per_second = 1000;
+ break;
+ case ParquetTimeUnit::MICROS:
+ units_per_second = 1000000;
+ break;
+ case ParquetTimeUnit::NANOS:
+ units_per_second = 1000000000;
+ break;
+ default:
+ DORIS_CHECK(false);
+ }
+ return format::floor_epoch_seconds(value, units_per_second);
Review Comment:
This still accepts some corrupt timestamp ranges before the conservative
fallback can trigger. For Parquet, the raw INT64 endpoints are reduced to whole
seconds before `utc_timestamp_range_is_monotonic()` sees them, so
min=1.500000s/max=1.000000s floors to the same second and then becomes a
trusted inverted ZoneMap; `TYPE_TIMESTAMPTZ` returns true even earlier without
any ordering check. ORC has the same pattern in `set_timestamp_zone_map()`: the
TIMESTAMPTZ branch skips the helper, and the DATETIMEV2 branch checks only
floored millis seconds before storing the full millis+nanos values. Please
validate the raw timestamp endpoints first (`min_value <= max_value` at the
Parquet unit, and the full `(millis, nanos)` pair for ORC) so corrupt stats
fall back instead of pruning rows or feeding MIN/MAX aggregate pushdown.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -671,6 +687,7 @@ Status ParquetReader::get_aggregate_result(const
format::FileAggregateRequest& r
RETURN_IF_ERROR(find_projected_minmax_leaf(
*column_schema,
request.columns[request_column_idx].projection, &leaf_schema));
DORIS_CHECK(leaf_schema != nullptr);
Review Comment:
This aggregate path still does not account for non-trivial table/file
mappings. `TableReader::_supports_aggregate_pushdown()` can allow a leaf
mapping whose `projection` casts the file column to the table type, so Parquet
returns file-order min/max first and `_materialize_aggregate_pushdown_rows()`
casts only those two endpoint values afterward. That is not equivalent for
non-order-preserving casts: for a file INT column with values `{2, 9, 10}`
mapped to a table STRING column, file stats return endpoints `{2, 10}`, which
materialize as `{"2", "10"}`, but table-order `MAX(CAST(value AS STRING))`
should see all rows and return `"9"`. Please disable MIN/MAX pushdown unless
the mapping is trivial, or explicitly allow only casts proven to preserve the
table aggregate ordering.
--
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]
