924060929 commented on code in PR #65569:
URL: https://github.com/apache/doris/pull/65569#discussion_r3578201849
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazyMaterializeTopN.java:
##########
@@ -19,301 +19,170 @@
import org.apache.doris.catalog.AggregateType;
import org.apache.doris.catalog.Column;
-import org.apache.doris.catalog.KeysType;
-import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.TableIf;
import org.apache.doris.catalog.Type;
import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.processor.post.PlanPostProcessor;
import org.apache.doris.nereids.processor.post.Validator;
-import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
import org.apache.doris.nereids.trees.plans.AbstractPlan;
import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.RelationId;
import org.apache.doris.nereids.trees.plans.algebra.CatalogRelation;
import org.apache.doris.nereids.trees.plans.algebra.Relation;
-import org.apache.doris.nereids.trees.plans.physical.PhysicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalJoin;
import org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterialize;
import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
import org.apache.doris.nereids.trees.plans.physical.PhysicalTVFRelation;
import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
import org.apache.doris.qe.SessionVariable;
-import com.google.common.collect.BiMap;
-import com.google.common.collect.HashBiMap;
import com.google.common.collect.ImmutableList;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
+import java.util.TreeMap;
/**
* Post rule to insert MaterializeNode for TopN lazy materialization.
- * Expression pull-up is handled by PullUpProjectExprUnderTopN in the logical
phase.
+ * Expression pull-up is handled by PullUpProjectExprUnderTopN in the logical
phase. That stage may only move
+ * expressions accepted by TopNLazyExpressionEligibility and must preserve
their input slots below TopN. This
+ * physical stage never moves expressions; it resolves source lineage, chooses
a fetch schedule, and rewrites slots.
+ *
+ * <p>Separate branches receive independent specs. For nested TopNs, the
outermost eligible TopN wins so that
+ * materialization remains above the complete local/merge TopN chain. Inner
TopNs are considered only when the
+ * outer candidate is not applicable.
*/
public class LazyMaterializeTopN extends PlanPostProcessor {
private static final Logger LOG =
LogManager.getLogger(LazyMaterializeTopN.class);
- private boolean hasMaterialized = false;
@Override
public Plan visitPhysicalTopN(PhysicalTopN<? extends Plan> topN,
CascadesContext ctx) {
- try {
- Plan result = computeTopN(topN, ctx);
- if (SessionVariable.isFeDebug()) {
- Validator validator = new Validator();
- validator.processRoot(result, ctx);
- }
+ Plan result = computeTopN(topN);
+ if (result != topN) {
+ new Validator().processRoot(result, ctx);
Review Comment:
Robustness: this drops the previous defensive fallback. The old
`visitPhysicalTopN`/`computeTopN` wrapped analysis in `try { ... } catch (...)
{ LOG.warn(); return topN; }`, and the `Validator` only ran under
`SessionVariable.isFeDebug()`. Now there is no guard and
`Validator.processRoot` runs unconditionally whenever materialization applies.
`PlanPostProcessors.process()` is itself unguarded, so any
`RuntimeException` from the new analyzer/rewriter/spec -- and there are many
`Preconditions.checkArgument`/`checkNotNull`/`IllegalStateException`/casts (the
`TopNLazyMaterializationSpec` ctor invariants, `DeferredColumnSpec.from`,
`restoreSourceMetadata`'s `getOriginalColumn().get()`, ...) -- now fails the
whole query instead of degrading to an eager TopN.
For a session-var-gated optimization that's a robustness regression.
Suggestion: keep a top-level `try/catch` that logs and returns the original
`topN`, and gate this eager `Validator` behind `isFeDebug()` (the
post-processor pipeline already ends with a `Validator`, so prod validation
isn't lost). Fail-fast in debug, degrade in prod.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazyMaterializeTopN.java:
##########
@@ -19,301 +19,170 @@
import org.apache.doris.catalog.AggregateType;
import org.apache.doris.catalog.Column;
-import org.apache.doris.catalog.KeysType;
-import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.TableIf;
import org.apache.doris.catalog.Type;
import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.processor.post.PlanPostProcessor;
import org.apache.doris.nereids.processor.post.Validator;
-import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
import org.apache.doris.nereids.trees.plans.AbstractPlan;
import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.RelationId;
import org.apache.doris.nereids.trees.plans.algebra.CatalogRelation;
import org.apache.doris.nereids.trees.plans.algebra.Relation;
-import org.apache.doris.nereids.trees.plans.physical.PhysicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalJoin;
import org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterialize;
import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
import org.apache.doris.nereids.trees.plans.physical.PhysicalTVFRelation;
import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
import org.apache.doris.qe.SessionVariable;
-import com.google.common.collect.BiMap;
-import com.google.common.collect.HashBiMap;
import com.google.common.collect.ImmutableList;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
+import java.util.TreeMap;
/**
* Post rule to insert MaterializeNode for TopN lazy materialization.
- * Expression pull-up is handled by PullUpProjectExprUnderTopN in the logical
phase.
+ * Expression pull-up is handled by PullUpProjectExprUnderTopN in the logical
phase. That stage may only move
+ * expressions accepted by TopNLazyExpressionEligibility and must preserve
their input slots below TopN. This
+ * physical stage never moves expressions; it resolves source lineage, chooses
a fetch schedule, and rewrites slots.
+ *
+ * <p>Separate branches receive independent specs. For nested TopNs, the
outermost eligible TopN wins so that
+ * materialization remains above the complete local/merge TopN chain. Inner
TopNs are considered only when the
+ * outer candidate is not applicable.
*/
public class LazyMaterializeTopN extends PlanPostProcessor {
private static final Logger LOG =
LogManager.getLogger(LazyMaterializeTopN.class);
- private boolean hasMaterialized = false;
@Override
public Plan visitPhysicalTopN(PhysicalTopN<? extends Plan> topN,
CascadesContext ctx) {
- try {
- Plan result = computeTopN(topN, ctx);
- if (SessionVariable.isFeDebug()) {
- Validator validator = new Validator();
- validator.processRoot(result, ctx);
- }
+ Plan result = computeTopN(topN);
+ if (result != topN) {
+ new Validator().processRoot(result, ctx);
return result;
- } catch (Exception e) {
- LOG.warn("lazy materialize topn failed", e);
- return topN;
}
+ return DefaultPlanRewriter.visitChildren(this, topN, ctx);
Review Comment:
Removing the old `hasMaterialized` flag means a plan can now contain
multiple `PhysicalLazyMaterialize` nodes (one per independent branch), whereas
before there was at most one per query -- e.g. `(SELECT ... ORDER BY x LIMIT
10) UNION ALL (SELECT ... ORDER BY y LIMIT 10)` can now emit two materialize
nodes with two independent row-id column sets.
Per-node the thrift format is unchanged, but "does not alter the BE
execution protocol" holds per-node, not per-plan -- going from <=1 to N
materialize nodes is a real change in what BE sees. Could we confirm BE has no
implicit single-materialize / single-row-id assumption, and add a regression
case with two independent lazy TopNs? The current `topn_lazy` suites look
single-branch.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/SlotLineageAnalyzer.java:
##########
@@ -0,0 +1,304 @@
+// 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.
+
+package org.apache.doris.nereids.processor.post.materialize;
+
+import org.apache.doris.catalog.HiveTable;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.datasource.hive.HMSExternalTable;
+import org.apache.doris.datasource.iceberg.IcebergExternalTable;
+import
org.apache.doris.nereids.processor.post.materialize.MaterializationAnalysisResult.NotApplicableReason;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.algebra.Relation;
+import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalFilter;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterialize;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalOneRowRelation;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalRepeat;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalSetOperation;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalTVFRelation;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Stateless bottom-up resolver from an output slot to its physical
source-column lineage. */
+public final class SlotLineageAnalyzer {
+ private static final ImmutableSet<Class<?>> SUPPORTED_RELATION_TYPES =
ImmutableSet.of(
+ OlapTable.class, HiveTable.class, IcebergExternalTable.class,
HMSExternalTable.class);
+
+ public MaterializationAnalysisResult<OutputLineage> analyze(Plan plan,
SlotReference slot) {
+ return analyze(plan, slot, new IdentityHashMap<>());
+ }
+
+ /** Resolve all requested outputs with one shared lineage cache for this
candidate TopN. */
+ public Map<Slot, MaterializationAnalysisResult<OutputLineage>> analyze(
+ Plan plan, List<? extends Slot> slots) {
+ Map<Plan, Map<SlotReference,
MaterializationAnalysisResult<OutputLineage>>> cache =
+ new IdentityHashMap<>();
+ Map<Slot, MaterializationAnalysisResult<OutputLineage>> results = new
LinkedHashMap<>();
+ for (Slot slot : slots) {
+ if (slot instanceof SlotReference) {
+ results.put(slot, analyzeFromProducer(plan, (SlotReference)
slot, cache));
+ }
+ }
+ return results;
+ }
+
+ private MaterializationAnalysisResult<OutputLineage> analyze(Plan plan,
SlotReference slot,
+ Map<Plan, Map<SlotReference,
MaterializationAnalysisResult<OutputLineage>>> cache) {
+ Map<SlotReference, MaterializationAnalysisResult<OutputLineage>>
planCache =
+ cache.computeIfAbsent(plan, ignored -> new LinkedHashMap<>());
+ MaterializationAnalysisResult<OutputLineage> cached =
planCache.get(slot);
+ if (cached != null) {
+ return cached;
+ }
+ MaterializationAnalysisResult<OutputLineage> result =
analyzeUncached(plan, slot, cache);
+ planCache.put(slot, result);
+ return result;
+ }
+
+ private MaterializationAnalysisResult<OutputLineage>
analyzeFromProducer(Plan plan, SlotReference slot,
+ Map<Plan, Map<SlotReference,
MaterializationAnalysisResult<OutputLineage>>> cache) {
+ if (plan.getOutput().contains(slot)) {
+ return analyze(plan, slot, cache);
+ }
+ List<Plan> matchingChildren = new ArrayList<>();
+ for (Plan child : plan.children()) {
+ if (containsOutput(child, slot)) {
+ matchingChildren.add(child);
+ }
+ }
+ if (matchingChildren.size() != 1) {
+ return notApplicable(NotApplicableReason.AMBIGUOUS_LINEAGE,
+ "slot " + slot + " is produced by " +
matchingChildren.size()
+ + " subtrees of " +
plan.getClass().getSimpleName());
+ }
+ return analyzeFromProducer(matchingChildren.get(0), slot, cache);
+ }
+
+ private boolean containsOutput(Plan plan, SlotReference slot) {
Review Comment:
`containsOutput` is an uncached full-subtree scan, and `analyzeFromProducer`
calls it per child while descending; only the per-`(plan, slot)` `analyze`
result is cached. Combined with
`TopNLazyMaterializationAnalyzer.collectInputSlots` resolving lineage for
*every* operator input slot in the whole subtree (not just `effectiveOutput`),
worst case is about O(S * depth * P).
Fine for typical small-limit TopN, but wide/deep subtrees (hundreds of
columns) could add measurable planning latency. Consider caching containment,
or resolving only `effectiveOutput` plus `requiredEagerSlots`.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/TopNLazyMaterializationRewriter.java:
##########
@@ -0,0 +1,185 @@
+// 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.
+
+package org.apache.doris.nereids.processor.post.materialize;
+
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalPlan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalFileScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalFilter;
+import
org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterializeFileScan;
+import
org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterializeOlapScan;
+import
org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterializeTVFScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalRelation;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalTVFRelation;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** One-pass physical rewrite driven only by an immutable TopN
lazy-materialization spec. */
+public final class TopNLazyMaterializationRewriter {
+ private final TopNLazyMaterializationSpec spec;
+ private final Map<RelationId, LazySourceSpec> sourcesByRelationId;
+ private final Map<RelationId, List<Slot>> baseSlotsByRelationId;
+
+ /** Create a one-shot rewriter driven by the supplied specification. */
+ public TopNLazyMaterializationRewriter(TopNLazyMaterializationSpec spec) {
+ this.spec = Preconditions.checkNotNull(spec, "spec must not be null");
+ this.sourcesByRelationId =
ImmutableMap.copyOf(spec.getSourceByRelationId());
+ Map<RelationId, List<Slot>> baseSlots = new HashMap<>();
+ for (LazySourceSpec source : spec.getSources()) {
+ ImmutableList.Builder<Slot> relationBaseSlots =
ImmutableList.builder();
+ for (DeferredColumnSpec deferred : source.getDeferredColumns()) {
+ relationBaseSlots.add(deferred.getBaseSlot());
+ }
+ baseSlots.put(source.getRelation().getRelationId(),
relationBaseSlots.build());
+ }
+ baseSlotsByRelationId = ImmutableMap.copyOf(baseSlots);
+ }
+
+ /** Rewrite the candidate TopN subtree exactly once. */
+ public Plan rewrite(Plan plan) {
+ if (plan instanceof PhysicalProject) {
+ return rewriteProject((PhysicalProject<?>) plan);
+ }
+ if (plan instanceof PhysicalFilter) {
+ return rewriteFilter((PhysicalFilter<?>) plan);
+ }
+ if (plan instanceof PhysicalRelation) {
+ return rewriteRelation((PhysicalRelation) plan);
+ }
+ return rewriteChildren(plan);
+ }
+
+ private Plan rewriteProject(PhysicalProject<?> project) {
+ Plan rewrittenChild = rewrite(project.child());
+ List<NamedExpression> projections = new ArrayList<>();
+ for (NamedExpression projection : project.getProjects()) {
+ if (!hasDeferredInput(projection, rewrittenChild)) {
+ projections.add(projection);
+ }
+ }
+ Set<Slot> projectedSlots = new LinkedHashSet<>();
+ for (NamedExpression projection : projections) {
+ projectedSlots.add(projection.toSlot());
+ }
+ for (Slot outputSlot : rewrittenChild.getOutput()) {
+ if (isAssignedRowId(outputSlot) &&
!projectedSlots.contains(outputSlot)) {
+ projections.add((SlotReference) outputSlot);
+ }
+ }
+ if (rewrittenChild == project.child() &&
projections.equals(project.getProjects())) {
+ return project;
+ }
+ return project.withProjectionsAndChild(projections,
rewrittenChild).resetLogicalProperties();
+ }
+
+ private Plan rewriteFilter(PhysicalFilter<?> filter) {
+ if (!SessionVariable.getTopNLazyMaterializationUsingIndex()
+ || !(filter.child() instanceof PhysicalOlapScan)) {
+ return rewriteChildren(filter);
+ }
+ PhysicalOlapScan scan = (PhysicalOlapScan) filter.child();
+ LazySourceSpec source = sourcesByRelationId.get(scan.getRelationId());
+ if (source == null) {
+ return filter;
Review Comment:
Minor/readability: when the index path applies but the scan isn't a lazy
source, this returns the original `filter` rather than
`rewriteChildren(filter)` (used in the fall-through above). It's equivalent
here because `filter.child()` is a leaf `PhysicalOlapScan` with nothing below
to rewrite, but it reads as an inconsistency -- a one-line comment ("child is
the scan; nothing below to rewrite") would prevent a future footgun.
--
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]
