morrySnow commented on code in PR #65569:
URL: https://github.com/apache/doris/pull/65569#discussion_r3578426377
##########
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:
Thanks. I restored the debug-only gating for the eager Validator; the final
postprocessor pipeline still runs Validator in all modes. I did not restore a
catch-all fallback: expected ineligibility is represented by NotApplicable,
while Preconditions and IllegalStateException indicate violated planner
invariants. Doris coding policy requires those invariant failures to surface
instead of being silently converted to an eager plan.
##########
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:
Confirmed on BE: every MATERIALIZATION_NODE creates its own
MaterializationOperator in pipeline_fragment_context, and the thrift node plus
RPC, row-id, and response state are owned per operator/local state. There is no
shared single-materialize state. I added a regression with two independent
TopNs under UNION ALL that asserts two VMaterializeNodes and executes the plan.
##########
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:
Added a containment cache keyed by plan identity and slot, shared across
batch lineage resolution. Each plan-and-slot subtree containment result is now
computed at most once, removing repeated full-subtree scans while retaining
short-circuit traversal.
##########
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:
Changed this branch to rewriteChildren(filter), so it stays consistent and
remains correct if the index-filter child shape grows in the future.
--
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]
