NereidsPlanner.java
// 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;
import org.apache.doris.analysis.DescriptorTable;
import org.apache.doris.analysis.ExplainOptions;
import org.apache.doris.analysis.StatementBase;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.NereidsException;
import org.apache.doris.common.Pair;
import org.apache.doris.common.UserException;
import org.apache.doris.common.profile.SummaryProfile;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.common.util.Util;
import org.apache.doris.foundation.format.FormatOptions;
import org.apache.doris.mysql.FieldInfo;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
import org.apache.doris.nereids.glue.translator.PhysicalPlanTranslator;
import org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
import org.apache.doris.nereids.hint.DistributeHint;
import org.apache.doris.nereids.hint.Hint;
import org.apache.doris.nereids.jobs.executor.Optimizer;
import org.apache.doris.nereids.jobs.executor.Rewriter;
import org.apache.doris.nereids.memo.Group;
import org.apache.doris.nereids.memo.GroupExpression;
import org.apache.doris.nereids.memo.Memo;
import org.apache.doris.nereids.processor.post.PlanPostProcessors;
import org.apache.doris.nereids.processor.pre.PlanPreprocessors;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.rules.exploration.mv.MaterializationContext;
import org.apache.doris.nereids.rules.exploration.mv.MaterializedViewUtils;
import org.apache.doris.nereids.rules.exploration.mv.PreMaterializedViewRewriter;
import org.apache.doris.nereids.stats.GroupStructInfo;
import org.apache.doris.nereids.stats.HboJoinConditions;
import org.apache.doris.nereids.stats.HboPlanInfoProvider;
import org.apache.doris.nereids.stats.StatsCalculator;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.plans.AbstractPlan;
import org.apache.doris.nereids.trees.plans.ComputeResultSet;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.algebra.CatalogRelation;
import org.apache.doris.nereids.trees.plans.commands.ExplainCommand.ExplainLevel;
import org.apache.doris.nereids.trees.plans.distribute.DistributePlanner;
import org.apache.doris.nereids.trees.plans.distribute.DistributedPlan;
import org.apache.doris.nereids.trees.plans.distribute.FragmentIdMapping;
import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalSqlCache;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalJoin;
import org.apache.doris.nereids.trees.plans.physical.PhysicalDictionarySink;
import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute;
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.PhysicalOlapScan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapTableSink;
import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
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.PhysicalSqlCache;
import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
import org.apache.doris.nereids.trees.plans.physical.TopnFilter;
import org.apache.doris.nereids.util.MutableState;
import org.apache.doris.planner.AddLocalExchange;
import org.apache.doris.planner.PlanFragment;
import org.apache.doris.planner.PlanNodeId;
import org.apache.doris.planner.Planner;
import org.apache.doris.planner.RuntimeFilter;
import org.apache.doris.planner.ScanNode;
import org.apache.doris.planner.normalize.QueryCacheNormalizer;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.ResultSet;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.qe.TimeBasedChangeVisibleWaiter;
import org.apache.doris.qe.VariableMgr;
import org.apache.doris.qe.cache.CacheAnalyzer;
import org.apache.doris.statistics.query.QueryStatsRecorder;
import org.apache.doris.statistics.util.StatisticsUtil;
import org.apache.doris.thrift.TQueryCacheParam;
import org.apache.doris.tso.MasterTsoProvider;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import org.apache.commons.codec.binary.Hex;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.lang.management.GarbageCollectorMXBean;
import java.lang.management.ManagementFactory;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;
/**
* Planner to do query plan in Nereids.
*/
public class NereidsPlanner extends Planner {
// private static final AtomicInteger executeCount = new AtomicInteger(0);
public static final Logger LOG = LogManager.getLogger(NereidsPlanner.class);
public static AtomicLong runningPlanNum = new AtomicLong(0L);
protected Plan parsedPlan;
protected Plan analyzedPlan;
protected Plan rewrittenPlan;
protected Plan optimizedPlan;
protected PhysicalPlan physicalPlan;
private CascadesContext cascadesContext;
private final StatementContext statementContext;
private final List<ScanNode> scanNodeList = Lists.newArrayList();
private final List<PhysicalRelation> physicalRelations = Lists.newArrayList();
private DescriptorTable descTable;
private FragmentIdMapping<DistributedPlan> distributedPlans;
private PlanTranslatorContext planTranslatorContext;
// The cost of optimized plan
private double cost = 0;
private LogicalPlanAdapter logicalPlanAdapter;
public NereidsPlanner(StatementContext statementContext) {
this.statementContext = statementContext;
}
@Override
public void plan(StatementBase queryStmt, org.apache.doris.thrift.TQueryOptions queryOptions) throws UserException {
this.queryOptions = queryOptions;
if (!(queryStmt instanceof LogicalPlanAdapter)) {
throw new RuntimeException("Wrong type of queryStmt, expected: <? extends LogicalPlanAdapter>");
}
logicalPlanAdapter = (LogicalPlanAdapter) queryStmt;
ExplainLevel explainLevel = getExplainLevel(queryStmt.getExplainOptions());
LogicalPlan parsedPlan = logicalPlanAdapter.getLogicalPlan();
setParsedPlan(parsedPlan);
PhysicalProperties requireProperties = buildInitRequireProperties();
statementContext.getStopwatch().reset().start();
NereidsPlanner.runningPlanNum.incrementAndGet();
try {
boolean showPlanProcess = showPlanProcess(queryStmt.getExplainOptions());
planWithLock(parsedPlan, requireProperties, explainLevel, showPlanProcess, plan -> {
setOptimizedPlan(plan);
if (plan instanceof PhysicalPlan) {
physicalPlan = (PhysicalPlan) plan;
distribute(physicalPlan, explainLevel);
}
});
} finally {
statementContext.getStopwatch().stop();
NereidsPlanner.runningPlanNum.decrementAndGet();
}
if (LOG.isDebugEnabled()) {
LOG.info(getExplainString(new ExplainOptions(ExplainLevel.SHAPE_PLAN, false)));
LOG.info(getExplainString(new ExplainOptions(ExplainLevel.DISTRIBUTED_PLAN, false)));
}
if (physicalPlan != null) {
QueryStatsRecorder.record(physicalPlan, statementContext);
}
}
@VisibleForTesting
public void plan(StatementBase queryStmt) {
try {
plan(queryStmt, statementContext.getConnectContext().getSessionVariable().toThrift());
} catch (Exception e) {
throw new NereidsException(e.getMessage(), e);
}
}
@VisibleForTesting
public PhysicalPlan planWithLock(LogicalPlan plan, PhysicalProperties outputProperties) {
return (PhysicalPlan) planWithLock(plan, outputProperties, ExplainLevel.NONE, false);
}
// TODO check all caller
public Plan planWithLock(LogicalPlan plan, PhysicalProperties requireProperties, ExplainLevel explainLevel) {
return planWithLock(plan, requireProperties, explainLevel, false);
}
@VisibleForTesting
public Plan planWithLock(LogicalPlan plan, PhysicalProperties requireProperties,
ExplainLevel explainLevel, boolean showPlanProcess) {
Consumer<Plan> noCallback = p -> {};
return planWithLock(plan, requireProperties, explainLevel, showPlanProcess, noCallback);
}
/**
* Do analyze and optimize for query plan.
*
* @param plan wait for plan
* @param requireProperties request physical properties constraints
* @param showPlanProcess is record plan process to CascadesContext
* @param lockCallback this callback function will invoke the table lock
* @return plan generated by this planner
* @throws AnalysisException throw exception if failed in ant stage
*/
private Plan planWithLock(LogicalPlan plan, PhysicalProperties requireProperties,
ExplainLevel explainLevel, boolean showPlanProcess, Consumer<Plan> lockCallback) {
try {
long beforePlanGcTime = getGarbageCollectionTime();
if (plan instanceof LogicalSqlCache) {
rewrittenPlan = analyzedPlan = plan;
LogicalSqlCache logicalSqlCache = (LogicalSqlCache) plan;
optimizedPlan = physicalPlan = new PhysicalSqlCache(
logicalSqlCache.getQueryId(),
logicalSqlCache.getColumnLabels(), logicalSqlCache.getFieldInfos(),
logicalSqlCache.getResultExprs(), logicalSqlCache.getResultSetInFe(),
logicalSqlCache.getCacheValues(), logicalSqlCache.getBackendAddress(),
logicalSqlCache.getPlanBody()
);
if (explainLevel != ExplainLevel.NONE) {
this.cascadesContext = CascadesContext.initContext(
statementContext, parsedPlan, PhysicalProperties.ANY);
switch (explainLevel) {
case OPTIMIZED_PLAN:
case ALL_PLAN:
cascadesContext.addPlanProcess(
new PlanProcess("ImplementSqlCache",
parsedPlan.treeString(false, parsedPlan),
physicalPlan.treeString(false, physicalPlan)
)
);
break;
default: {
}
}
}
return physicalPlan;
}
if (explainLevel == ExplainLevel.PARSED_PLAN || explainLevel == ExplainLevel.ALL_PLAN) {
parsedPlan = plan;
if (explainLevel == ExplainLevel.PARSED_PLAN) {
return parsedPlan;
}
}
// pre-process logical plan out of memo, e.g. process SET_VAR hint
plan = preprocess(plan);
initCascadesContext(plan, requireProperties);
// collect table and lock them in the order of table id
collectAndLockTable(showAnalyzeProcess(explainLevel, showPlanProcess),
explainLevel == ExplainLevel.NONE);
// after table collector, we should use a new context.
Plan resultPlan = planWithoutLock(plan, requireProperties, explainLevel, showPlanProcess);
lockCallback.accept(resultPlan);
if (statementContext.getConnectContext().getExecutor() != null) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setNereidsGarbageCollectionTime(getGarbageCollectionTime() - beforePlanGcTime);
}
return resultPlan;
} finally {
statementContext.releasePlannerResources();
}
}
/**
* do plan but not lock any table
*/
private Plan planWithoutLock(
LogicalPlan plan, PhysicalProperties requireProperties, ExplainLevel explainLevel,
boolean showPlanProcess) {
// analyze this query, resolve column, table and function
analyze(showAnalyzeProcess(explainLevel, showPlanProcess));
if (explainLevel == ExplainLevel.ANALYZED_PLAN || explainLevel == ExplainLevel.ALL_PLAN) {
analyzedPlan = cascadesContext.getRewritePlan();
if (explainLevel == ExplainLevel.ANALYZED_PLAN) {
return analyzedPlan;
}
}
// rule-based optimize
rewrite(showRewriteProcess(explainLevel, showPlanProcess));
// try to pre mv rewrite
preMaterializedViewRewrite();
if (explainLevel == ExplainLevel.REWRITTEN_PLAN || explainLevel == ExplainLevel.ALL_PLAN) {
rewrittenPlan = cascadesContext.getRewritePlan();
if (explainLevel == ExplainLevel.REWRITTEN_PLAN) {
return rewrittenPlan;
}
}
optimize(showPlanProcess);
// print memo before choose plan.
// if chooseNthPlan failed, we could get memo to debug
if (cascadesContext.getConnectContext().getSessionVariable().dumpNereidsMemo) {
Memo memo = cascadesContext.getMemo();
if (memo != null) {
LOG.info("{}\n{}", ConnectContext.get().getQueryIdentifier(), memo.toString());
} else {
LOG.info("{}\nMemo is null", ConnectContext.get().getQueryIdentifier());
}
}
Set<Integer> requiredGroupIds = cascadesContext.getConnectContext()
.getSessionVariable().getRequiredGroupIds();
PhysicalPlan physicalPlan;
if (!requiredGroupIds.isEmpty()) {
Map<Group, Set<Integer>> reachableCache = new HashMap<>();
computeReachableGroupIds(getRoot(), reachableCache);
// Validate all required groups are reachable from root
Set<Integer> rootReachable = reachableCache.get(getRoot());
Set<Integer> unreachable = new HashSet<>(requiredGroupIds);
unreachable.removeAll(rootReachable);
if (!unreachable.isEmpty()) {
throw new AnalysisException("Required group IDs not found in memo: " + unreachable);
}
physicalPlan = chooseBestPlanWithRequiredGroups(
getRoot(), requireProperties, requiredGroupIds, reachableCache);
} else {
int nth = cascadesContext.getConnectContext().getSessionVariable().getNthOptimizedPlan();
physicalPlan = chooseNthPlan(getRoot(), requireProperties, nth);
}
physicalPlan = postProcess(physicalPlan);
if (cascadesContext != null
&& cascadesContext.getMemo() != null
&& ConnectContext.get() != null
&& ConnectContext.get().getSessionVariable().isShowHboFingerprint()) {
Map<Integer, Group> hboGroupsById = new HashMap<>();
for (Group group : cascadesContext.getMemo().getGroups()) {
hboGroupsById.put(group.getGroupId().asInt(), group);
}
attachHboExplainInfoToTree(physicalPlan, hboGroupsById);
}
if (cascadesContext.getConnectContext().getSessionVariable().dumpNereidsMemo) {
String tree = physicalPlan.treeString();
LOG.info("{}\n{}", ConnectContext.get().getQueryIdentifier(), tree);
}
if (explainLevel == ExplainLevel.OPTIMIZED_PLAN
|| explainLevel == ExplainLevel.ALL_PLAN
|| explainLevel == ExplainLevel.SHAPE_PLAN) {
optimizedPlan = physicalPlan;
}
return physicalPlan;
}
protected LogicalPlan preprocess(LogicalPlan logicalPlan) {
return new PlanPreprocessors(statementContext).process(logicalPlan);
}
/**
* config rf wait time if wait time is the same as default value
* 1. local mode, config according to max table row count
* a. olap table:
* row < 1G: 1 sec
* 1G <= row < 10G: 5 sec
* 10G < row: 20 sec
* b. external table:
* row < 1G: 5 sec
* 1G <= row < 10G: 10 sec
* 10G < row: 50 sec
* 2. cloud mode, config it as query time out
*/
private void configRuntimeFilterWaitTime() {
if (ConnectContext.get() != null && ConnectContext.get().getSessionVariable() != null
&& ConnectContext.get().getSessionVariable().getRuntimeFilterWaitTimeMs()
== VariableMgr.getDefaultSessionVariable().getRuntimeFilterWaitTimeMs()) {
SessionVariable sessionVariable = ConnectContext.get().getSessionVariable();
if (Config.isCloudMode()) {
sessionVariable.setVarOnce(SessionVariable.RUNTIME_FILTER_WAIT_TIME_MS,
String.valueOf(Math.max(VariableMgr.getDefaultSessionVariable().getRuntimeFilterWaitTimeMs(),
1000 * sessionVariable.getQueryTimeoutS())));
} else {
List<LogicalCatalogRelation> scans = cascadesContext.getRewritePlan()
.collectToList(LogicalCatalogRelation.class::isInstance);
double maxRow = StatsCalculator.getMaxTableRowCount(scans, cascadesContext);
boolean hasExternalTable = scans.stream().anyMatch(scan -> !(scan instanceof LogicalOlapScan));
if (hasExternalTable) {
if (maxRow < 1_000_000_000L) {
sessionVariable.setVarOnce(SessionVariable.RUNTIME_FILTER_WAIT_TIME_MS, "5000");
} else if (maxRow < 10_000_000_000L) {
sessionVariable.setVarOnce(SessionVariable.RUNTIME_FILTER_WAIT_TIME_MS, "20000");
} else {
sessionVariable.setVarOnce(SessionVariable.RUNTIME_FILTER_WAIT_TIME_MS, "50000");
}
} else {
if (maxRow < 1_000_000_000L) {
sessionVariable.setVarOnce(SessionVariable.RUNTIME_FILTER_WAIT_TIME_MS, "1000");
} else if (maxRow < 10_000_000_000L) {
sessionVariable.setVarOnce(SessionVariable.RUNTIME_FILTER_WAIT_TIME_MS, "5000");
} else {
sessionVariable.setVarOnce(SessionVariable.RUNTIME_FILTER_WAIT_TIME_MS, "20000");
}
}
}
}
}
private void initCascadesContext(LogicalPlan plan, PhysicalProperties requireProperties) {
cascadesContext = CascadesContext.initContext(statementContext, plan, requireProperties);
}
protected void collectAndLockTable(boolean showPlanProcess) {
collectAndLockTable(showPlanProcess, false);
}
protected void collectAndLockTable(boolean showPlanProcess, boolean waitForChangeVisible) {
if (LOG.isDebugEnabled()) {
LOG.debug("Start collect and lock table");
}
keepOrShowPlanProcess(showPlanProcess, () -> cascadesContext.newTableCollector(true, true).collect());
if (statementContext.isRowBinlogReferenceTsoRequired()) {
statementContext.getOrRegisterRowBinlogReferenceTso(
() -> MasterTsoProvider.getCurrentTso(statementContext.getConnectContext()));
}
// Read the preload result produced by the collect-phase rule before taking internal table locks.
ExternalMetadataPreloadResult preloadResult = statementContext.getExternalMetadataPreloadResult()
.orElse(ExternalMetadataPreloadResult.skipped(
statementContext.getExternalTablePreloadCandidateCount(), "preload rule did not run"));
// Record preload timing in the query profile as a dedicated planner sub-stage.
if (statementContext.getConnectContext().getExecutor() != null && preloadResult.isExecuted()) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.addNereidsPreloadExternalMetadataTime(preloadResult.getElapsedTimeMs());
}
// Keep a concise debug summary for the entire preload phase.
if (LOG.isDebugEnabled()) {
if (preloadResult.isExecuted()) {
LOG.debug("{} preloaded external metadata for {} of {} candidate tables in {} ms",
statementContext.getConnectContext().getQueryIdentifier(),
preloadResult.getPreloadedTableCount(),
preloadResult.getCandidateTableCount(),
preloadResult.getElapsedTimeMs());
} else {
LOG.debug("{} skip external metadata preload before lock: {} [candidateTableCount={}]",
statementContext.getConnectContext().getQueryIdentifier(), preloadResult.getSkipReason(),
preloadResult.getCandidateTableCount());
}
}
if (waitForChangeVisible) {
waitForTimeBasedChangeVisibleBeforeLock();
}
if (statementContext.getConnectContext().getExecutor() != null) {
// Track only the actual lock() call here so the dedicated preload stage is not double counted.
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setNereidsLockTableStartTime(TimeUtils.getStartTimeMs());
}
statementContext.lock();
cascadesContext.setCteContext(new CTEContext());
if (LOG.isDebugEnabled()) {
LOG.debug("End collect and lock table");
}
if (statementContext.getConnectContext().getExecutor() != null) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setNereidsLockTableFinishTime(TimeUtils.getStartTimeMs());
}
}
private void waitForTimeBasedChangeVisibleBeforeLock() {
try {
if (statementContext.getConnectContext().getExecutor() != null) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setWaitChangeVisibleStartTime(TimeUtils.getStartTimeMs());
}
TimeBasedChangeVisibleWaiter.waitForVisible(
statementContext.getConnectContext(),
cascadesContext.getRewritePlan(),
statementContext.getTables());
} catch (UserException e) {
throw new NereidsException(e.getMessage(), e);
}
if (statementContext.getConnectContext().getExecutor() != null) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setWaitChangeVisibleEndTime(TimeUtils.getStartTimeMs());
}
}
protected void analyze(boolean showPlanProcess) {
if (LOG.isDebugEnabled()) {
LOG.debug("Start analyze plan");
}
statementContext.getPlannerHooks().forEach(hook -> hook.beforeAnalyze(this));
keepOrShowPlanProcess(showPlanProcess, () -> cascadesContext.newAnalyzer().analyze());
statementContext.getPlannerHooks().forEach(hook -> hook.afterAnalyze(this));
if (LOG.isDebugEnabled()) {
LOG.debug("End analyze plan");
}
if (statementContext.getConnectContext().getExecutor() != null) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setNereidsAnalysisTime(TimeUtils.getStartTimeMs());
}
}
/**
* Logical plan rewrite based on a series of heuristic rules.
*/
protected void rewrite(boolean showPlanProcess) {
if (LOG.isDebugEnabled()) {
LOG.debug("Start rewrite plan");
}
keepOrShowPlanProcess(showPlanProcess, () -> {
Rewriter.getWholeTreeRewriter(cascadesContext).execute();
});
if (LOG.isDebugEnabled()) {
LOG.debug("End rewrite plan");
}
if (statementContext.getConnectContext().getExecutor() != null) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setNereidsRewriteTime(TimeUtils.getStartTimeMs());
}
statementContext.setNeedPreMvRewrite(PreMaterializedViewRewriter.needPreRewrite(cascadesContext));
// init materialization context for mv rewrite
cascadesContext.getStatementContext().getPlannerHooks().forEach(hook -> hook.afterRewrite(cascadesContext));
}
protected void preMaterializedViewRewrite() {
if (!cascadesContext.getStatementContext().isNeedPreMvRewrite()) {
return;
}
try {
if (LOG.isDebugEnabled()) {
LOG.debug("Start pre rewrite plan by mv");
}
List<Plan> tmpPlansForMvRewrite = cascadesContext.getStatementContext().getTmpPlanForMvRewrite();
Plan originalPlan = cascadesContext.getRewritePlan();
List<Plan> plansWhichContainMv = new ArrayList<>();
// because tmpPlansForMvRewrite only one, so timeout is cumulative which is ok
for (Plan planForRewrite : tmpPlansForMvRewrite) {
SessionVariable sessionVariable = cascadesContext.getConnectContext()
.getSessionVariable();
int timeoutSecond = sessionVariable.nereidsTimeoutSecond;
boolean enableTimeout = sessionVariable.enableNereidsTimeout;
try {
// set mv rewrite timeout
sessionVariable.nereidsTimeoutSecond = PreMaterializedViewRewriter.convertMillisToCeilingSeconds(
sessionVariable.materializedViewRewriteDurationThresholdMs);
sessionVariable.enableNereidsTimeout = true;
// pre rewrite
Plan rewrittenPlan = MaterializedViewUtils.rewriteByRules(cascadesContext,
PreMaterializedViewRewriter::rewrite, planForRewrite, planForRewrite, true);
Plan ruleOptimizedPlan = MaterializedViewUtils.rewriteByRules(cascadesContext,
childOptContext -> {
Rewriter.getWholeTreeRewriterWithoutCostBasedJobs(childOptContext).execute();
return childOptContext.getRewritePlan();
}, rewrittenPlan, planForRewrite, false);
if (ruleOptimizedPlan == null) {
continue;
}
// after rbo, maybe the plan changed a lot, so we need to normalize it with original plan
Plan normalizedPlan = MaterializedViewUtils.normalizeSinkExpressions(
ruleOptimizedPlan, originalPlan);
if (normalizedPlan != null) {
plansWhichContainMv.add(normalizedPlan);
}
} catch (Exception e) {
LOG.error("pre mv rewrite in rbo rewrite fail, query id is {}",
cascadesContext.getConnectContext().getQueryIdentifier(), e);
} finally {
sessionVariable.nereidsTimeoutSecond = timeoutSecond;
sessionVariable.enableNereidsTimeout = enableTimeout;
}
}
// clear the rewritten plans which are tmp optimized, should be filled by full optimize later
statementContext.getRewrittenPlansByMv().clear();
// if rule-based optimized, would not be rewritten by cbo, so clear materialized hooks
this.cascadesContext.getStatementContext().setPreMvRewritten(true);
if (!plansWhichContainMv.isEmpty()) {
plansWhichContainMv.forEach(statementContext::addRewrittenPlanByMv);
}
if (LOG.isDebugEnabled()) {
LOG.debug("End pre rewrite plan by mv");
}
} finally {
if (statementContext.getConnectContext().getExecutor() != null) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setNereidsPreRewriteByMvFinishTime(TimeUtils.getStartTimeMs());
}
}
}
// DependsRules: AddProjectForJoin
protected void optimize(boolean showPlanProcess) {
// if we cannot get table row count, skip join reorder
// except:
// 1. user set leading hint
// 2. ut test. In ut test, FeConstants.enableInternalSchemaDb is false or FeConstants.runningUnitTest is true
if (FeConstants.enableInternalSchemaDb && !FeConstants.runningUnitTest
&& !cascadesContext.isLeadingDisableJoinReorder()) {
List<CatalogRelation> scans = cascadesContext.getRewritePlan()
.collectToList(CatalogRelation.class::isInstance);
Optional<String> disableJoinReorderReason = StatsCalculator
.disableJoinReorderIfStatsInvalid(scans, cascadesContext);
disableJoinReorderReason.ifPresent(statementContext::setDisableJoinReorderReason);
}
configRuntimeFilterWaitTime();
if (LOG.isDebugEnabled()) {
LOG.debug("Start optimize plan");
}
keepOrShowPlanProcess(showPlanProcess, () -> {
new Optimizer(cascadesContext).execute();
});
if (LOG.isDebugEnabled()) {
LOG.debug("End optimize plan");
}
if (statementContext.getConnectContext().getExecutor() != null) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setNereidsOptimizeTime(TimeUtils.getStartTimeMs());
}
}
/**
* Collect plan info for hbo usage.
* @param queryId queryId
* @param root physical plan
* @param context PlanTranslatorContext
*/
private void collectHboPlanInfo(String queryId, PhysicalPlan root, PlanTranslatorContext context) {
// map of memo group id -> group, built lazily for the KEY_GROUP fallback of nodes that
// lost their group-expression back reference during post process (only needed when the
// struct-info fingerprint is in use)
Map<Integer, Group> groupsById = null;
if (cascadesContext != null && cascadesContext.getMemo() != null) {
groupsById = new HashMap<>();
for (Group group : cascadesContext.getMemo().getGroups()) {
groupsById.put(group.getGroupId().asInt(), group);
}
}
collectHboPlanInfo(queryId, root, context, groupsById);
}
private void collectHboPlanInfo(String queryId, PhysicalPlan root, PlanTranslatorContext context,
Map<Integer, Group> groupsById) {
for (Object child : root.children()) {
collectHboPlanInfo(queryId, (PhysicalPlan) child, context, groupsById);
}
if (root instanceof AbstractPlan) {
int nodeId = ((AbstractPlan) root).getId();
PlanNodeId planId = context.getNereidsIdToPlanNodeIdMap().get(nodeId);
if (planId != null) {
Map<Integer, PhysicalPlan> idToPlanMap = Env.getCurrentEnv().getHboPlanStatisticsManager()
.getHboPlanInfoProvider().getIdToPlanMap(queryId);
if (idToPlanMap.isEmpty()) {
Env.getCurrentEnv().getHboPlanStatisticsManager()
.getHboPlanInfoProvider().putIdToPlanMap(queryId, idToPlanMap);
}
idToPlanMap.put(planId.asInt(), root);
Map<PhysicalPlan, Integer> planToIdMap = Env.getCurrentEnv().getHboPlanStatisticsManager()
.getHboPlanInfoProvider().getPlanToIdMap(queryId);
if (planToIdMap.isEmpty()) {
Env.getCurrentEnv().getHboPlanStatisticsManager()
.getHboPlanInfoProvider().putPlanToIdMap(queryId, planToIdMap);
}
planToIdMap.put(root, planId.asInt());
// snapshot the hbo fingerprint (simplified group struct info) per plan node id;
// consumed by the profile publish path after the memo has been released
// publish keys are constant agnostic: join / aggregation fingerprints never
// carry literals, and scan tokens contain none by construction
Optional<String> fingerprint = GroupStructInfo.fingerprintOfPlanNode(
(AbstractPlan) root, groupsById, GroupStructInfo.LiteralMode.NO_LITERAL);
HboPlanInfoProvider planInfoProvider = Env.getCurrentEnv()
.getHboPlanStatisticsManager().getHboPlanInfoProvider();
Map<Integer, String> nodeFingerprints = planInfoProvider
.getNodeIdToFingerprintMap(queryId);
if (nodeFingerprints.isEmpty()) {
planInfoProvider.putNodeIdToFingerprintMap(queryId, nodeFingerprints);
}
// keyed by the real PlanNodeId (the same id used by the profile publish
// path via runtime stats item.node_id), not by the nereids plan id. A null value
// marks a node which is deliberately excluded from the hbo fingerprint (an
// unsupported group such as sort / topn), so that the publish path does not warn
// about a fingerprint which was never meant to exist
nodeFingerprints.put(planId.asInt(), fingerprint.orElse(null));
}
}
}
/**
* Attach the hbo fingerprint and the simplified struct info to the mutable state of a plan node,
* so that {@link #appendHboFingerprintAnnotations()} can print the injectable hbo entries of the
* plan (the node {@code toString()} itself stays free of hbo text).
* <p>For join / aggregation / filter nodes only: a filter that sits above an olap scan
* (filter-on-scan) carries the fingerprint of the scan group, which is the read-side lookup
* key constraining the filter output row count.
*/
private void attachHboExplainInfoToTree(Plan plan, Map<Integer, Group> groupsById) {
ConnectContext connectContext = ConnectContext.get();
if (connectContext == null || !connectContext.getSessionVariable().isShowHboFingerprint()) {
return;
}
for (Plan child : plan.children()) {
attachHboExplainInfoToTree(child, groupsById);
}
if (plan instanceof AbstractPlan) {
attachHboExplainInfo((AbstractPlan) plan, groupsById);
}
}
private void attachHboExplainInfo(AbstractPlan node, Map<Integer, Group> groupsById) {
boolean isFilter = node instanceof PhysicalFilter;
boolean isJoinOrAgg = node instanceof AbstractPhysicalJoin
|| node instanceof PhysicalHashAggregate
|| node instanceof PhysicalStorageLayerAggregate;
if (!isFilter && !isJoinOrAgg) {
return;
}
// filter roots have two fingerprints (exact with literals + constant agnostic shape);
// join / aggregation roots only have the constant agnostic one (their read side key)
GroupStructInfo.LiteralMode primaryMode = isFilter
? GroupStructInfo.LiteralMode.WITH_LITERAL
: GroupStructInfo.LiteralMode.NO_LITERAL;
// the printed struct info is the data state a user records for an injection, so it is
// computed now instead of being taken from the per-group cache (whose baseline is a snapshot
// of the moment the first hbo lookup of that group happened)
Optional<GroupStructInfo> structInfo = GroupStructInfo.dataStateOfPlanNode(node, groupsById, primaryMode);
Optional<GroupStructInfo> noLiteralStructInfo = structInfo;
if (isFilter) {
noLiteralStructInfo = GroupStructInfo.dataStateOfPlanNode(node, groupsById,
GroupStructInfo.LiteralMode.NO_LITERAL);
}
if (isJoinOrAgg && node instanceof AbstractPhysicalJoin) {
attachHboJoinConditionInfo(node);
}
if (structInfo.isPresent()) {
node.setMutableState(MutableState.KEY_HBO_FP, structInfo.get().getFingerprint());
if (noLiteralStructInfo.isPresent()) {
node.setMutableState(MutableState.KEY_HBO_FP_NO_LITERAL,
noLiteralStructInfo.get().getFingerprint());
}
node.setMutableState(MutableState.KEY_HBO_STRUCT, structInfo.get().getCanonicalString());
// the constant agnostic struct info is built by the traversal as well: folding the
// literal carrying form as text would mis-handle a literal whose value contains a
// parenthesis, and the printed struct= must hash to the fingerprint next to it
noLiteralStructInfo.ifPresent(noLiteral -> node.setMutableState(
MutableState.KEY_HBO_STRUCT_NO_LITERAL, noLiteral.getCanonicalString()));
} else {
// no struct info: record why, when the reason is a limit rather than an unsupported plan
// pattern, so the annotation can tell "no fingerprint by design" from "hbo is broken"
GroupStructInfo.limitSkipReasonOfPlanNode(node, groupsById)
.ifPresent(reason -> node.setMutableState(MutableState.KEY_HBO_STRUCT_SKIP, reason));
}
}
/**
* Attach the join equality-condition fingerprint (the {@code HBO SET EXPANSION} key) and the
* applied / skipped expansion state of this query, so that the physical plan and the explain
* annotation can show what to inject and whether it took effect.
*/
private void attachHboJoinConditionInfo(AbstractPlan node) {
AbstractPhysicalJoin<?, ?> join = (AbstractPhysicalJoin<?, ?>) node;
Optional<String> condFingerprint = HboJoinConditions.fingerprintOf(join);
if (!condFingerprint.isPresent()) {
return;
}
node.setMutableState(MutableState.KEY_HBO_COND_FP, condFingerprint.get());
HboJoinConditions.canonicalOf(join)
.ifPresent(canonical -> node.setMutableState(MutableState.KEY_HBO_COND, canonical));
if (ConnectContext.get() == null) {
return;
}
String queryId = DebugUtil.printId(ConnectContext.get().queryId());
HboPlanInfoProvider provider = Env.getCurrentEnv().getHboPlanStatisticsManager().getHboPlanInfoProvider();
String applied = provider.getExpansionApplied(queryId).get(condFingerprint.get());
if (applied != null) {
node.setMutableState(MutableState.KEY_HBO_EXPANSION, applied);
}
String skipped = provider.getPinnedGuardSkip(queryId).get(condFingerprint.get());
if (skipped != null && applied == null) {
node.setMutableState(MutableState.KEY_HBO_EXPANSION, "skipped=" + skipped);
}
}
protected void splitFragments(PhysicalPlan resultPlan) {
if (resultPlan instanceof PhysicalSqlCache) {
return;
}
this.planTranslatorContext = new PlanTranslatorContext(cascadesContext);
PhysicalPlanTranslator physicalPlanTranslator = new PhysicalPlanTranslator(planTranslatorContext,
statementContext.getConnectContext().getStatsErrorEstimator());
SessionVariable sessionVariable = cascadesContext.getConnectContext().getSessionVariable();
PlanFragment root = physicalPlanTranslator.translatePlan(physicalPlan);
if (statementContext.getConnectContext().getExecutor() != null) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setNereidsTranslateTime(TimeUtils.getStartTimeMs());
}
String queryId = DebugUtil.printId(cascadesContext.getConnectContext().queryId());
boolean showHboFingerprint = ConnectContext.get() != null
&& ConnectContext.get().getSessionVariable().isShowHboFingerprint();
// plan-info registration runs for learned collection, and also whenever the hbo
// fingerprint/struct info must be printed inline in the physical plan
if (StatisticsUtil.isEnableHboInfoCollection() || showHboFingerprint) {
collectHboPlanInfo(queryId, physicalPlan, planTranslatorContext);
}
scanNodeList.addAll(planTranslatorContext.getScanNodes());
physicalRelations.addAll(planTranslatorContext.getPhysicalRelations());
descTable = planTranslatorContext.getDescTable();
fragments = new ArrayList<>(planTranslatorContext.getPlanFragments());
boolean enableQueryCache = sessionVariable.getEnableQueryCache();
for (int seq = 0; seq < fragments.size(); seq++) {
PlanFragment fragment = fragments.get(seq);
fragment.setFragmentSequenceNum(seq);
if (enableQueryCache) {
try {
QueryCacheNormalizer normalizer = new QueryCacheNormalizer(fragment, descTable);
Optional<TQueryCacheParam> queryCacheParam =
normalizer.normalize(cascadesContext.getConnectContext());
if (queryCacheParam.isPresent()) {
fragment.queryCacheParam = queryCacheParam.get();
// after commons-codec 1.14 (include), Hex.encodeHexString will change ByteBuffer.pos,
// so we should copy a new byte buffer to print it
ByteBuffer digestCopy = fragment.queryCacheParam.digest.duplicate();
LOG.info("Use query cache for fragment {}, node id: {}, digest: {}, queryId: {}",
seq,
fragment.queryCacheParam.node_id,
Hex.encodeHexString(digestCopy), queryId);
}
} catch (Throwable t) {
// do nothing
}
}
}
// set output exprs
logicalPlanAdapter.setResultExprs(root.getOutputExprs());
ArrayList<String> columnLabels = Lists.newArrayListWithExpectedSize(physicalPlan.getOutput().size());
List<FieldInfo> fieldInfos = Lists.newArrayListWithExpectedSize(physicalPlan.getOutput().size());
for (NamedExpression output : physicalPlan.getOutput()) {
Optional<Column> column = Optional.empty();
Optional<TableIf> table = Optional.empty();
if (output instanceof SlotReference) {
SlotReference slotReference = (SlotReference) output;
column = slotReference.getOneLevelColumn();
table = slotReference.getOneLevelTable();
}
columnLabels.add(output.getName());
FieldInfo fieldInfo = new FieldInfo(
table.isPresent() ? (table.get().getDatabase() != null
? table.get().getDatabase().getFullName() : "") : "",
!output.getQualifier().isEmpty() ? output.getQualifier().get(output.getQualifier().size() - 1)
: (table.isPresent() ? Util.getTempTableDisplayName(table.get().getName()) : ""),
table.isPresent() ? Util.getTempTableDisplayName(table.get().getName()) : "",
output.getName(),
column.isPresent() ? column.get().getName() : ""
);
fieldInfos.add(fieldInfo);
}
logicalPlanAdapter.setColLabels(columnLabels);
logicalPlanAdapter.setFieldInfos(fieldInfos);
logicalPlanAdapter.setViewDdlSqls(statementContext.getViewDdlSqls());
if (statementContext.getSqlCacheContext().isPresent()) {
SqlCacheContext sqlCacheContext = statementContext.getSqlCacheContext().get();
sqlCacheContext.setColLabels(columnLabels);
sqlCacheContext.setFieldInfos(fieldInfos);
sqlCacheContext.setResultExprs(root.getOutputExprs());
sqlCacheContext.setPhysicalPlan(resultPlan.treeString());
}
cascadesContext.releaseMemo();
// update scan nodes visible version at the end of plan phase for cloud mode.
try {
ScanNode.setVisibleVersionForOlapScanNodes(getScanNodes());
} catch (UserException ue) {
throw new NereidsException(ue.getMessage(), ue);
}
}
protected void distribute(PhysicalPlan physicalPlan, ExplainLevel explainLevel) {
boolean canUseNereidsDistributePlanner = SessionVariable.canUseNereidsDistributePlanner()
|| (physicalPlan instanceof PhysicalDictionarySink); // dic sink only supported in new Coordinator
if ((!canUseNereidsDistributePlanner && explainLevel.isPlanLevel)) {
return;
} else if ((canUseNereidsDistributePlanner && explainLevel.isPlanLevel
&& (explainLevel != ExplainLevel.ALL_PLAN && explainLevel != ExplainLevel.DISTRIBUTED_PLAN))) {
return;
}
splitFragments(physicalPlan);
doDistribute(canUseNereidsDistributePlanner, explainLevel);
addLocalExchangeAfterDistribute();
}
private void addLocalExchangeAfterDistribute() {
SessionVariable sessionVariable = cascadesContext.getConnectContext().getSessionVariable();
if (!sessionVariable.isEnableLocalShufflePlanner() || !sessionVariable.isEnableLocalShuffle()) {
return;
}
AddLocalExchange adder = new AddLocalExchange();
if (distributedPlans != null && !distributedPlans.isEmpty()) {
adder.addLocalExchange(distributedPlans, planTranslatorContext);
}
}
protected void doDistribute(boolean canUseNereidsDistributePlanner, ExplainLevel explainLevel) {
if (!canUseNereidsDistributePlanner) {
return;
}
switch (explainLevel) {
case NONE:
case ALL_PLAN:
case DISTRIBUTED_PLAN:
break;
default: {
return;
}
}
boolean notNeedBackend = false;
// if the query can compute without backend, we can skip check cluster privileges
if (Config.isCloudMode()
&& cascadesContext.getConnectContext().supportHandleByFe()
&& physicalPlan instanceof ComputeResultSet) {
Optional<ResultSet> resultSet = ((ComputeResultSet) physicalPlan).computeResultInFe(
cascadesContext, physicalPlan.getOutput());
if (resultSet.isPresent()) {
notNeedBackend = true;
}
}
boolean useLoadBackendSelection = physicalPlan.anyMatch(PhysicalOlapTableSink.class::isInstance);
distributedPlans = new DistributePlanner(
statementContext, fragments, notNeedBackend, false, useLoadBackendSelection).plan();
if (statementContext.getConnectContext().getExecutor() != null) {
statementContext.getConnectContext().getExecutor().getSummaryProfile()
.setNereidsDistributeTime(TimeUtils.getStartTimeMs());
}
}
protected PhysicalPlan postProcess(PhysicalPlan physicalPlan) {
return new PlanPostProcessors(cascadesContext).process(physicalPlan);
}
@Override
public List<ScanNode> getScanNodes() {
return scanNodeList;
}
public List<PhysicalRelation> getPhysicalRelations() {
return physicalRelations;
}
public Group getRoot() {
return cascadesContext.getMemo().getRoot();
}
protected PhysicalPlan chooseNthPlan(Group rootGroup, PhysicalProperties physicalProperties, int nthPlan) {
if (nthPlan <= 1) {
cost = rootGroup.getLowestCostPlan(physicalProperties).orElseThrow(
() -> new AnalysisException("lowestCostPlans with physicalProperties("
+ physicalProperties + ") doesn't exist in root group")).first.getValue();
return chooseBestPlan(rootGroup, physicalProperties, cascadesContext);
}
Memo memo = cascadesContext.getMemo();
Pair<Long, Double> idCost = memo.rank(nthPlan);
cost = idCost.second;
return memo.unrank(idCost.first);
}
/**
* Doc
*/
public static PhysicalPlan chooseBestPlan(Group rootGroup, PhysicalProperties physicalProperties,
CascadesContext cascadesContext)
throws AnalysisException {
try {
GroupExpression groupExpression = rootGroup.getLowestCostPlan(physicalProperties).orElseThrow(
() -> new AnalysisException("lowestCostPlans with physicalProperties("
+ physicalProperties + ") doesn't exist in root group")).second;
if ((groupExpression.getPlan() instanceof PhysicalDistribute
&& rootGroup.getEnforcerSpecs().containsKey(
((PhysicalDistribute<?>) groupExpression.getPlan()).getDistributionSpec()))
|| rootGroup.getEnforcers().containsKey(groupExpression)) {
rootGroup.addChosenEnforcerId(groupExpression.getId().asInt());
rootGroup.addChosenEnforcerProperties(physicalProperties);
} else {
rootGroup.setChosenProperties(physicalProperties);
rootGroup.setChosenGroupExpressionId(groupExpression.getId().asInt());
}
List<PhysicalProperties> inputPropertiesList = groupExpression.getInputPropertiesList(physicalProperties);
List<Plan> planChildren = Lists.newArrayList();
for (int i = 0; i < groupExpression.arity(); i++) {
planChildren.add(chooseBestPlan(groupExpression.child(i), inputPropertiesList.get(i), cascadesContext));
}
Plan plan = groupExpression.getPlan().withChildren(planChildren);
if (!(plan instanceof PhysicalPlan)) {
// TODO need add some log
throw new AnalysisException("Result plan must be PhysicalPlan");
}
// add groupExpression to plan so that we could print group id in plan.treeString()
plan = plan.withGroupExpression(Optional.of(groupExpression));
PhysicalPlan physicalPlan = ((PhysicalPlan) plan).withPhysicalPropertiesAndStats(
physicalProperties, groupExpression.getOwnerGroup().getStatistics());
return physicalPlan;
} catch (Exception e) {
if (e instanceof AnalysisException && e.getMessage().contains("Failed to choose best plan")) {
throw e;
}
LOG.warn("Failed to choose best plan, memo structure:{}", cascadesContext.getMemo(), e);
throw new AnalysisException("Failed to choose best plan: " + e.getMessage(), e);
}
}
/**
* Memoized bottom-up DFS: compute all group IDs reachable from this group's subtree
* through physical GroupExpressions (including enforcers).
*/
private static Set<Integer> computeReachableGroupIds(Group group, Map<Group, Set<Integer>> cache) {
Set<Integer> cached = cache.get(group);
if (cached != null) {
return cached;
}
Set<Integer> reachable = new HashSet<>();
reachable.add(group.getGroupId().asInt());
// Put in cache BEFORE recursing to break cycles (enforcers reference the same group)
cache.put(group, reachable);
for (GroupExpression ge : group.getPhysicalExpressions()) {
for (Group child : ge.children()) {
reachable.addAll(computeReachableGroupIds(child, cache));
}
}
for (GroupExpression enforcer : group.getEnforcers().values()) {
for (Group child : enforcer.children()) {
reachable.addAll(computeReachableGroupIds(child, cache));
}
}
return reachable;
}
/**
* Choose the cheapest physical plan that contains all required group IDs.
* At each group, picks the cheapest physical GroupExpression whose children's
* reachable sets collectively cover all remaining required groups, then distributes
* the remaining requirements among children and recurses.
*/
private PhysicalPlan chooseBestPlanWithRequiredGroups(
Group group, PhysicalProperties properties,
Set<Integer> remaining, Map<Group, Set<Integer>> reachableCache) {
remaining = new HashSet<>(remaining);
remaining.remove(group.getGroupId().asInt());
if (remaining.isEmpty()) {
return chooseBestPlan(group, properties, cascadesContext);
}
GroupExpression chosenGE = null;
double chosenCost = Double.MAX_VALUE;
// Check all physical GEs in this group
for (GroupExpression ge : group.getPhysicalExpressions()) {
if (!ge.getLowestCostTable().containsKey(properties)) {
continue;
}
double geCost = ge.getCostByProperties(properties);
if (geCost >= chosenCost) {
continue;
}
Set<Integer> childReachable = new HashSet<>();
for (Group child : ge.children()) {
childReachable.addAll(reachableCache.getOrDefault(child, new HashSet()));
}
if (childReachable.containsAll(remaining)) {
chosenGE = ge;
chosenCost = geCost;
}
}
// Also check enforcers
for (GroupExpression enforcer : group.getEnforcers().values()) {
if (!enforcer.getLowestCostTable().containsKey(properties)) {
continue;
}
double eCost = enforcer.getCostByProperties(properties);
if (eCost >= chosenCost) {
continue;
}
Set<Integer> childReachable = new HashSet<>();
for (Group child : enforcer.children()) {
childReachable.addAll(reachableCache.getOrDefault(child, new HashSet()));
}
if (childReachable.containsAll(remaining)) {
chosenGE = enforcer;
chosenCost = eCost;
}
}
if (chosenGE == null) {
throw new AnalysisException("Cannot find physical plan containing required group IDs: "
+ remaining + " at group " + group.getGroupId());
}
// Handle enforcer marking (same as chooseBestPlan)
if ((chosenGE.getPlan() instanceof PhysicalDistribute
&& group.getEnforcerSpecs().containsKey(
((PhysicalDistribute<?>) chosenGE.getPlan()).getDistributionSpec()))
|| group.getEnforcers().containsKey(chosenGE)) {
group.addChosenEnforcerId(chosenGE.getId().asInt());
group.addChosenEnforcerProperties(properties);
} else {
group.setChosenProperties(properties);
group.setChosenGroupExpressionId(chosenGE.getId().asInt());
}
// Distribute remaining among children and recurse
List<PhysicalProperties> inputProps = chosenGE.getInputPropertiesList(properties);
List<Plan> planChildren = Lists.newArrayList();
for (int i = 0; i < chosenGE.arity(); i++) {
Group child = chosenGE.child(i);
Set<Integer> childReachable = reachableCache.getOrDefault(child, new HashSet());
Set<Integer> childRemaining = new HashSet<>(remaining);
childRemaining.retainAll(childReachable);
planChildren.add(chooseBestPlanWithRequiredGroups(
child, inputProps.get(i), childRemaining, reachableCache));
}
Plan plan = chosenGE.getPlan().withChildren(planChildren);
if (!(plan instanceof PhysicalPlan)) {
throw new AnalysisException("Result plan must be PhysicalPlan");
}
plan = plan.withGroupExpression(Optional.of(chosenGE));
PhysicalPlan physicalPlan = ((PhysicalPlan) plan).withPhysicalPropertiesAndStats(
properties, group.getStatistics());
cost = chosenCost;
return physicalPlan;
}
private long getGarbageCollectionTime() {
if (!ConnectContext.get().getSessionVariable().enableProfile()) {
return 0;
}
List<GarbageCollectorMXBean> gcMxBeans = ManagementFactory.getGarbageCollectorMXBeans();
long initialGCTime = 0;
for (GarbageCollectorMXBean gcBean : gcMxBeans) {
initialGCTime += gcBean.getCollectionTime();
}
return initialGCTime;
}
/**
* getting hints explain string, which specified by enumerate and show in lists
* @param hints hint map recorded in statement context
* @return explain string shows using of hint
*/
public String getHintExplainString(List<Hint> hints) {
String used = "";
String unUsed = "";
String syntaxError = "";
int distributeHintIndex = 0;
for (Hint hint : hints) {
String distributeIndex = "";
if (hint instanceof DistributeHint) {
distributeHintIndex++;
if (!hint.getExplainString().equals("")) {
distributeIndex = "_" + distributeHintIndex;
}
}
switch (hint.getStatus()) {
case UNUSED:
unUsed = unUsed + " " + hint.getExplainString() + distributeIndex;
break;
case SYNTAX_ERROR:
syntaxError = syntaxError + " " + hint.getExplainString() + distributeIndex
+ " Msg:" + hint.getErrorMessage();
break;
case SUCCESS:
used = used + " " + hint.getExplainString() + distributeIndex;
break;
default:
break;
}
}
return "\nHint log:" + "\nUsed:" + used + "\nUnUsed:" + unUsed + "\nSyntaxError:" + syntaxError;
}
@Override
public String getExplainString(ExplainOptions explainOptions) {
if (getConnectContext().getSessionVariable().enableExplainNone) {
return "";
}
ExplainLevel explainLevel = getExplainLevel(explainOptions);
String plan = "";
String mvSummary = "";
if ((this.getPhysicalPlan() != null || this.getOptimizedPlan() != null) && cascadesContext != null) {
mvSummary = cascadesContext.getMaterializationContexts().isEmpty() ? "" :
"\n\n========== MATERIALIZATIONS ==========\n"
+ MaterializationContext.toSummaryString(cascadesContext,
this.getPhysicalPlan() == null ? this.getOptimizedPlan() : this.getPhysicalPlan());
}
switch (explainLevel) {
case PARSED_PLAN:
plan = parsedPlan.treeString();
break;
case ANALYZED_PLAN:
plan = analyzedPlan.treeString();
break;
case REWRITTEN_PLAN:
plan = rewrittenPlan.treeString();
break;
case OPTIMIZED_PLAN:
plan = "cost = " + cost + "\n" + optimizedPlan.treeString() + mvSummary;
break;
case SHAPE_PLAN:
plan = optimizedPlan.shape("");
break;
case MEMO_PLAN:
Memo memo = cascadesContext.getMemo();
if (memo == null) {
plan = "Memo is null";
} else {
plan = memo.toString()
+ "\n\n========== OPTIMIZED PLAN ==========\n"
+ optimizedPlan.treeString()
+ mvSummary;
}
break;
case DISTRIBUTED_PLAN:
StringBuilder distributedPlanStringBuilder = new StringBuilder();
distributedPlanStringBuilder.append("========== DISTRIBUTED PLAN ==========\n");
if (distributedPlans == null || distributedPlans.isEmpty()) {
plan = "Distributed plan not generated, please set enable_nereids_distribute_planner "
+ "and enable_pipeline_x_engine to true";
} else {
plan += DistributedPlan.toString(Lists.newArrayList(distributedPlans.values())) + "\n\n";
}
break;
case ALL_PLAN:
plan = "========== PARSED PLAN "
+ getTimeMetricString(SummaryProfile::getPrettyParseSqlTime) + " ==========\n"
+ parsedPlan.treeString() + "\n\n"
+ "========== LOCK TABLE "
+ getTimeMetricString(SummaryProfile::getPrettyNereidsLockTableTime) + " ==========\n"
+ "\n\n"
+ "========== ANALYZED PLAN "
+ getTimeMetricString(SummaryProfile::getPrettyNereidsAnalysisTime) + " ==========\n"
+ analyzedPlan.treeString() + "\n\n"
+ "========== REWRITTEN PLAN "
+ getTimeMetricString(SummaryProfile::getPrettyNereidsRewriteTime) + " ==========\n"
+ rewrittenPlan.treeString() + "\n\n"
+ "========== PRE REWRITTEN BY MV "
+ getTimeMetricString(SummaryProfile::getPrettyNereidsPreRewriteByMvTime) + " ==========\n"
+ "========== OPTIMIZED PLAN "
+ getTimeMetricString(SummaryProfile::getPrettyNereidsOptimizeTime) + " ==========\n"
+ optimizedPlan.treeString() + "\n\n";
if (cascadesContext != null && cascadesContext.getMemo() != null) {
plan += "========== MEMO " + cascadesContext.getMemo().toString() + "\n\n";
}
if (distributedPlans != null && !distributedPlans.isEmpty()) {
plan += "========== DISTRIBUTED PLAN "
+ getTimeMetricString(SummaryProfile::getPrettyNereidsDistributeTime) + " ==========\n";
plan += DistributedPlan.toString(Lists.newArrayList(distributedPlans.values())) + "\n\n";
}
plan += mvSummary;
break;
default:
plan = super.getExplainString(explainOptions);
plan += mvSummary;
plan += "\n\n\n========== STATISTICS ==========\n";
if (statementContext != null) {
if (statementContext.isHasUnknownColStats()) {
plan += "planned with unknown column statistics\n";
}
}
}
if (ConnectContext.get() != null && cascadesContext != null
&& cascadesContext.getGroupExpressionCount()
> ConnectContext.get().getSessionVariable().memoMaxGroupExpressionSize) {
plan += "\n\n\n group expression count exceeds memo_max_group_expression_size("
+ ConnectContext.get().getSessionVariable().memoMaxGroupExpressionSize + ")\n";
}
String hint = "";
if (statementContext != null && !statementContext.getHints().isEmpty()) {
hint = getHintExplainString(statementContext.getHints());
}
// the annotation block is meaningful only once a physical plan exists (fragment-form
// explain); parsed/analyzed/rewritten tree output must not carry the misleading
// "(no hbo fingerprint attached ...)" fallback
if (explainLevel != ExplainLevel.PARSED_PLAN
&& explainLevel != ExplainLevel.ANALYZED_PLAN
&& explainLevel != ExplainLevel.REWRITTEN_PLAN
&& ConnectContext.get() != null
&& ConnectContext.get().getSessionVariable().isShowHboFingerprint()
&& physicalPlan != null && cascadesContext != null) {
plan += appendHboFingerprintAnnotations();
}
return plan + hint;
}
/**
* Append per-node hbo fingerprint annotations (join / aggregation / filter) to the explain
* string. Controlled by session variable {@code show_hbo_fingerprint} (default off) so that
* the default explain output is unchanged. For a filter that sits above an olap scan
* (filter-on-scan) the annotation shows the fingerprint of the scan group, which is the key
* used by the hbo read side to override the filter row count ��� users copy the quoted
* {@code fingerprint=} / {@code struct=} (or {@code condFingerprint=} / {@code cond=} for a
* join expansion) straight into the {@code HBO SET STATISTICS} statement.
*/
private String appendHboFingerprintAnnotations() {
StringBuilder sb = new StringBuilder("\n\nHBO fingerprint annotations (join/aggregation/filter):\n");
if (ConnectContext.get() != null && !ConnectContext.get().getSessionVariable().isEnableHboOptimization()) {
sb.append(" note: enable_hbo_optimization is off; injected hbo statistics will not take "
+ "effect until it is enabled\n");
}
List<AbstractPlan> nodes = new ArrayList<>();
collectPlanNodes(physicalPlan, nodes);
nodes.sort((a, b) -> Integer.compare(a.getId(), b.getId()));
Map<String, String> guardSkips = Collections.emptyMap();
Map<String, String> applyStates = Collections.emptyMap();
Map<String, String> appliedModes = Collections.emptyMap();
if (ConnectContext.get() != null) {
HboPlanInfoProvider provider = Env.getCurrentEnv().getHboPlanStatisticsManager()
.getHboPlanInfoProvider();
String queryId = DebugUtil.printId(ConnectContext.get().queryId());
guardSkips = provider.getPinnedGuardSkip(queryId);
applyStates = provider.getPinnedApplyState(queryId);
appliedModes = provider.getPinnedLiteralMode(queryId);
}
for (AbstractPlan node : nodes) {
String kind;
if (node instanceof AbstractPhysicalJoin) {
kind = "join";
} else if (node instanceof PhysicalHashAggregate || node instanceof PhysicalStorageLayerAggregate) {
kind = "aggregation";
} else if (node instanceof PhysicalFilter) {
// a filter root is injectable: its primary fingerprint carries the literals, and
// the constant agnostic shape fingerprint is printed next to it; filters directly
// above a scan additionally have a learned read-side key derived from the scan
AbstractPlan scan = findScanUnder((PhysicalFilter<?>) node);
kind = scan == null ? "filter" : "filter-on-scan(table=" + scanName(scan) + ")";
} else {
continue;
}
Object fingerprint = node.getMutableState(MutableState.KEY_HBO_FP).orElse(null);
Object struct = node.getMutableState(MutableState.KEY_HBO_STRUCT).orElse(null);
if (fingerprint == null || struct == null) {
Object skip = node.getMutableState(MutableState.KEY_HBO_STRUCT_SKIP).orElse(null);
if (skip != null) {
// a node whose sub tree is too wide for hbo: it has no fingerprint by design, and
// saying so is the difference between "hbo is off for this shape" and "hbo broke"
sb.append(" [").append(node.getId()).append("] ").append(kind)
.append(" hbo fingerprint skipped: ").append(skip).append('\n');
}
continue;
}
Object expansion = node.getMutableState(MutableState.KEY_HBO_EXPANSION).orElse(null);
Object condFingerprint = node.getMutableState(MutableState.KEY_HBO_COND_FP).orElse(null);
Object cond = node.getMutableState(MutableState.KEY_HBO_COND).orElse(null);
// the type of the row count entry is read from the entry of that line's own fingerprint
// (EXACT unless a guarded entry was injected), so the two lines of a filter root can not
// report each other's type
if (node instanceof PhysicalFilter) {
// a filter root can be keyed in both literal modes: the literal carrying form (only
// that constant matches) and the constant agnostic form (every constant of the
// predicate shape matches), so both are printed as separate injectable entries
appendHboEntryLine(sb, node.getId(), kind, "with_literal",
hboEntryTypeOf(String.valueOf(fingerprint)), fingerprint, struct,
guardSkips, applyStates, appliedModes, expansion, false);
Object noLiteralFingerprint = node.getMutableState(MutableState.KEY_HBO_FP_NO_LITERAL).orElse(null);
Object noLiteralStruct = node.getMutableState(MutableState.KEY_HBO_STRUCT_NO_LITERAL).orElse(null);
if (noLiteralFingerprint != null && noLiteralStruct != null
&& !noLiteralFingerprint.equals(fingerprint)) {
appendHboEntryLine(sb, node.getId(), kind, "no_literal",
hboEntryTypeOf(String.valueOf(noLiteralFingerprint)), noLiteralFingerprint,
noLiteralStruct, guardSkips, applyStates, appliedModes, expansion, false);
}
} else {
// a join / aggregation entry only has the constant agnostic form
appendHboEntryLine(sb, node.getId(), kind, "no_literal",
hboEntryTypeOf(String.valueOf(fingerprint)), fingerprint, struct,
guardSkips, applyStates, appliedModes, expansion, false);
}
if (condFingerprint != null && cond != null) {
// the join condition key of this node: an expansion entry, kept in the same table and
// listed like any other entry of this node
appendHboEntryLine(sb, node.getId(), kind, "no_literal", "join_expansion", condFingerprint,
cond, guardSkips, applyStates, appliedModes, expansion, true);
}
}
if (sb.toString().indexOf("fingerprint=") < 0) {
sb.append(" (no hbo fingerprint attached; check that the plan went through the "
+ "planner attach step)\n");
}
return sb.toString();
}
/**
* Append one injectable hbo entry of a node. The parameters are printed with the spelling
* {@code HBO SET STATISTICS} uses (quoted values), so the line can be copied as it is, and the
* line of the entry which was actually used (or skipped by its guard) is marked.
*/
private void appendHboEntryLine(StringBuilder sb, int nodeId, String kind, String literalMode,
String type, Object fingerprint, Object struct, Map<String, String> guardSkips,
Map<String, String> applyStates, Map<String, String> appliedModes, Object expansion,
boolean expansionLine) {
String key = String.valueOf(fingerprint);
sb.append(" [").append(nodeId).append("] ").append(kind)
.append(" type=").append(type)
.append(" literal_mode=").append(literalMode)
.append(" fingerprint='").append(key).append("'")
.append(" struct='").append(struct).append("'");
// the entry was rejected (its data moved too far, or its guard failed), or it was applied
// with the data state the read side saw (live / drifted / unknown); a learned entry has no
// state of its own and is only reported as used
String skipReason = guardSkips.get(key);
String applyState = applyStates.get(key);
if (skipReason != null) {
sb.append(" skipped=").append(skipReason);
} else if (applyState != null) {
sb.append(" used=").append(applyState);
} else if (literalMode.equals(appliedModes.get(key))) {
sb.append(" used=true");
}
if (expansion != null && expansionLine) {
sb.append(" expansion=").append(expansion);
}
sb.append("\n");
}
/**
* The type of the pinned row count entry of a fingerprint: EXACT, unless the user injected a
* guarded ({@code FILTER_SMALL}) entry for it. A fingerprint without an entry is listed with the
* default type, which is what an injection for it would use.
*/
private static String hboEntryTypeOf(String fingerprint) {
return Env.getCurrentEnv().getHboPlanStatisticsManager().getPinnedPlanStatistics(fingerprint)
.map(pinned -> pinned.getType().name().toLowerCase(java.util.Locale.ROOT))
.orElse("exact");
}
private static void collectPlanNodes(Plan plan, List<AbstractPlan> nodes) {
for (Plan child : plan.children()) {
collectPlanNodes(child, nodes);
}
if (plan instanceof AbstractPlan) {
nodes.add((AbstractPlan) plan);
}
}
/**
* Find the olap scan directly below a filter, only through a chain of projects. A filter
* that is not directly on a scan (e.g. above a join) is not a "filter-on-scan" and must not
* borrow the fingerprint of an inner scan.
*/
private static AbstractPlan findScanUnder(PhysicalFilter<?> filter) {
return scanBelowProjects(filter.child());
}
private static AbstractPlan scanBelowProjects(Plan plan) {
if (plan instanceof PhysicalOlapScan) {
return (AbstractPlan) plan;
}
if (plan instanceof PhysicalProject && plan.arity() == 1) {
return scanBelowProjects(plan.child(0));
}
return null;
}
private static String scanName(AbstractPlan scan) {
if (scan instanceof PhysicalOlapScan) {
return ((PhysicalOlapScan) scan).getTable().getNameWithFullQualifiers();
}
return String.valueOf(scan);
}
@Override
public DescriptorTable getDescTable() {
return descTable;
}
@Override
public void appendTupleInfo(StringBuilder str) {
str.append(descTable.getExplainString());
}
@Override
public List<RuntimeFilter> getRuntimeFilters() {
return new ArrayList<>(cascadesContext.getRuntimeFilterContext().getLegacyFilters());
}
@Override
public Optional<ResultSet> handleQueryInFe(StatementBase parsedStmt) {
if (!(parsedStmt instanceof LogicalPlanAdapter)) {
return Optional.empty();
}
setFormatOptions();
if (physicalPlan instanceof ComputeResultSet) {
Optional<ResultSet> resultSet = ((ComputeResultSet) physicalPlan)
.computeResultInFe(cascadesContext, physicalPlan.getOutput());
if (resultSet.isPresent()) {
tryAddResultSetToSqlCache(resultSet.get());
return resultSet;
}
}
return Optional.empty();
}
private void tryAddResultSetToSqlCache(ResultSet resultSet) {
if (physicalPlan instanceof PhysicalSqlCache) {
return;
}
Optional<SqlCacheContext> sqlCacheContext = statementContext.getSqlCacheContext();
if (!sqlCacheContext.isPresent()
|| !CacheAnalyzer.canUseSqlCache(statementContext.getConnectContext().getSessionVariable())) {
return;
}
sqlCacheContext.get().setResultSetInFe(resultSet);
Env.getCurrentEnv().getSqlCacheManager().tryAddFeSqlCache(
statementContext.getConnectContext(),
statementContext.getOriginStatement().originStmt);
}
private void setFormatOptions() {
ConnectContext ctx = statementContext.getConnectContext();
SessionVariable sessionVariable = ctx.getSessionVariable();
switch (sessionVariable.serdeDialect) {
case "presto":
case "trino":
statementContext.setFormatOptions(FormatOptions.getForPresto());
break;
case "hive":
statementContext.setFormatOptions(FormatOptions.getForHive());
break;
case "doris":
statementContext.setFormatOptions(FormatOptions.getDefault());
break;
default:
throw new AnalysisException("Unsupported serde dialect: " + sessionVariable.serdeDialect);
}
}
@VisibleForTesting
public CascadesContext getCascadesContext() {
return cascadesContext;
}
public ConnectContext getConnectContext() {
return cascadesContext == null ? ConnectContext.get() : cascadesContext.getConnectContext();
}
public StatementContext getStatementContext() {
return statementContext;
}
public static PhysicalProperties buildInitRequireProperties() {
return PhysicalProperties.GATHER;
}
protected ExplainLevel getExplainLevel(ExplainOptions explainOptions) {
if (explainOptions == null) {
return ExplainLevel.NONE;
}
ExplainLevel explainLevel = explainOptions.getExplainLevel();
return explainLevel == null ? ExplainLevel.NONE : explainLevel;
}
@VisibleForTesting
public Plan getParsedPlan() {
return parsedPlan;
}
@VisibleForTesting
public void setParsedPlan(Plan parsedPlan) {
this.parsedPlan = parsedPlan;
}
@VisibleForTesting
public void setOptimizedPlan(Plan optimizedPlan) {
this.optimizedPlan = optimizedPlan;
}
public void setAnalyzedPlan(Plan analyzedPlan) {
this.analyzedPlan = analyzedPlan;
}
@VisibleForTesting
public Plan getAnalyzedPlan() {
return analyzedPlan;
}
@VisibleForTesting
public Plan getRewrittenPlan() {
return rewrittenPlan;
}
@VisibleForTesting
public Plan getOptimizedPlan() {
return optimizedPlan;
}
public PhysicalPlan getPhysicalPlan() {
return physicalPlan;
}
public FragmentIdMapping<DistributedPlan> getDistributedPlans() {
return distributedPlans;
}
public LogicalPlanAdapter getLogicalPlanAdapter() {
return logicalPlanAdapter;
}
private String getTimeMetricString(Function<SummaryProfile, String> profileSupplier) {
return getProfile(summaryProfile -> {
String metricString = profileSupplier.apply(summaryProfile);
return (metricString == null || "N/A".equals(metricString)) ? "" : "(time: " + metricString + ")";
}, "");
}
private <T> T getProfile(Function<SummaryProfile, T> profileSupplier, T defaultMetric) {
T metric = null;
if (statementContext.getConnectContext().getExecutor() != null) {
SummaryProfile summaryProfile = statementContext.getConnectContext().getExecutor().getSummaryProfile();
if (summaryProfile != null) {
metric = profileSupplier.apply(summaryProfile);
}
}
return metric == null ? defaultMetric : metric;
}
private boolean showAnalyzeProcess(ExplainLevel explainLevel, boolean showPlanProcess) {
return showPlanProcess
&& (explainLevel == ExplainLevel.ANALYZED_PLAN || explainLevel == ExplainLevel.ALL_PLAN);
}
private boolean showRewriteProcess(ExplainLevel explainLevel, boolean showPlanProcess) {
return showPlanProcess
&& (explainLevel == ExplainLevel.REWRITTEN_PLAN || explainLevel == ExplainLevel.ALL_PLAN);
}
private boolean showPlanProcess(ExplainOptions explainOptions) {
return explainOptions != null && explainOptions.showPlanProcess();
}
protected void keepOrShowPlanProcess(boolean showPlanProcess, Runnable task) {
if (showPlanProcess) {
cascadesContext.withPlanProcess(showPlanProcess, task);
} else {
task.run();
}
}
@Override
public List<TopnFilter> getTopnFilters() {
return cascadesContext.getTopnFilterContext().getTopnFilters();
}
}