[fix](fe) Fold external partition min max aggregate to constant - #67136
[fix](fe) Fold external partition min max aggregate to constant#67136felixwluo wants to merge 2 commits into
Conversation
… segments in the rowset reader (apache#35484) ## Proposed changes pick apache#35432 ## Further comments If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
/review |
|
run buildall |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Requesting changes: the new fold assumes that the selected partition map is an exact, complete, row-bearing view of the relation read by the connector and that a primitive partition type can reproduce the aggregate result type. Those assumptions do not hold across the supported external-table paths, so the rewrite can return wrong constants, return NULL for nonempty data, suppress an expected table-not-found error, or fail planning. Seven blocking correctness/coverage issues and one P2 missed-optimization issue are detailed inline.
Critical checkpoint conclusions:
- Concurrency and thread safety: No threads, locks, or atomics are added. External metadata mutation exposes a read-generation consistency defect rather than a locking defect: MaxCompute can fold a cached partition generation while the scan reads the current source generation.
- Error handling: Blocking. An unresolved connector handle can be reduced to an empty map and then NULL, bypassing the table-not-found error that scan creation would raise. Parameterized decimal reconstruction can also introduce a planning exception.
- Memory safety: No BE/C++ ownership or reservation path is changed; the FE allocations are bounded planner objects and partition-map traversal. No memory-safety issue found.
- Data correctness: Blocking. The fold admits Iceberg transformed partition values, registered partitions with no visible rows, deferred historical maps, Paimon incremental maps broader than the requested window, and stale MaxCompute listings that differ from scan-all. Primitive-only literal reconstruction additionally loses value and schema fidelity.
- Observability: No new metric is inherently required for a safe planner rule, but these failures are silent wrong-result paths; EXPLAIN may show scan removal without exposing the invalid metadata assumption. Correct applicability checks are required rather than logging.
- BE null and nullable handling: BE column handling is not touched. FE aggregate-null semantics were checked: genuine null partition keys are skipped consistently, but empty/deferred maps are incorrectly treated as an empty relation, and both NULL and non-NULL replacements must preserve the normalized aggregate result type.
Supporting review conclusions: The scan-elimination goal is useful, but the implementation is gated by a generic external-table pruning capability that is much broader than an identity, complete, snapshot-consistent extrema contract. Both rewrite registrations share the new RuleType and terminate correctly for supported direct/one-project shapes; however, the operative whole-plan placement precedes project merging and misses an ordinary derived alias. No configuration, persistence/EditLog, FE-BE protocol, storage-format, transaction, or data-write change is involved. The metadata traversal is O(partitions) and intended to save a scan, but correctness blocks rollout and the alias path retains the scan. The authoritative change set contains no FE unit or regression test and no test-result file; repository search found no existing reference to the rule or RuleType. Builds/tests were not run because the review prompt explicitly prohibits them. The user focus file supplied no additional focus points.
Review status: converged after three rounds. The final two normal reviewers and the separate risk challenger all returned NO_NEW_VALUABLE_FINDINGS after rechecking the eight accepted findings. The live PR head/base still match the authoritative bundle, and live PR-scoped review/comment context was empty immediately before submission.
| new NormalizeAggregate(), | ||
| new CountLiteralRewrite(), | ||
| new RewriteSimpleAggToConstantRule(), | ||
| new RewritePartitionColumnMinMaxToConstantRule(), |
There was a problem hiding this comment.
[P2] Revisit this rule after project merging. For SELECT max(q) FROM (SELECT p AS q FROM ext) v, NormalizeAggregate first adds its own bottom project, so the operative whole-plan job here sees Aggregate -> Project -> Project -> FileScan; the factory matches at most one project. MergeProjectable runs only in the following traversal, which never reruns this rule, so the intended alias case still scans files. Move/rerun the fold after project merging in both pipelines, or handle the normalized project chain directly.
| */ | ||
| public class RewritePartitionColumnMinMaxToConstantRule implements RewriteRuleFactory { | ||
|
|
||
| @Override |
There was a problem hiding this comment.
[P1] Add tests before enabling this semantic rewrite. The PR changes only production files, and there is no existing FE or regression reference to this rule or RuleType. Please cover direct and scalar-subquery plans, the normalized alias shape, rule disabling, all applicability gates, null/default/empty partitions, connector transforms and scan selectors, and parameterized output types; assert both result value/type and whether the file scan is removed.
|
|
||
| private Plan tryRewrite(LogicalAggregate<?> agg, LogicalFileScan scan, | ||
| Optional<LogicalProject<LogicalFileScan>> project, StatementContext statementContext) { | ||
| if (scan.getTableSample().isPresent() || !agg.getGroupByExpressions().isEmpty()) { |
There was a problem hiding this comment.
[P1] Skip this fold for incremental scan params. The Paimon @incr path intentionally freezes the whole latest partition map here, then applies incremental-between* only when planning the connector scan. If the requested window contains only p=2024 while latest also has p=2025, this rule rewrites the window's MAX(p) to 2025. Reject row-set-altering scan params, or derive extrema from metadata scoped to the exact incremental relation.
| } | ||
|
|
||
| ExternalTable table = scan.getTable(); | ||
| if (!table.supportInternalPartitionPruned()) { |
There was a problem hiding this comment.
[P1] Require an exact identity-partition contract here. supportInternalPartitionPruned() is true for every plugin-driven table, including Iceberg specs such as truncate(10, id) and bucket(2, id). Iceberg publishes the source name id, but its partition map stores the transformed value. Thus Aggregate(max(id)) -> FileScan(selected={10,20}) for rows 19 and 22 is rewritten to constant 20 instead of 22. Please gate this on connector metadata proving identity values (or restrict it to a connector with that guarantee), rather than generic pruning support.
| return slot.getName().equalsIgnoreCase(partitionColumn.getName()); | ||
| } | ||
|
|
||
| private Optional<Literal> findPartitionMinMaxLiteral(boolean isMin, LogicalFileScan scan, Column partitionColumn) { |
There was a problem hiding this comment.
[P1] Do not derive query results from a partition generation that the scan does not use. MaxCompute serves this map from a cross-query cache with a default 600-second TTL, but initSelectedPartitions marks it unpruned; the scan path consequently passes no required partitions and ODPS scans the current source set. If cached p=1,p=9 loses p=9 remotely, this fold returns 9 while the scan returns 1 (and a remote add misses a new extremum). Require metadata pinned to the scan's exact generation/read set, or decline this optimization for cached scan-all listings.
|
|
||
| private Optional<Literal> findPartitionMinMaxLiteral(boolean isMin, LogicalFileScan scan, Column partitionColumn) { | ||
| PartitionKey selectedKey = null; | ||
| for (PartitionItem item : scan.getSelectedPartitions().selectedPartitions.values()) { |
There was a problem hiding this comment.
[P1] Partition existence is not row-existence evidence. Hive lists every HMS partition name while its ConnectorPartitionInfo row/file counts are explicitly unknown, and an ALTER TABLE ... ADD PARTITION (p=999) may leave an empty directory. With the only row in p=1, SQL MAX(p) is 1 but this loop returns 999; if 999 is the only empty partition, SQL returns NULL while the rewrite still returns 999. Please decline the fold unless the connector supplies a snapshot-consistent guarantee that every considered identity key contributes a visible row.
| } | ||
| } | ||
|
|
||
| if (selectedKey == null) { |
There was a problem hiding this comment.
[P1] Do not treat an unenumerated or unresolved partition universe as an empty relation. Explicit FOR VERSION AS OF, @tag, @branch, and snapshot-selecting options intentionally keep partition columns with an empty map because enumeration is deferred; an absent connector handle also returns {} here while scan creation would throw Table '...' does not exist. Both reach selectedKey == null, either corrupting a nonempty historical result or suppressing the expected error. Require an explicitly complete map for the exact read and preserve unresolved-handle failure.
| if (selectedKey == null) { | ||
| return Optional.of(new NullLiteral(DataType.fromCatalogType(partitionColumn.getType()))); | ||
| } | ||
| Type literalType = Type.fromPrimitiveType(selectedKey.getTypes().get(0)); |
There was a problem hiding this comment.
[P1] Preserve the normalized aggregate return type instead of rebuilding it from PrimitiveType. PartitionKey has already discarded scale/precision/length: DATETIMEV2(6) is rebuilt as scale 0 and rounded, DECIMAL(10,2) is rebuilt with scale 0 and can fail literal validation, CHAR/VARCHAR lose their declared length, and DECIMAL256 has no mapping here. This can change MAX(p)'s value or schema, or fail planning. Construct and validate both non-null and typed-NULL replacements against func.getDataType(); MIN/MAX.customSignature() can normalize DecimalV2 to DecimalV3, so the partition-column type alone is not always the result type.
TPC-H: Total hot run time: 17272 ms |
TPC-DS: Total hot run time: 82832 ms |
ClickBench: Total hot run time: 14.73 s |
FE Regression Coverage ReportIncrement line coverage |
3 participants
What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
For external Hive tables, queries using predicates like
dt = (select max(dt) from table)could not use partition pruning effectively.The scalar subquery
select max(dt)was planned as a normal file scan. In alarge online cloud-mode cluster, this caused Doris to scan all partitions and
billions of rows only to compute the latest partition value. The outer scan also
kept
dt = MAX(dt)as a join condition instead of a literal partition predicate,so it scanned all
dtpartitions before filtering rows.This patch adds a Nereids rewrite rule for external tables. For simple
no-group-by
MIN/MAXaggregate queries on a single external list partitioncolumn, Doris now computes the result from partition metadata and rewrites the
aggregate to a one-row constant relation. This allows predicates such as
dt = (select max(dt) ...)to becomedt = 'literal'before file-scan partitionpruning, so the outer scan can prune to the target partition.
Explain:
before:
now:
Release note
Fix slow queries on external Hive partitioned tables when filtering by
partition_col = (select max(partition_col) from table).None
Check List (For Author)
Test
Behavior changed:
Does this need documentation?
Check List (For Reviewer who merge this PR)