github-actions[bot] commented on code in PR #68452:
URL: https://github.com/apache/doris/pull/68452#discussion_r4084171149
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java:
##########
@@ -2358,6 +2365,23 @@ private boolean supportsHiveSampleAnalyze(HmsTableInfo
tableInfo) {
return !isView(tableInfo) && HiveTableFormatDetector.detect(tableInfo)
== HiveTableType.HIVE;
}
+ /**
+ * Whether this table's partition column values can be reconstructed from
the data file path, i.e.
+ * whether a partition-column-only aggregation may be answered without
opening any data file.
+ *
+ * <p>HIVE and hudi-on-HMS both qualify: HMS stores their partition values
as the directory path
+ * (`dt=2026-08-11/`), and BE carries them per scan range as
`columns_from_path`. ICEBERG is excluded
+ * (hidden partitioning / partition transforms / v2 delete files) and so
is UNKNOWN. This is the
+ * modern replacement for the legacy {@code HMSExternalTable.DLAType}
whitelist {@code HIVE || HUDI}.
+ */
+ private boolean supportsPartitionValueOnly(HmsTableInfo tableInfo) {
+ if (isView(tableInfo)) {
+ return false;
+ }
+ HiveTableType tableType = HiveTableFormatDetector.detect(tableInfo);
+ return tableType == HiveTableType.HIVE || tableType ==
HiveTableType.HUDI;
Review Comment:
[P1] Exclude transactional Hive from this capability
This test admits full-ACID Hive tables because they still detect as `HIVE`.
Their ranges carry base files plus delete-delta descriptors, and the normal
`TransactionalHiveReader` applies those deletes; the new
`PartitionColumnReader` is installed before table-format dispatch and never
reads them. If all rows in the highest partition were deleted,
`max(partition_col)` now returns that deleted partition instead of the previous
value/NULL. Please withhold this capability for transactional tables unless the
fast path can prove a surviving row after delete deltas.
##########
be/src/exec/scan/file_scanner.cpp:
##########
@@ -1052,6 +1053,38 @@ Status FileScanner::_get_next_reader() {
}
}
+ // partition_column_value_only optimization:
+ // when the pushed-down aggregation only depends on partition columns,
we do not open any
+ // data file. Each scan range simply emits one row carrying its
partition column values,
+ // which PartitionColumnReader fills from `_partition_col_descs`
(itself derived from the
+ // range's `columns_from_path`). This is placed after
_generate_partition_columns() and
+ // runtime filter partition pruning, so pruned ranges are already
skipped above.
+ //
+ // Guard: only take this fast path when EVERY requested column is a
partition column whose
+ // value this range actually carries. Otherwise fall back to the
normal per-format reader, so
+ // an unexpected non-partition column (or a partition value missing
from the path) can only
+ // cost performance, never produce a wrong result.
+ if (_get_push_down_agg_type() == TPushAggOp::type::PARTITION_VALUE &&
Review Comment:
[P1] Route the default FileScannerV2 path through this optimization
This implementation is only reachable in legacy `FileScanner`, but
`enable_file_scanner_v2` defaults to true and ordinary nontransactional Hive
Parquet/ORC/text scans are routed to V2. V2 passes `PARTITION_VALUE` to
`TableReader`, whose shortcut accepts only COUNT/MINMAX, so it opens and reads
the file normally; meanwhile the connector has disabled file splitting,
potentially making the default path slower. Please implement the synthetic
partition result in V2 or explicitly route PARTITION_VALUE scans to V1, and
cover the default session setting.
##########
be/src/format/partition_column_reader.h:
##########
@@ -0,0 +1,133 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "core/block/block.h"
+#include "format/column_descriptor.h"
+#include "format/generic_reader.h"
+#include "format/table/partition_column_filler.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+
+// PartitionColumnReader is used for the "partition_column_value_only"
optimization.
+//
+// When an aggregation (min/max) only depends on the partition columns of an
external
+// (Hive/Hudi) table, the value can be derived purely from partition metadata.
Each scan range
+// corresponds to one data file living under a partition directory, and its
partition column
+// values are already carried by `columns_from_path` in the TFileRangeDesc --
FileScanner turns
+// them into `_partition_col_descs` in `_generate_partition_columns()`.
+//
+// This reader therefore does NOT open or read any data file. It emits exactly
ONE row per scan
+// range, filling every requested column with its own partition value. This
preserves the exact
+// "a partition without any data file produces no row" semantics, because a
partition with no
+// file never generates a scan range in the first place, while a partition
that owns an (even
+// empty) file still gets one scan range and thus contributes its partition
value.
+//
+// NOTE: unlike the legacy FileScanner::_fill_columns_from_path(),
partition/missing/synthesized
+// columns are filled by the READER in the current architecture (see
+// TableFormatReader::on_after_read_block). So this reader must fill the
partition values itself;
+// it cannot just report a row count.
+class PartitionColumnReader : public GenericReader {
+public:
+ // All four arguments are owned by FileScanner and outlive this reader (it
is created per scan
+ // range inside FileScanner::_get_next_reader()). `_partition_col_descs`
and
+ // `_partition_value_is_null` are re-filled per range by
_generate_partition_columns(), so they
+ // are held by pointer and read lazily in _do_get_next_block().
+ PartitionColumnReader(
+ const std::vector<ColumnDescriptor>* column_descs,
+ const std::unordered_map<std::string, std::tuple<std::string,
const SlotDescriptor*>>*
+ partition_col_descs,
+ const std::unordered_map<std::string, bool>*
partition_value_is_null,
+ const std::unordered_map<std::string, uint32_t>*
col_name_to_block_idx)
+ : _column_descs(column_descs),
+ _partition_col_descs(partition_col_descs),
+ _partition_value_is_null(partition_value_is_null),
+ _col_name_to_block_idx(col_name_to_block_idx) {}
+
+ ~PartitionColumnReader() override = default;
+
+protected:
+ // No file is opened: everything this reader needs is already in the scan
range.
+ Status _open_file_reader(ReaderInitContext* /*ctx*/) override { return
Status::OK(); }
+
+ Status _do_init_reader(ReaderInitContext* /*ctx*/) override {
+ _emitted = false;
+ return Status::OK();
+ }
+
+ // Emit exactly one row (per scan range / data file) whose every column
carries that range's
+ // partition column value. FileScanner only installs this reader when ALL
requested columns are
+ // partition columns, so filling them is all that is needed to produce a
complete row.
+ Status _do_get_next_block(Block* block, size_t* read_rows, bool* eof)
override {
+ if (_emitted) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+ _emitted = true;
+
+ for (const ColumnDescriptor& col_desc : *_column_descs) {
+ auto value_it = _partition_col_descs->find(col_desc.name);
+ // FileScanner guards against this before installing the reader;
stay defensive so an
+ // unexpected shape surfaces as a clear error instead of a
silently short column.
+ if (value_it == _partition_col_descs->end()) {
+ return Status::InternalError("Partition column {} has no value
from path",
+ col_desc.name);
+ }
+ auto idx_it = _col_name_to_block_idx->find(col_desc.name);
+ if (idx_it == _col_name_to_block_idx->end()) {
+ return Status::InternalError("Partition column {} not found in
block",
+ col_desc.name);
+ }
+ bool explicit_null_marker = false;
+ auto null_it = _partition_value_is_null->find(col_desc.name);
+ if (null_it != _partition_value_is_null->end()) {
+ explicit_null_marker = null_it->second;
+ }
+ const auto& [value, slot_desc] = value_it->second;
+ auto column_guard = block->mutate_column_scoped(idx_it->second);
+ auto& col_ptr = column_guard.mutable_column();
+ RETURN_IF_ERROR(fill_partition_column_from_path_value(*col_ptr,
*slot_desc, value,
+ 1,
explicit_null_marker));
+ }
+
+ *read_rows = 1;
Review Comment:
[P1] Do not infer a source row from range existence
A listed range does not guarantee that the original scan contributes a row.
`splitFile` skips only byte-size-zero files, but a valid Parquet/ORC file can
contain zero rows while still having a nonzero header/footer; the normal reader
returns EOF for it. This reader instead opens nothing and always returns one
row, so a partition containing only such a file appears in `MAX`, `GROUP BY`,
and `DISTINCT` when it was absent before. It also bypasses the normal
missing/corrupt-file contract after listing. Please require a
snapshot-consistent proof of at least one visible row or fall back to the
format reader, and cover nonzero-length empty files.
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorCapability.java:
##########
@@ -167,6 +167,24 @@ public enum ConnectorCapability {
* whose scan path supports storage-level predicate pruning.</p>
*/
SUPPORTS_STORAGE_PREDICATE_PRUNING,
+ /**
+ * Indicates the connector derives a table's partition column values from
the DATA FILE PATH
+ * ({@code columns_from_path}) rather than from data-file or manifest
contents. The planner may then
+ * answer an aggregation that only depends on partition columns without
opening any data file: each
+ * scan range emits exactly one row carrying its own partition column
values.
+ *
+ * <p>This is the modern equivalent of the legacy {@code
HMSExternalTable.DLAType} whitelist
+ * {@code HIVE || HUDI}. Both keep their partition values in the directory
path, so a path-derived
+ * value is exact. Iceberg and Paimon MUST NOT declare it: Iceberg
supports hidden partitioning and
+ * partition transforms (bucket, truncate, days, ...) that cannot be
reconstructed from the path, and
+ * its v2 position/equality deletes break the "one row per scan range"
assumption; Paimon resolves
+ * partitions from its own manifest metadata.</p>
+ *
+ * <p><b>Scope: catalog-wide OR per-table.</b> hive declares it per-table,
because a single HMS catalog
+ * serves HIVE, HUDI, ICEBERG and PAIMON tables side by side through
sibling connectors, so a
+ * catalog-wide flag would wrongly admit the delegated
(iceberg/paimon-on-HMS) ones.</p>
+ */
+ SUPPORTS_PARTITION_VALUE_ONLY,
Review Comment:
[P1] Version and record the connector SPI change
`ConnectorPluginSurfaceTest` freezes every `ConnectorCapability` constant,
so this addition necessarily differs from `connector-plugin-surface.txt`.
Repository policy requires any shared SPI surface change (including the new
request methods) to refresh the baseline and increment
`connector.plugin.api.version`, but both the POM and the test still pin 10.0.
As written the SPI surface test fails, and a new plugin can advertise API 10
while linking fields/methods an older API-10 FE lacks. Please update the
baseline and bump/pin the next major.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java:
##########
@@ -560,6 +602,63 @@ private LogicalAggregate<? extends Plan>
storageLayerAggregate(
}
}
List<Expression> groupByExpressions =
aggregate.getGroupByExpressions();
+ // Partition-value DISTINCT / GROUP BY pushdown -- extends the MIN/MAX
+ // partition_column_value_only optimization to a pure distinct on
partition columns, e.g.
+ // SELECT DISTINCT dt FROM hive_tbl
+ // SELECT dt FROM hive_tbl GROUP BY dt
+ // -- including when wrapped by a window to pick the latest
partitions (the online pattern):
+ // SELECT dt FROM (SELECT dt, ROW_NUMBER() OVER (ORDER BY dt DESC) rn
+ // FROM hive_tbl GROUP BY dt) t WHERE rn <= 2
+ // The scanner emits one row of partition values per data file (no
file IO, PARTITION_VALUE)
+ // and the GROUP BY above dedups them. Safe only when there is NO
aggregate function (a COUNT
+ // would count files, a MAX/SUM over a data column would need the
data) and every group-by key
+ // references partition columns only, so the scan output is
partition-columns-only and the BE
+ // fast path (_file_slot_descs empty) fires. This is checked before
the generic group-by bail
+ // below.
+ if (!groupByExpressions.isEmpty()
+ && aggregate.getAggregateFunctions().isEmpty()
+ && aggregate.getDistinctArguments().isEmpty()
+ && logicalScan instanceof LogicalFileScan
+ && enablePartitionColumnValueOnly()) {
+ List<? extends Expression> groupByAfterProject = project == null
+ ? groupByExpressions
+ : Project.findProject(groupByExpressions,
project.getProjects());
+ Set<SlotReference> groupBySlots =
+ ExpressionUtils.collect(groupByAfterProject,
SlotReference.class::isInstance);
+ List<SlotReference> groupBySlotsInTable = (List<SlotReference>)
Project.findProject(
+ groupBySlots, logicalScan.getOutput());
+ if (!groupBySlotsInTable.isEmpty()
+ && isAllPartitionColumns(groupBySlotsInTable,
(LogicalFileScan) logicalScan)) {
+ PhysicalFileScan physicalScan = toPhysicalFileScan(
+ (LogicalFileScan) logicalScan, cascadesContext);
+ PhysicalStorageLayerAggregate storageLayerAgg = new
PhysicalStorageLayerAggregate(
+ physicalScan, PushDownAggOp.PARTITION_VALUE);
+ if (project != null) {
+ return aggregate.withChildren(ImmutableList.of(
+
project.withChildren(ImmutableList.of(storageLayerAgg))));
+ } else {
+ return
aggregate.withChildren(ImmutableList.of(storageLayerAgg));
+ }
+ }
+ }
+ // Pure MIN/MAX over partition columns only, without any filter.
Checked here -- BEFORE the
+ // per-column checks further below -- because those checks walk the
arguments of the aggregate
+ // functions, which ConstantPropagation may already have folded into
literals (see
+ // canUsePartitionValueOnly). Keying off the scan output columns
instead keeps the
+ // optimization alive for `select max(dt) from tbl` as well as for the
folded variants.
+ if (logicalScan instanceof LogicalFileScan
+ && canUsePartitionValueOnly(aggregate, (LogicalFileScan)
logicalScan)) {
Review Comment:
[P1] Reject volatile expressions before reducing scan cardinality
This early path checks only the aggregate class and the scan's operative
slots. For example, `Aggregate(max(x)) -> Project(p + random() AS x) ->
FileScan(p)` is rewritten to use `PARTITION_VALUE`, so `random()` is evaluated
once per file instead of once per input row. The pure-GROUP-BY and
filter-crossing entries have the same issue, and `NoneMovableFunction`
expressions can change error behavior too. Please centralize a
`containsVolatileOrNoneMovableExpression()` guard over the aggregate/group
keys, retained project, and retained filter before any PARTITION_VALUE rewrite.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java:
##########
@@ -194,7 +194,7 @@ private List<ConnectorScanRange>
doPlanScan(ConnectorSession session, ConnectorS
HiveFileFormat fileFormat = HiveFileFormat.detect(
hiveHandle.getInputFormat(), hiveHandle.getSerializationLib(),
readHiveJsonInOneColumn(session),
hiveHandle.isFirstColumnString());
- long targetSplitSize = getTargetSplitSize(session);
+ long targetSplitSize = getTargetSplitSize(session, request);
Review Comment:
[P1] Include partition-value mode in the scan reuse key
This value now changes the returned split list, but `planScan` still caches
solely by `HiveScanReuseKey(hiveHandle)`. With two aliases over the same
table/partitions, whichever plans first supplies its whole-file or normally
split ranges to the other. Besides losing parallelism or creating duplicate
synthetic rows, a TABLESAMPLE alias deliberately requests normal splitting and
then samples the reused list, so its sampling granularity can be changed by an
unrelated PARTITION_VALUE alias. Please include this mode/effective target size
in the key or bypass reuse across differing requests.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPruner.java:
##########
@@ -122,11 +132,62 @@ public PhysicalIntersect
visitPhysicalIntersect(PhysicalIntersect intersect, Cas
public PhysicalCTEAnchor<? extends Plan, ? extends Plan>
visitPhysicalCTEAnchor(
PhysicalCTEAnchor<? extends Plan, ? extends Plan> cteAnchor,
CascadesContext context) {
+ // Visit the producer subtree first: if its root is an effective RF
source
+ // (bounded/selective output, e.g. TopN/Limit/visible-column
filter/global agg),
+ // record it so that consumers of this CTE can inherit the
effectiveness.
+ // Without this, a join whose build side is a CTE consumer of such a
producer gets
+ // its runtime filters pruned as "ineffective" merely because the
consumer's
+ // statistics are unknown (always the case for external tables).
cteAnchor.child(0).accept(this, context);
+ RuntimeFilterContext rfCtx = context.getRuntimeFilterContext();
+ if (rfCtx.isEffectiveSrcNode(cteAnchor.child(0))) {
+ effectiveCteProducers.put(cteAnchor.getCteId(),
rfCtx.getEffectiveSrcType(cteAnchor.child(0)));
+ }
cteAnchor.child(1).accept(this, context);
return cteAnchor;
}
+ @Override
+ public PhysicalCTEConsumer visitPhysicalCTEConsumer(PhysicalCTEConsumer
consumer, CascadesContext context) {
+ RuntimeFilterContext rfCtx = context.getRuntimeFilterContext();
+ // Inherit effectiveness recorded from the producer subtree (see
visitPhysicalCTEAnchor).
+ RuntimeFilterContext.EffectiveSrcType producerType =
effectiveCteProducers.get(consumer.getCteId());
+ if (producerType != null) {
+ rfCtx.addEffectiveSrcNode(consumer, producerType);
+ }
+ // A consumer is also a relation that can be the target of RFs.
+ List<Slot> slots = rfCtx.getTargetListByScan(consumer);
+ for (Slot slot : slots) {
+ if
(!rfCtx.getTargetExprIdToFilter().get(slot.getExprId()).isEmpty()) {
+ rfCtx.addEffectiveSrcNode(consumer,
RuntimeFilterContext.EffectiveSrcType.REF);
+ break;
+ }
+ }
+ return consumer;
+ }
+
+ @Override
+ public PhysicalPartitionTopN<? extends Plan> visitPhysicalPartitionTopN(
+ PhysicalPartitionTopN<? extends Plan> partitionTopN,
CascadesContext context) {
+ partitionTopN.child().accept(this, context);
+ // Only mark NATIVE when the output is really bounded.
`partitionLimit` is a PER-PARTITION
+ // limit, so the total row count is NDV(partition keys) *
partitionLimit unless the operator
+ // also carries a global limit or has no partition key at all (see
+ // StatsCalculator#computePartitionTopN). Marking a high-NDV partition
key as "maximally
+ // selective" would keep RFs that filter nothing: the build side stays
huge (RF construction
+ // and memory cost) while every probe row pays an RF evaluation that
never rejects a row --
+ // exactly the "selectivity 100%" case this pruner exists to remove.
+ //
+ // The canonical `ROW_NUMBER() OVER (ORDER BY dt DESC)` + `rn <= N`
shape is still marked:
+ // it has no PARTITION BY, so it takes the global-limit branch and
emits at most N rows.
+ boolean bounded = partitionTopN.hasGlobalLimit() ||
partitionTopN.getPartitionKeys().isEmpty();
Review Comment:
[P2] Account for RANK ties before marking this source bounded
An empty `PARTITION BY` bounds `ROW_NUMBER`, but not `RANK` or `DENSE_RANK`:
`rank() over (order by flag) <= 1` can return every row tied on the leading
value, and the BE sorter explicitly continues through that peer group. If the
build key is another column, this can retain the full build domain while this
mark bypasses the statistics-based RF pruning. Please require `ROW_NUMBER` for
the empty-key branch (while retaining `hasGlobalLimit()` for a real cap) and
add tied rank/dense-rank cases.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java:
##########
@@ -582,6 +582,13 @@ public ConnectorTableSchema getTableSchema(
perTableCapabilities.add(ConnectorCapability.SUPPORTS_TOPN_LAZY_MATERIALIZE);
perTableCapabilities.add(ConnectorCapability.SUPPORTS_STORAGE_PREDICATE_PRUNING);
}
+ // Partition values of a HIVE table and of a hudi-on-HMS table both
live in the directory path, so
+ // BE can emit one row of partition values per scan range without
opening the file. Delegated
+ // (iceberg/paimon-on-HMS) tables never reach this branch -- they are
served by the sibling branch
+ // above -- which is exactly what keeps them out of the optimization.
+ if (supportsPartitionValueOnly(tableInfo)) {
Review Comment:
[P1] Make the claimed Hudi capability reachable or remove it
Every Hudi-on-HMS table is diverted to a sibling `HudiTableHandle` in
`getTableHandle`, so this HiveTableHandle branch never runs for it. The
sibling-reflection allowlist also omits `SUPPORTS_PARTITION_VALUE_ONLY`, and
the Hudi connector declares no capabilities, making
`supportsPartitionValueOnly()` always false. Please either remove the Hudi
claim or expose a capability from the delegated Hudi path after separately
proving COW/MOR/delete semantics safe, with an end-to-end plan test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java:
##########
@@ -1700,6 +1701,15 @@ public List<Split> getSplits(int numBackends) throws
UserException {
.limit(sourceLimit)
.requiredPartitions(requiredPartitions)
.countPushdown(countPushdown)
+ // Forward the PARTITION_VALUE signal to the connector. The op
is set on this node by the
+ // Nereids translator and shipped to BE via
FileScanNode.toThrift, but split planning is the
+ // connector's job: with partition-column-value-only output
every scan range emits one row
+ // from `columns_from_path`, so splitting one file into
several ranges only produces
+ // duplicate partition-value rows and extra scheduler work.
Min/max stay correct either way
+ // -- splitting is skipped purely to avoid the waste (mirrors
legacy HiveScanNode, which set
+ // needSplit=false for the same op). Connectors that do not
read the field are unaffected.
+ .partitionValuePushdown(
Review Comment:
[P2] Forward this mode on the batch split path too
Large partitioned Hive scans do not use this request builder: `startSplit`
creates a separate `batchRequest` with only `.filter(...)`. Once the selected
partition count reaches the batch threshold, `planScanForPartitionBatch`
therefore sees this flag as false and performs normal file splitting, making V1
emit one duplicate synthetic row per chunk and recreating the scheduler work
this change is meant to avoid. Please set the same flag on the batch request
and cover the batch-mode path.
##########
gensrc/thrift/PlanNodes.thrift:
##########
@@ -1031,7 +1031,11 @@ enum TPushAggOp {
MINMAX = 1,
COUNT = 2,
MIX = 3,
- COUNT_ON_INDEX = 4
+ COUNT_ON_INDEX = 4,
+ // The aggregation only depends on partition columns of an external
table.
+ // The scanner just returns one row (partition column values) per scan
range
+ // without opening/reading any data file.
+ PARTITION_VALUE = 5
Review Comment:
[P1] Update exhaustive BE switches for the new enum value
Adding this generated enum member makes `push_down_agg_to_string` in
`be/src/format_v2/table_reader.cpp` non-exhaustive: it handles the five
previous values but has neither `PARTITION_VALUE` nor a `default`. BE builds
with `-Wall -Wextra -Werror`, so Clang's `-Wswitch` diagnostic is fatal after
Thrift regeneration. Please add the new case (and audit other exhaustive
switches) before this can compile.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java:
##########
@@ -813,4 +929,228 @@ private boolean enablePushDownNoGroupAgg() {
ConnectContext connectContext = ConnectContext.get();
return connectContext == null ||
connectContext.getSessionVariable().enablePushDownNoGroupAgg();
}
+
+ /**
+ * Try the PARTITION_VALUE fast path across a LogicalFilter.
+ *
+ * <p>Safety conditions -- ALL of them must hold:
+ * <ol>
+ * <li>the scan's partitions are already pruned on FE side
+ * ({@link LogicalFileScan.SelectedPartitions#isPruned}), i.e. the
predicate has been fully
+ * resolved from partition metadata rather than left for the data
files;</li>
+ * <li>every conjunct of the filter references partition columns
only;</li>
+ * <li>the aggregate only uses MIN/MAX (group by is allowed) and every
column the scan must
+ * output is a partition column (checked by {@link
#canUsePartitionValueOnly}).</li>
+ * </ol>
+ *
+ * <p>Under these conditions the BE emits exactly one row of partition
column values per scan
+ * range, the filter re-evaluates its predicate on those partition values
(still correct, just
+ * redundant), and the aggregate observes exactly the set of surviving
partition values.
+ *
+ * <p>Condition 2 is implied by condition 3 (the filter can only reference
slots that the scan
+ * outputs), but it is checked explicitly so the intent stays obvious and
so the rule keeps
+ * behaving safely if the plan shape changes later.
+ *
+ * <p>Note this returns {@code aggregate} (the very object handed in by
the rule) when it gives
+ * up, because {@code ApplyRuleJob} compares the returned plan with the
original one by
+ * reference to detect "rule declined to rewrite".
+ */
+ private Plan partitionValueThroughFilter(
+ LogicalAggregate<? extends Plan> aggregate,
+ @Nullable LogicalProject<? extends Plan> project,
+ LogicalFilter<? extends Plan> filter,
+ LogicalFileScan logicalScan,
+ CascadesContext cascadesContext) {
+ final Plan canNotPush = aggregate;
+
+ // (1) partitions must have been pruned from partition metadata on FE
side
+ if (!logicalScan.getSelectedPartitions().isPruned) {
+ return canNotPush;
+ }
+
+ // (2) the filter must touch partition columns only.
+ // The filter sits directly on top of the scan, so its input slots ARE
scan output slots.
+ // Do NOT route them through Project#findProject: that would be a
no-op mapping here and it
+ // throws AnalysisException when an ExprId is missing.
+ Set<SlotReference> filterSlots = ExpressionUtils.collect(
+ filter.getConjuncts(), SlotReference.class::isInstance);
+ if (!isAllPartitionColumns(ImmutableList.copyOf(filterSlots),
logicalScan)) {
+ return canNotPush;
+ }
+
+ // (3) aggregate shape + scan output columns
+ if (!canUsePartitionValueOnly(aggregate, logicalScan)) {
+ return canNotPush;
+ }
+
+ PhysicalFileScan physicalScan = toPhysicalFileScan(logicalScan,
cascadesContext);
+ Plan storageLayerAgg = new PhysicalStorageLayerAggregate(physicalScan,
PushDownAggOp.PARTITION_VALUE);
+ // Keep the LogicalFilter: its conjuncts are still needed and will be
translated onto the
+ // ScanNode by PhysicalPlanTranslator#visitPhysicalFilter. This
mirrors the existing
+ // pushdownCountOnIndex / pushdownMinMaxOnUniqueTable rules, which
also return a logical
+ // filter wrapping a PhysicalStorageLayerAggregate.
+ Plan newFilter =
filter.withChildren(ImmutableList.of(storageLayerAgg));
+ if (project != null) {
+ return aggregate.withChildren(ImmutableList.of(
+ project.withChildren(ImmutableList.of(newFilter))));
+ }
+ return aggregate.withChildren(ImmutableList.of(newFilter));
+ }
+
+ /**
+ * Aggregate-shape check for the PARTITION_VALUE fast path.
+ *
+ * <p>The decision is keyed off <b>the columns the scan must output</b>,
not off the arguments of
+ * the aggregate functions. This matters a lot, because {@code
ConstantPropagation} rewrites
+ * <pre>
+ * select max(dt) from tbl where dt = '2026-08-11'
+ * --> max('2026-08-11')
+ * </pre>
+ * leaving no SlotReference at all inside the aggregate. Keying off the
aggregate arguments (which
+ * is what the legacy `partitionValueOnly` check in storageLayerAggregate
does) silently loses the
+ * optimization for exactly the most common query pattern, while the scan
still has to output `dt`
+ * because the filter references it.
+ *
+ * <p><b>Why only MIN/MAX, and why GROUP BY is fine.</b> PARTITION_VALUE
makes the scan emit one
+ * row per data file instead of the file's real rows, so the row multiset
is wrong while the
+ * <i>distinct set</i> of partition column values is exactly right (every
non-empty file
+ * contributes its partition value once). MIN/MAX are idempotent w.r.t.
duplicates -- they only
+ * depend on the distinct set -- so they stay correct. GROUP BY likewise
only depends on the
+ * distinct set of its keys, and its keys can only reference partition
columns here (the scan
+ * outputs nothing else). Hence all of these are safe:
+ * <pre>
+ * select max(dt) from t where dt >= '2025-01-01' group by dt --
correct
+ * select min(dt), max(dt) from t group by hh --
correct
+ * select max(hh) from t where dt >= '2025-01-01' group by dt --
correct (see below)
+ * select max(hh) from t group by substr(dt, 1, 7) --
correct
+ * </pre>
+ * COUNT/SUM/AVG are NOT: they depend on the row multiset, so a count()
would silently return
+ * the number of FILES.
+ *
+ * <p><b>The group-by key and the aggregated column may be DIFFERENT
partition columns.</b>
+ * For {@code select max(hh) ... group by dt} with partitions (d, h):
since `hh` is a partition
+ * column, every row's `hh` equals the `h` of its own partition. So the
true result per d is
+ * {@code max{h : partition(d,h) has rows}} while PARTITION_VALUE yields
+ * {@code max{h : partition(d,h) has at least one file}} -- differing only
for partitions whose
+ * files are all empty, which is the pre-existing "empty file" caveat, not
a new one.
+ * This is also why no check on the relationship between the group-by key
and the aggregated
+ * column is needed: a group-by key can only reference columns the scan
outputs, and
+ * isAllPartitionColumns(scanOutputSlots(...)) already guarantees those
are all partition columns.
+ */
+ private boolean canUsePartitionValueOnly(
+ LogicalAggregate<? extends Plan> aggregate, LogicalFileScan
logicalScan) {
+ if (!enablePartitionColumnValueOnly()) {
+ return false;
+ }
+ // count(distinct x) / sum(distinct x): distinct semantics inside an
aggregate function is
+ // computed over the real row multiset, and count(distinct) would need
the real rows.
+ if (!aggregate.getDistinctArguments().isEmpty()) {
+ return false;
+ }
+ Set<AggregateFunction> aggregateFunctions =
aggregate.getAggregateFunctions();
+ // A LogicalAggregate always has at least a group by key or an
aggregate function; require it
+ // explicitly so a degenerate aggregate never reaches the fast path.
+ if (aggregateFunctions.isEmpty() &&
aggregate.getGroupByExpressions().isEmpty()) {
+ return false;
+ }
+ for (AggregateFunction function : aggregateFunctions) {
+ if (!(function instanceof Min) && !(function instanceof Max)) {
+ return false;
+ }
+ }
+ return isAllPartitionColumns(scanOutputSlots(logicalScan),
logicalScan);
+ }
+
+ /**
+ * The columns the scan actually has to read (its OPERATIVE slots), or an
empty list if any of
+ * them is not a plain SlotReference (an empty list makes {@link
#isAllPartitionColumns} bail out).
+ *
+ * <p>Must use {@code getOperativeSlots()}, NOT {@code getOutput()}:
{@code getOutput()} is the
+ * scan's full nominal schema (e.g. {@code [id, name, dt]}) even when
+ * a Project above only needs {@code dt}, whereas column pruning trims the
operative slots to the
+ * columns really materialized ({@code [dt]}). This also matches BE, whose
PARTITION_VALUE fast
+ * path only fires when no non-partition file slot is materialized
+ * ({@code _file_slot_descs.empty()}). Keying off {@code getOutput()}
makes the check fail for
+ * every table that has non-partition columns, which is the common case.
+ */
+ private List<SlotReference> scanOutputSlots(LogicalFileScan logicalScan) {
+ List<Slot> operative = logicalScan.getOperativeSlots();
+ if (operative.isEmpty()) {
+ return ImmutableList.of();
+ }
+ ImmutableList.Builder<SlotReference> slots =
+ ImmutableList.builderWithExpectedSize(operative.size());
+ for (Slot slot : operative) {
+ if (!(slot instanceof SlotReference)) {
+ return ImmutableList.of();
+ }
+ slots.add((SlotReference) slot);
+ }
+ return slots.build();
+ }
+
+ /** Implement a LogicalFileScan into its PhysicalFileScan. */
+ private PhysicalFileScan toPhysicalFileScan(
+ LogicalFileScan logicalScan, CascadesContext cascadesContext) {
+ Rule rule = new LogicalFileScanToPhysicalFileScan().build();
+ return (PhysicalFileScan) rule.transform(logicalScan,
cascadesContext).get(0);
+ }
+
+ private boolean enablePartitionColumnValueOnly() {
+ ConnectContext connectContext = ConnectContext.get();
+ return connectContext == null
+ ||
connectContext.getSessionVariable().isEnablePartitionColumnValueOnlyOptimization();
+ }
+
+ /**
+ * Check whether all the slots used by the aggregate functions are
partition columns of the
+ * scanned external table. Only in this case can we rely purely on
partition metadata (the
+ * scanner returns partition column values per scan range) without reading
any data file.
+ */
+ private boolean isAllPartitionColumns(List<SlotReference> usedSlotInTable,
LogicalFileScan fileScan) {
+ if (usedSlotInTable.isEmpty()) {
+ return false;
+ }
+ // Partition values must be reconstructible from the data file path.
Hive and hudi-on-HMS both
+ // fill them from the path on BE side (columns_from_path), so both
qualify.
+ //
+ // ATTN: this must be a per-table capability check, NOT a table-class
check. Iceberg and Paimon
+ // tables reached through an HMS catalog (`type = hms`) are served by
the very same
+ // PluginDrivenExternalTable class, and
supportInternalPartitionPruned() is true for them too, so
+ // PruneFileScanPartition marks their SelectedPartitions as pruned and
the `isPruned` guard does
+ // not filter them out either. They are excluded because their
partition values do NOT come from
+ // the file path: Iceberg supports hidden partitioning and partition
transforms (bucket,
+ // truncate, days, ...) that cannot be reconstructed from the path,
and v2 position/equality
+ // deletes break the "one row per scan range" assumption; Paimon
resolves partitions from its
+ // own metadata.
+ if (!(fileScan.getTable() instanceof PluginDrivenExternalTable)) {
+ return false;
+ }
+ PluginDrivenExternalTable table = (PluginDrivenExternalTable)
fileScan.getTable();
+ if (!table.supportsPartitionValueOnly()) {
+ return false;
+ }
+ Set<String> partitionColumnNames = new HashSet<>();
+ try {
+ List<Column> partitionColumns =
table.getPartitionColumns(Optional.empty());
+ if (partitionColumns == null || partitionColumns.isEmpty()) {
+ return false;
+ }
+ for (Column column : partitionColumns) {
+ partitionColumnNames.add(column.getName().toLowerCase());
+ }
+ } catch (Throwable t) {
Review Comment:
[P1] Do not turn fatal errors into an optimization miss
`getPartitionColumns(Optional)` declares no recoverable checked failure and
already returns an empty list for a cache miss, while
`supportsPartitionValueOnly()` has initialized the same table just above.
Catching `Throwable` here also swallows `OutOfMemoryError`,
`StackOverflowError`, `LinkageError`, and schema-cache invariant failures, then
continues into the heavier normal scan. Please remove this catch or narrow it
to a concrete recoverable metadata exception whose fallback semantics are
understood.
--
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]
