englefly commented on code in PR #67549: URL: https://github.com/apache/doris/pull/67549#discussion_r4240221401
########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/GroupStructInfo.java: ########## @@ -0,0 +1,905 @@ +// 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.stats; + +import org.apache.doris.nereids.memo.Group; +import org.apache.doris.nereids.memo.GroupExpression; +import org.apache.doris.nereids.trees.expressions.And; +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.Or; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.plans.AbstractPlan; +import org.apache.doris.nereids.trees.plans.JoinType; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalJoin; +import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.util.MutableState; +import org.apache.doris.qe.ConnectContext; + +import com.google.common.hash.Hashing; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeSet; +import java.util.stream.Collectors; + +/** + * Simplified struct info of a memo {@link Group}, used by History Based Optimization (HBO). + * + * <p>The struct info is a property of the <b>group</b>, not of one of its group expressions: all + * expressions of a group are logically equivalent, so they describe the same set of relations with + * the same predicates, and the canonical form below is deliberately insensitive to which expression + * is picked. + * <ul> + * <li>join commutativity: children of a join are canonicalized by sorting;</li> + * <li>join associativity: a whole inner / cross join chain is emitted as one node + * ({@code J{inner,c:[...]}(...;...)}) whose children are the chain leaves sorted by their + * canonical form and whose conditions are the sorted union of every condition of the chain, + * so {@code (A join B) join C} and {@code A join (B join C)} produce the same descriptor;</li> + * <li>aggregation: only the group by keys and the child participate, because the output row count + * of an aggregation is the number of groups and cannot be changed by the aggregate functions + * or by the output expressions;</li> + * <li>project is transparent for structure matching.</li> + * </ul> + * This mirrors what the MV oriented {@code rules.exploration.mv.StructInfo} does (it identifies a + * sub tree by its relation set and by level independent - shuttled - predicates), with every MV + * specific field dropped: no hyper graph, no shuttle map, no equivalence class, no lineage. + * + * <p>What the canonical form keeps: + * <ul> + * <li>for each scan: the table qualifier and - as an annotation only - the data state of the scan + * (its visible version, the rows it reads and how many of the table partitions it selects, + * see {@link HboScanDescriptor});</li> + * <li>for filter / join / aggregate: the normalized predicates / join conditions / grouping keys, + * i.e. the structural pattern only (slot and expression ids are stripped by + * {@link #normalizeExpression}).</li> + * </ul> + * The string contains no per-query identifier, so it is reproducible across runs and queries for + * structurally identical sub trees, and its sha256 is used as the HBO cache key (fingerprint). + * + * <p>The sha256 is taken over the canonical string <b>without</b> the scan baselines + * ({@link #stripScanBaseline}): a hbo key identifies a plan pattern and the relation it reads, not a + * data state, so an entry whose table simply keeps growing stays matchable instead of silently never + * matching again. Whether an entry may still be applied is decided on the read side, by comparing + * the baseline of the entry with the data state of the current query (see + * {@link HboStructFreshness}). + * + * <p>The descriptor is derived from the group's logical expression and its child groups (memo level + * traversal, with a visited set to guard shared sub graphs such as CTE). Groups whose content is not + * supported (e.g. contains TVF / CTE consumer, or the head operator is not one of the supported + * kinds) are {@link #isValid() invalid} and callers must fall back to the legacy behavior. + */ +public class GroupStructInfo { + private static final List<HboScanDescriptor> INVALID_SCANS = Collections.emptyList(); + + /** Shared invalid instance: the group content is not supported by the simplified struct info. */ + public static final GroupStructInfo INVALID = new GroupStructInfo(false, "", "", "", INVALID_SCANS); + /** + * Shared instance for a failure which is not a property of the plan: the table state could not be + * read (e.g. the visible version of a cloud table could not be fetched). Unlike + * {@link #INVALID} it is not cached by the memo (see {@code Group#getOrComputeHboStructInfo}), so + * the next lookup of the same group retries instead of disabling hbo struct info for the whole + * query. + */ + public static final GroupStructInfo TRANSIENT_FAILURE = + new GroupStructInfo(false, "", "", "", INVALID_SCANS); + + private static final String SEP = ";"; + private static final Logger LOG = LogManager.getLogger(GroupStructInfo.class); + + /** + * Whether literal values participate in the canonical string / fingerprint. + * <ul> + * <li>{@link #WITH_LITERAL} keeps the literal value ({@code lit(10:int)}): the fingerprint is + * bound to the exact constants, used for filter roots where the constant changes the + * output row count;</li> + * <li>{@link #NO_LITERAL} replaces every literal value with {@code *} ({@code lit(*)}): the + * fingerprint is constant agnostic, used for join / aggregate roots and for the + * constant-free filter shape.</li> + * </ul> + * The literal data type is intentionally not kept: nereids inserts an explicit {@code Cast} + * for mismatching operand types, so two semantically different literal forms already differ by + * the cast node, while equivalent forms (e.g. {@code d > 1} and {@code d > 1.0} on a decimal + * column) are allowed to collapse. + */ + public enum LiteralMode { + WITH_LITERAL, + NO_LITERAL + } + + private final boolean valid; + private final String canonicalString; + /** The canonical string without the scan baselines: the fingerprint input. */ + private final String shapeString; + private final String fingerprint; + /** The scans of the sub tree, in the order their tokens appear in the canonical string. */ + private final List<HboScanDescriptor> scans; + /** Why no struct info was built, when it was refused by a limit instead of being unsupported. */ + private final String skipReason; + + private GroupStructInfo(boolean valid, String canonicalString, String shapeString, String fingerprint, + List<HboScanDescriptor> scans) { + this(valid, canonicalString, shapeString, fingerprint, scans, ""); + } + + private GroupStructInfo(boolean valid, String canonicalString, String shapeString, String fingerprint, + List<HboScanDescriptor> scans, String skipReason) { + this.valid = valid; + this.canonicalString = canonicalString; + this.shapeString = shapeString; + this.fingerprint = fingerprint; + this.scans = scans; + this.skipReason = skipReason; + } + + /** The reason the struct info was refused by a limit, or empty. Only for diagnostics. */ + public String getSkipReason() { + return skipReason; + } + + public boolean isValid() { + return valid; + } + + public String getCanonicalString() { + return canonicalString; + } + + /** + * The fingerprint input: the canonical string with the baseline of every scan token removed + * ({@code S{db.t}}). It is built from the traversal itself and not by parsing the canonical + * string, so no text inside a table name or a literal can be mistaken for a scan token. + */ + public String getShapeString() { + return shapeString; + } + + /** + * The row count and partition selection of every scan of the sub tree, in the order the scan + * tokens appear in the canonical string (i.e. the order a user sees when pasting the struct + * info). Empty when the struct info is invalid. + */ + public List<HboScanDescriptor> getScans() { + return scans; + } + + public String getFingerprint() { + return fingerprint; + } + + /** + * Resolve the fingerprint of the memo group that {@code planNode} belongs to, when the node + * still carries its {@link GroupExpression} back reference (memo inner plans, or plans from + * {@code chooseBestPlan}). Empty when the group content is unsupported (invalid struct info). + */ + public static Optional<String> fingerprintOfPlanNode(AbstractPlan planNode) { + return fingerprintOfPlanNode(planNode, LiteralMode.WITH_LITERAL); + } + + /** + * Resolve the group fingerprint of a plan node in the given literal mode. + */ + public static Optional<String> fingerprintOfPlanNode(AbstractPlan planNode, LiteralMode mode) { + return fingerprintOfGroup(planNode.getGroupExpression() + .map(GroupExpression::getOwnerGroup).orElse(null), mode); + } + + /** + * Resolve the fingerprint of the group that {@code planNode} belongs to, with a fallback to + * the {@link MutableState#KEY_GROUP} group-id state that post processors propagate to their + * rewritten copies (see {@code copyStatsAndGroupIdFrom}). {@code groupsById} maps memo group + * id to group, and is looked up only when the back reference is absent. + */ + public static Optional<String> fingerprintOfPlanNode(AbstractPlan planNode, Map<Integer, Group> groupsById) { + return fingerprintOfPlanNode(planNode, groupsById, LiteralMode.WITH_LITERAL); + } + + /** + * Resolve the group fingerprint of a plan node in the given literal mode, with the + * {@link MutableState#KEY_GROUP} fallback for post-processed copies. + */ + public static Optional<String> fingerprintOfPlanNode(AbstractPlan planNode, Map<Integer, Group> groupsById, + LiteralMode mode) { + return structInfoOfPlanNode(planNode, groupsById, mode).map(GroupStructInfo::getFingerprint); + } + + private static Optional<String> fingerprintOfGroup(Group group, LiteralMode mode) { + return structInfoOfGroup(group, mode).map(GroupStructInfo::getFingerprint); + } + + /** + * Resolve the {@link GroupStructInfo} (with canonical string and fingerprint) of the group a + * plan node belongs to, using the group-expression back reference or the KEY_GROUP group-id + * state propagated by post processors (see {@link #fingerprintOfPlanNode}). + */ + public static Optional<GroupStructInfo> structInfoOfPlanNode(AbstractPlan planNode, + Map<Integer, Group> groupsById) { + return structInfoOfPlanNode(planNode, groupsById, LiteralMode.WITH_LITERAL); + } + + /** + * Resolve the struct info of the group a plan node belongs to, in the given literal mode. + */ + public static Optional<GroupStructInfo> structInfoOfPlanNode(AbstractPlan planNode, + Map<Integer, Group> groupsById, LiteralMode mode) { + Group group = planNode.getGroupExpression().map(GroupExpression::getOwnerGroup).orElse(null); + if (group == null) { + Optional<Object> groupState = planNode.getMutableState(MutableState.KEY_GROUP); + if (groupState.isPresent() && groupsById != null) { + try { + group = groupsById.get(Integer.valueOf(groupState.get().toString())); + } catch (NumberFormatException ignored) { + group = null; + } + } + } + return structInfoOfGroup(group, mode); + } + + private static Optional<GroupStructInfo> structInfoOfGroup(Group group, LiteralMode mode) { + if (group == null) { + return Optional.empty(); + } + GroupStructInfo structInfo = group.getOrComputeHboStructInfo(mode); + return structInfo.isValid() ? Optional.of(structInfo) : Optional.empty(); + } + + /** + * The struct info of the group a plan node belongs to, computed now instead of being taken from + * the per-group cache. The fingerprint does not depend on any data state, so the cached struct + * info is the right source for a lookup; its <b>baseline</b> (the visible version and the row + * counts of its scans) is however a snapshot of the moment it was computed, which may be well + * before the query is planned. Whoever displays that baseline (EXPLAIN prints it as the + * {@code struct=} a user pastes into {@code HBO SET STATISTICS}) or compares it (the drift + * verdict) must therefore ask for it here, so that the whole query reports one data state. + */ + public static Optional<GroupStructInfo> dataStateOfPlanNode(AbstractPlan planNode, + Map<Integer, Group> groupsById, LiteralMode mode) { + Group group = planNode.getGroupExpression().map(GroupExpression::getOwnerGroup).orElse(null); + if (group == null) { + Optional<Object> groupState = planNode.getMutableState(MutableState.KEY_GROUP); + if (groupState.isPresent() && groupsById != null) { + try { + group = groupsById.get(Integer.valueOf(groupState.get().toString())); + } catch (NumberFormatException ignored) { + group = null; + } + } + } + if (group == null) { + return Optional.empty(); + } + GroupStructInfo structInfo = of(group, mode); + return structInfo.isValid() ? Optional.of(structInfo) : Optional.empty(); + } + + /** + * Compute the simplified struct info (and its fingerprint) of a memo group, by traversing the + * group's logical expression and its child groups. + */ + public static GroupStructInfo of(Group group) { + return of(group, LiteralMode.WITH_LITERAL); + } + + /** + * Compute the simplified struct info of a memo group in the given literal mode. + */ + public static GroupStructInfo of(Group group, LiteralMode mode) { + HboStructSummary summary = group.getOrComputeHboStructSummary(); + int maxScansPerGroup = hboMaxScansPerGroup(); + if (maxScansPerGroup > 0 && summary.getScanCount() > maxScansPerGroup) { + // The canonical string of a group describes its whole sub tree, and for a chain of joins + // it is built once per group of that chain, so a cheap structural summary is used as a + // gate: a sub tree with more scan tokens than hbo_max_scans_per_group is not described at + // all (the read side falls back to the optimizer estimation and EXPLAIN reports why). + return new GroupStructInfo(false, "", "", "", INVALID_SCANS, "subtree scans=" + + summary.getScanCount() + " > hbo_max_scans_per_group=" + maxScansPerGroup); + } + Ctx ctx = new Ctx(mode); + try { + Canonical out = new Canonical(); + visit(group, out, ctx); + if (!ctx.valid) { + // a failure which came from reading the catalog (e.g. a cloud rpc error) may survive + // a later attempt: INVALID would be cached by the memo group and disable hbo for the + // whole query, TRANSIENT_FAILURE is retried instead + return ctx.retryable ? TRANSIENT_FAILURE : INVALID; + } + String canonicalString = out.annotated.toString(); + String shapeString = out.shape.toString(); + String fingerprint = Hashing.sha256() + .hashString(shapeString, StandardCharsets.UTF_8).toString(); + return new GroupStructInfo(true, canonicalString, shapeString, fingerprint, + new ArrayList<>(out.getScans())); + } catch (RuntimeException e) { + // memo content not supported by the simplified struct info: treat as invalid and + // fall back to legacy behavior instead of failing the optimizer on the hot path + LOG.debug("failed to compute hbo struct info for group {}", group.getGroupId(), e); + return ctx.retryable ? TRANSIENT_FAILURE : INVALID; + } + } + + /** + * Visit a group and append its canonical description to {@code sb}. + */ + private static void visit(Group group, Canonical out, Ctx ctx) { + if (!ctx.valid) { + return; + } + if (!ctx.visited.add(group)) { + // shared sub graph (e.g. CTE / repeated child group): cannot be expressed by a single + // canonical tree, mark invalid so that callers fall back to legacy behavior + invalid(ctx); + return; + } + // Use the first logical expression to know the kind of the group head. The canonical forms + // below are insensitive to which expression is picked, so this choice cannot change the + // fingerprint of the group (see the class comment). + GroupExpression ge = group.getFirstLogicalExpression(); + if (ge == null) { + invalid(ctx); + return; + } + appendPlan(ge.getPlan(), ge, out, ctx); + } + + private static void appendPlan(Plan plan, GroupExpression ge, Canonical out, Ctx ctx) { + if (!ctx.valid) { + return; + } + if (plan instanceof LogicalOlapScan) { + appendScan((LogicalOlapScan) plan, out, ctx); + } else if (plan instanceof LogicalFilter) { + LogicalFilter<?> filter = (LogicalFilter<?>) plan; + out.append("F{").append(normalizedSorted(filter.getConjuncts(), ctx.mode)).append("}("); + appendChild(ge, 0, out, ctx); + out.append(")"); + } else if (plan instanceof LogicalProject) { + // project is transparent for structure matching + if (ge.arity() == 1) { + appendChild(ge, 0, out, ctx); + } else { + invalid(ctx); + } + } else if (plan instanceof LogicalJoin) { + appendJoin((LogicalJoin<?, ?>) plan, ge, out, ctx); + } else if (plan instanceof LogicalAggregate) { + appendAggregate((LogicalAggregate<?>) plan, ge, out, ctx); + } else { + // unsupported head operator (sort/topn/limit/window/union/cte/tvf/...) + invalid(ctx); + } + } + + private static void appendJoin(LogicalJoin<?, ?> join, GroupExpression ge, Canonical out, Ctx ctx) { + if (isFlattenableJoinType(join.getJoinType())) { + appendJoinChain(join, ge, out, ctx); + return; + } + // Outer / semi / anti / asof joins: the two sides are semantically different and the join + // cannot be reassociated, so the memo order of the children and the join type are kept. + out.append("J{").append(join.getJoinType().toString()); + out.append(",c:[").append(normalizedJoinConditions(join, ctx.mode)).append("]}("); + appendChild(ge, 0, out, ctx); + out.append(SEP); + appendChild(ge, 1, out, ctx); + out.append(")"); + } + + /** + * Append the canonical form of a whole inner / cross join chain as a single node. + * + * <p>Inner and cross joins are associative and commutative, so the grouping of the chain is an + * artifact of the explored expression and must not be part of the fingerprint: the chain is + * emitted as {@code J{inner,c:[...]}(leaf;leaf;...)} with the conditions of every join of the + * chain merged into one sorted set and the leaves (scans, filters, aggregations, non + * flattenable joins) sorted by their canonical form. Every expression of a group therefore + * produces the same descriptor. Cross joins are merged with inner joins because a cross join is + * exactly an inner join without condition, and both have the same output row count. + */ + private static void appendJoinChain(LogicalJoin<?, ?> join, GroupExpression ge, Canonical out, Ctx ctx) { + TreeSet<String> conditions = new TreeSet<>(); + List<Leaf> leaves = new ArrayList<>(); + collectJoinConditions(join, conditions, ctx.mode); + for (int i = 0; i < ge.arity(); i++) { + collectJoinChain(ge.child(i), conditions, leaves, ctx); + } + if (!ctx.valid) { + return; + } + // the leaves are ordered by their fingerprint input (their own shape, i.e. without the data + // state of their scans): the order is part of the fingerprint, so it must not depend on a + // data state which the fingerprint deliberately ignores + leaves.sort(Comparator.comparing(Leaf::getShape)); + out.append("J{inner,c:[").append(String.join(SEP, conditions)).append("]}("); + for (int i = 0; i < leaves.size(); i++) { + if (i > 0) { + out.append(SEP); + } + out.append(leaves.get(i).getCanonical()); + } + out.append(")"); + } + + /** + * Collect one member of an inner / cross join chain: either descend into another join of the + * chain (associativity), or append the canonical form of the whole sub tree as one chain leaf. + * The group of {@code ge} is already marked visited by the caller. + */ + private static void collectJoinChain(Group group, TreeSet<String> conditions, List<Leaf> leaves, Ctx ctx) { + if (!ctx.valid) { + return; + } + if (!ctx.visited.add(group)) { + invalid(ctx); + return; + } + GroupExpression ge = group.getFirstLogicalExpression(); + if (ge == null) { + invalid(ctx); + return; + } + Plan plan = ge.getPlan(); + if (plan instanceof LogicalJoin && isFlattenableJoinType(((LogicalJoin<?, ?>) plan).getJoinType())) { + collectJoinConditions((LogicalJoin<?, ?>) plan, conditions, ctx.mode); + for (int i = 0; i < ge.arity(); i++) { + collectJoinChain(ge.child(i), conditions, leaves, ctx); + } + return; + } + if (plan instanceof LogicalProject && ge.arity() == 1) { + // keep descending through projects so that a project between two joins cannot break the + // chain in one expression and not in another + collectJoinChain(ge.child(0), conditions, leaves, ctx); + return; + } + Canonical leaf = new Canonical(); + appendPlan(plan, ge, leaf, ctx); + if (ctx.valid) { + leaves.add(new Leaf(leaf)); + } + } + + /** One leaf of a flattened join chain: both of its canonical forms and its scans. */ + private static class Leaf { + private final Canonical canonical; + + Leaf(Canonical canonical) { + this.canonical = canonical; + } + + Canonical getCanonical() { + return canonical; + } + + String getShape() { + return canonical.shape.toString(); + } + } + + /** Inner and cross joins may be reordered, reassociated and therefore flattened. */ + private static boolean isFlattenableJoinType(JoinType joinType) { + return joinType == JoinType.INNER_JOIN || joinType == JoinType.CROSS_JOIN; + } + + private static void appendAggregate(LogicalAggregate<?> agg, GroupExpression ge, Canonical out, Ctx ctx) { + // Only the grouping keys and the child take part: the output row count of an aggregation is + // the number of groups, which the aggregate functions and the output expressions cannot + // change, so keeping them would only split one plan pattern into several cache entries. + out.append("A{gb:").append(normalizedSorted(agg.getGroupByExpressions(), ctx.mode)); + out.append("}("); + appendChild(ge, 0, out, ctx); + out.append(")"); + } + + private static void invalid(Ctx ctx) { + ctx.valid = false; + } + + /** Mark the struct info invalid because the table state could not be read (retryable). */ + private static void transientFailure(Ctx ctx) { + ctx.valid = false; + ctx.retryable = true; + } + + private static void appendScan(LogicalOlapScan scan, Canonical out, Ctx ctx) { + try { + // the scan token is the table plus the data state of the scan (baseline). Its shape form + // is the table alone and is written by this very call, so the fingerprint input is never + // derived by parsing the canonical string: no table name and no literal value can be + // mistaken for a scan token. + HboScanDescriptor descriptor = HboScanDescriptor.of(scan); + if (descriptor.getTable().indexOf(',') >= 0 || descriptor.getTable().indexOf('}') >= 0) { + // the token delimiters cannot appear in a table name: the printed struct info could + // not be read back, and a pasted one could resolve to another table + LOG.debug("scan table {} contains a struct info delimiter, hbo is disabled for it", + descriptor.getTable()); + invalid(ctx); + return; + } + out.annotated.append("S{").append(descriptor.render()).append('}'); + out.shape.append("S{").append(descriptor.getTable()).append('}'); + out.getScans().add(descriptor); + } catch (org.apache.doris.rpc.RpcException e) { + // the table version may not be readable (e.g. a cloud rpc failure): the group stays + // invalid for this lookup, but the failure is retried by the next one + LOG.debug("failed to get visible version for scan {}", scan.getTable().getNameWithFullQualifiers(), e); + transientFailure(ctx); + } + } + + /** + * Remove the baseline of every scan token of a canonical string ({@code S{db.t,v3,r1000,p1/5}} + * becomes {@code S{db.t}}), i.e. the data state of the scan - visible version, scanned rows and + * pruned partition count. + * + * <p>The fingerprint of a plan and the sort key of a join chain leaf are built by the traversal + * itself, which knows where every scan token starts and ends. This text based variant is only for + * the struct info a user pasted into {@code HBO SET STATISTICS}, whose sha256 has to match the + * fingerprint the user copied. A literal value which looks like a scan token (a string which + * starts like one and contains a comma) cannot be told apart from one there, so such a struct + * info is rejected: that statement cannot be used for that node, in either literal mode + * (there is no workaround). That is a rejected statement, never a wrong key - the fingerprint of + * a plan is not computed by this method. A table name which contains the delimiters is not + * printed at all, because such a canonical form could not be read back (see the scan token in + * {@code appendScan}). + */ + public static String stripScanBaseline(String canonicalString) { + StringBuilder sb = new StringBuilder(canonicalString.length()); + int index = 0; + while (true) { + int start = HboScanDescriptor.nextScanTokenStart(canonicalString, index); + int end = start < 0 ? -1 : canonicalString.indexOf('}', start); + if (end < 0) { + sb.append(canonicalString, index, canonicalString.length()); + break; + } + String header = canonicalString.substring(start + 2, end); + int comma = header.indexOf(','); + sb.append(canonicalString, index, start).append("S{") + .append(comma < 0 ? header : header.substring(0, comma)).append('}'); + index = end + 1; + } + return sb.toString(); + } + + private static void appendChild(GroupExpression ge, int index, Canonical out, Ctx ctx) { + if (!ctx.valid) { + return; + } + // the child is built on its own form, so that its scans are appended in the order its tokens + // appear in the parent string + Canonical child = new Canonical(); + visit(ge.child(index), child, ctx); + out.append(child); + } + + // ----------------------------------------------------------------------------------- + // expression normalization: strip slot/expr ids, keep qualifier + column + literal value + // ----------------------------------------------------------------------------------- + + private static String normalizedSorted(Set<Expression> conjuncts, LiteralMode mode) { + return conjuncts.stream().map(e -> normalizeExpression(e, mode)).sorted() + .collect(Collectors.joining(SEP)); + } + + private static String normalizedSorted(List<Expression> conjuncts, LiteralMode mode) { + return conjuncts.stream().map(e -> normalizeExpression(e, mode)).sorted() + .collect(Collectors.joining(SEP)); + } + + /** + * Render all conditions of one join node (hash and other conjuncts) as one sorted, de + * duplicated set: whether a given conjunct is classified as a hash conjunct or as an other + * conjunct depends on the grouping of the chain, so that classification cannot be part of a + * canonical form that has to be insensitive to the grouping. + */ + private static String normalizedJoinConditions(LogicalJoin<?, ?> join, LiteralMode mode) { + TreeSet<String> conditions = new TreeSet<>(); + collectJoinConditions(join, conditions, mode); + return String.join(SEP, conditions); + } + + private static void collectJoinConditions(LogicalJoin<?, ?> join, Set<String> conditions, LiteralMode mode) { + for (Expression conjunct : join.getHashJoinConjuncts()) { + conditions.add(normalizeExpression(conjunct, mode)); + } + for (Expression conjunct : join.getOtherJoinConjuncts()) { + conditions.add(normalizeExpression(conjunct, mode)); + } + } + + /** + * Rewrite every literal of a canonical string into its constant agnostic form ({@code lit(*)}), + * i.e. turn a WITH_LITERAL canonical string into the NO_LITERAL one. Used to accept a struct info + * which a user copied from a filter node of EXPLAIN for the constant agnostic fingerprint of the + * same node. + * + * <p>A literal value is not quoted in the canonical form and a data type may even contain + * parentheses ({@code lit(abc:VARCHAR(10))}), so the closing parenthesis is found by counting + * instead of by a regular expression; a value with balanced parentheses + * ({@code lit(IL (north))}) is therefore folded correctly. Only an <b>unbalanced</b> parenthesis + * in a value (e.g. {@code where city = 'IL)'}) makes the fold consume too much, and the pasted + * struct info is then rejected by the fingerprint check of {@code HBO SET STATISTICS} with the + * generic "does not match the fingerprint" message; there is no workaround, because the check + * runs in both literal modes. The constant agnostic struct info printed by EXPLAIN is built by the + * group traversal itself (see {@link #dataStateOfPlanNode}), so it has no such limitation. + */ + public static String toNoLiteral(String canonicalString) { + StringBuilder sb = new StringBuilder(canonicalString.length()); + int index = 0; + while (index < canonicalString.length()) { + int start = canonicalString.indexOf("lit(", index); + if (start < 0) { + sb.append(canonicalString, index, canonicalString.length()); + break; + } + sb.append(canonicalString, index, start).append("lit(*)"); + int depth = 0; + int cursor = start + 3; + while (cursor < canonicalString.length()) { + char c = canonicalString.charAt(cursor); + if (c == '(') { + depth++; + } else if (c == ')') { + depth--; + if (depth == 0) { + cursor++; + break; + } + } + cursor++; + } + index = cursor; + } + return sb.toString(); + } + + /** + * Normalize a single expression into its canonical component (slot -> {@code col(qualifier.name)}, + * literal -> {@code lit(value:type)} / {@code lit(*)}, other nodes -> {@code ClassName(children)}). + * Shared with the hbo join-condition canonicalizer. + */ + public static String normalizeExpression(Expression expression) { + return normalizeExpression(expression, LiteralMode.WITH_LITERAL); + } + + private static String normalizeExpression(Expression expression, LiteralMode mode) { + if (expression instanceof SlotReference) { + SlotReference slot = (SlotReference) expression; + return "col(" + String.join(".", slot.getQualifier()) + "." + slot.getName() + ")"; + } else if (expression instanceof Literal) { + if (mode == LiteralMode.NO_LITERAL) { + return "lit(*)"; + } + Literal literal = (Literal) expression; + return "lit(" + literal.getValue() + ":" + literal.getDataType() + ")"; + } else if (expression.children().isEmpty()) { + return expression.getClass().getSimpleName(); + } else { + List<Expression> children = expression.children(); + List<String> normalizedChildren = children.stream() + .map(e -> normalizeExpression(e, mode)).collect(Collectors.toList()); + if (isOrderInsensitive(expression)) { + // only order-insensitive operators (equal-to / and / or) may have their operands + // sorted; ordered comparisons (e.g. '<') must keep the original order, otherwise + // `col < 5` and `5 < col` would collapse to the same descriptor (review B2). + normalizedChildren.sort(String::compareTo); + } + String childrenStr = String.join(",", normalizedChildren); + return expression.getClass().getSimpleName() + "(" + childrenStr + ")"; + } + } + + /** Equal-to, boolean and/or are commutative; everything else keeps its operand order. */ + private static boolean isOrderInsensitive(Expression expression) { + return expression instanceof EqualTo + || expression instanceof Or + || expression instanceof And; + } + + /** Traversal state; shared along the whole subtree so the visited set guards shared sub graphs. */ + /** + * The default of {@code hbo_max_scans_per_group}, used when no session is available. 20 keeps + * every TPC-DS shape observed in this repository (the largest single sub tree is query64's + * 19 scan join group) while refusing trees which are pathologically wide. + */ + public static final int DEFAULT_MAX_SCANS_PER_GROUP = 20; + + private static int hboMaxScansPerGroup() { + ConnectContext connectContext = ConnectContext.get(); + if (connectContext == null || connectContext.getSessionVariable() == null) { + return DEFAULT_MAX_SCANS_PER_GROUP; + } + return connectContext.getSessionVariable().getHboMaxScansPerGroup(); + } + + /** + * The structural summary of a memo group: how many scan tokens the struct info of its sub tree + * would contain, and which tables it reads. + * + * <p>It is aggregated bottom up from the children's summaries - no string is rendered and no + * catalog is read - and memoized per group (see {@code Group#getOrComputeHboStructSummary}), so + * it is the cheap gate in front of the canonical string: a group which is too large for + * {@code hbo_max_scans_per_group} never renders one, and a group whose relation key can not + * match any entry does not render one either (see {@code HboPlanStatisticsManager}). + */ + public static final class HboStructSummary { + private final int scanCount; + /** The tables of the sub tree, in traversal order and with duplicates (a self join twice). */ + private final List<String> relations; + private final String relationKey; + + private HboStructSummary(int scanCount, List<String> relations) { + this.scanCount = scanCount; + this.relations = Collections.unmodifiableList(new ArrayList<>(relations)); + List<String> sorted = new ArrayList<>(relations); + Collections.sort(sorted); + this.relationKey = String.join(",", sorted); + } + + /** The number of scan tokens in the sub tree (a self join counts the table twice). */ + public int getScanCount() { + return scanCount; + } + + /** + * The tables of the sub tree as one order independent key (sorted, duplicates kept). A group + * whose struct info could equal another one's necessarily shares this key, so it is a safe + * necessary condition for an entry lookup; a group with no scan at all keys to "". + */ + public String getRelationKey() { + return relationKey; + } + + List<String> getRelations() { + return relations; + } + } + + /** The summary of a group, computed from the (memoized) summaries of its child groups. */ + public static HboStructSummary summaryOf(Group group) { + GroupExpression ge = group.getFirstLogicalExpression(); Review Comment: 认可，已修复（da26dd850df）：storage layer aggregate 会插入只含 physical expression 的子 group，原先 getFirstLogicalExpression() 会抛异常。现在 summaryOf 改为遍历 getLogicalExpressions() 并判空，这类 group 直接返回 INVALID，不再影响整条查询的 fingerprint 计算。 ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboJoinConditions.java: ########## @@ -0,0 +1,85 @@ +// 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.stats; + +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.plans.JoinType; +import org.apache.doris.nereids.trees.plans.algebra.Join; + +import com.google.common.hash.Hashing; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +/** + * Canonical description and fingerprint of the equality conditions of a join, used as the key of + * manually injected join expansion entries ({@code HBO SET EXPANSION}). + * + * <p>Only the hash (equality) conjuncts participate: the expansion behaviour of a join is a + * property of the joined column pairs, so the key deliberately excludes the join type, the shape + * of both sub trees and every non equality condition. That makes one injected entry reusable by + * every query joining the same columns. + * + * <p>The canonical form is {@code JE{EqualTo(...);EqualTo(...)}} with the pairs sorted; a pair is + * rendered by {@link GroupStructInfo#normalizeExpression(Expression)} which already sorts the two + * operands of an equality (equality is commutative), so {@code a = b} and {@code b = a} collapse. + */ +public class HboJoinConditions { + private static final String PREFIX = "JE{"; + private static final String SEP = ";"; + + private HboJoinConditions() { + } + + /** True when the injected expansion of this join type is applicable at all. */ + public static boolean isExpansionApplicable(JoinType joinType) { + // semi / anti / mark style joins can never produce more rows than their left input, and + // asof joins match at most one row per left row, so expansion can not happen there + return joinType == JoinType.INNER_JOIN + || joinType == JoinType.LEFT_OUTER_JOIN + || joinType == JoinType.RIGHT_OUTER_JOIN + || joinType == JoinType.FULL_OUTER_JOIN; + } + + /** + * Canonical string of the equality conditions, empty when the join has no equality condition + * (e.g. a cross join). + */ + public static Optional<String> canonicalOf(Join join) { + List<String> pairs = new ArrayList<>(); + for (Expression conjunct : join.getHashJoinConjuncts()) { + if (conjunct instanceof EqualTo) { Review Comment: 认可，已修复（da26dd850df）：NullSafeEqual 同样是 hash conjunct（getHashJoinConjuncts 会返回），现在 <=> 与普通等值一起参与 canonical。 ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboStructFreshness.java: ########## @@ -0,0 +1,585 @@ +// 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.stats; + +import org.apache.doris.catalog.DatabaseIf; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.Config; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan; +import org.apache.doris.rpc.RpcException; +import org.apache.doris.statistics.hbo.PlanStatistics; +import org.apache.doris.statistics.hbo.ScanPlanStatistics; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +/** + * Whether a hbo entry may still be applied: the recorded data state of an entry (the baseline of + * every scan of its struct info, see {@link HboScanDescriptor}) compared with the data state of the + * query which wants to use it. + * + * <p>A hbo fingerprint does not contain any data state any more, so an entry keeps matching while a + * table grows. The read side therefore has to decide whether the injected row count still describes + * the data it was measured on: + * <ul> + * <li>{@link #STATE_LIVE}: the data the entry was measured on is still the current one, the entry + * is applied;</li> + * <li>{@link #STATE_DRIFTED}: the data moved, but by at most + * {@code Config.hbo_row_count_change_ratio} of the recorded row count (10% by default): the + * entry is applied, which is the point of the tolerance - a table which merely keeps growing + * (e.g. a few new rows in a partition) does not invalidate a measured row count;</li> + * <li>{@link #STATE_STALE}: the data moved further than that: the entry is <b>not</b> applied, and + * the query falls back to the optimizer estimation;</li> + * <li>{@link #STATE_UNKNOWN}: nothing can be compared (the entry records no row count, or the + * current one cannot be read / was obtained a different way), so there is nothing to judge: the + * entry is applied as it always was.</li> + * </ul> + * The recorded row count is never scaled: an entry either applies with the value the user measured + * or does not apply at all ("soft strategy" - a user who needs the exact recorded state in any case + * should not inject hbo statistics for that query). Setting + * {@code hbo_row_count_change_ratio} to 0 or less switches the verdict back to the strict semantics + * hbo had before the baseline existed: the recorded visible version has to still be the current one. + * + * <p>The rows are compared <b>before</b> the version, and only against a number which was obtained + * the same way ({@link HboScanDescriptor.RowsKind}): the visible version is a property of the table, + * so two different partition selections of one table share it while reading completely different + * amounts of data, and a measured row count must not be compared with one that was derived from the + * table average. Whenever nothing can be compared - a hand written struct literal, an entry whose + * rows were never measured, a current row count which cannot be read, or two numbers of different + * kinds - the state is {@link #STATE_UNKNOWN} and the entry is applied: a transient failure to read + * the catalog must not drop an entry the user injected. The version is only decisive in strict mode + * ({@code hbo_row_count_change_ratio <= 0}). + * + * <p>A scan is judged by the rows of its <b>selected</b> partitions, so pruning decides which data + * the entry depends on: an unrelated partition which grows does not invalidate a pruned entry. + * {@code HBO SHOW STATISTICS} and {@code HBO DELETE STALE STATISTICS} cannot run the query of an + * entry, so there the recorded state is compared with the table as a whole; an entry whose struct + * prunes partitions can therefore only be judged by rows when it was recorded as a whole table scan + * (its state is reported as unknown otherwise, and such an entry is kept unless the user asks for + * {@code OLDER_THAN}). + * + * <p>Learned (profile collected) entries carry the input table statistics of the run they were + * measured in, so they are judged by the same tolerance over those recorded rows; an entry injected + * by {@code HBO SET LEARNED STATISTICS} carries no such statistics and is applied as it always was. + */ +public class HboStructFreshness { + /** The recorded data state is still the current one. */ + public static final String STATE_LIVE = "live"; + /** The data moved, but within the tolerance of {@code Config.hbo_row_count_change_ratio}. */ + public static final String STATE_DRIFTED = "drifted"; + /** The data moved further than the tolerance: the entry must not be applied. */ + public static final String STATE_STALE = "stale"; + /** The entry records no data state (or it cannot be read any more): nothing to judge. */ + public static final String STATE_UNKNOWN = "unknown"; + + private static final Logger LOG = LogManager.getLogger(HboStructFreshness.class); + + private final String state; + // the recorded baseline and the current state, both rendered for display + private final String recorded; + private final String live; + // the row counts of the scan which decided the state, {@link HboScanDescriptor#UNKNOWN} when the + // state was decided by something else (e.g. a version which is not readable) + private final long recordedRows; + private final long liveRows; + + private HboStructFreshness(String state, String recorded, String live, long recordedRows, long liveRows) { + this.state = state; + this.recorded = recorded; + this.live = live; + this.recordedRows = recordedRows; + this.liveRows = liveRows; + } + + public String getState() { + return state; + } + + public boolean isStale() { + return STATE_STALE.equals(state); + } + + public boolean isUnknown() { + return STATE_UNKNOWN.equals(state); + } + + public boolean isDrifted() { + return STATE_DRIFTED.equals(state); + } + + /** The data state the entry was recorded in, e.g. {@code hbo_test.t1:v3,r1000,p1/5}. */ + public String getRecorded() { + return recorded; + } + + /** The data state the entry is compared with (empty when it cannot be read). */ + public String getLive() { + return live; + } + + /** + * The verdict as one token, printed after {@code used=} / {@code skipped=} in the explain + * annotation, e.g. {@code drifted(rows=1030,rec=1000,+3.0%)}. + */ + public String getSummary() { + if (!STATE_DRIFTED.equals(state) && !STATE_STALE.equals(state)) { + return state; + } + if (recordedRows == HboScanDescriptor.UNKNOWN || liveRows == HboScanDescriptor.UNKNOWN) { + return state; + } + return state + "(rows=" + liveRows + ",rec=" + recordedRows + "," + + changeText(recordedRows, liveRows) + ")"; + } + + /** The {@code now} part of {@code HBO SHOW STATISTICS}: the state read from the catalog. */ + public String getLiveDetail() { + if (live.isEmpty()) { + return "-"; + } + if (recordedRows == HboScanDescriptor.UNKNOWN || liveRows == HboScanDescriptor.UNKNOWN) { + return live; + } + return live + "," + changeText(recordedRows, liveRows); + } + + /** + * Whether the entry records any data state at all. An entry which does not (a struct literal + * typed by hand, or one copied from a plan whose tables were never reported) is always applied, + * so a caller does not have to compute the state of the current query to find that out. + */ + public static boolean hasRecordedDataState(String recordedCanonical) { + for (HboScanDescriptor scan : HboScanDescriptor.parseAll(recordedCanonical)) { + if (scan.hasVisibleVersion() || scan.hasScanRows()) { + return true; + } + } + return false; + } + + /** + * Judge an entry against the current query: the data state recorded in {@code recordedCanonical} + * (the struct info of the entry) against the state of {@code liveCanonical} (the struct info the + * read side just computed for the group). This is the decision of the read side. + */ + public static HboStructFreshness between(String recordedCanonical, GroupStructInfo liveStructInfo) { + return between(recordedCanonical, liveStructInfo.getScans()); + } + + /** The same comparison for a caller which has the scans of the current query only. */ + static HboStructFreshness between(String recordedCanonical, List<HboScanDescriptor> liveScans) { + List<HboScanDescriptor> recordedScans = HboScanDescriptor.parseAll(recordedCanonical); + String liveText = renderScans(liveScans); + if (recordedScans.isEmpty()) { + // an entry without a baseline (a hand written struct literal): nothing to judge + return new HboStructFreshness(STATE_UNKNOWN, "", liveText, + HboScanDescriptor.UNKNOWN, HboScanDescriptor.UNKNOWN); + } + if (recordedScans.size() != liveScans.size()) { + // the entry describes a different number of scans than the group: cannot be compared + return new HboStructFreshness(STATE_UNKNOWN, renderScans(recordedScans), liveText, + HboScanDescriptor.UNKNOWN, HboScanDescriptor.UNKNOWN); + } + String state = STATE_LIVE; + String recordedText = renderScans(recordedScans); + + long recordedRows = HboScanDescriptor.UNKNOWN; + long liveRows = HboScanDescriptor.UNKNOWN; + for (int i = 0; i < recordedScans.size(); i++) { + HboScanDescriptor recorded = recordedScans.get(i); + HboScanDescriptor live = liveScans.get(i); + String scanState; + if (!recorded.getTable().equals(live.getTable())) { + // the same fingerprint describes the same scans in the same order, so this cannot + // happen for an entry of this group: refuse to judge rather than guess + scanState = STATE_UNKNOWN; + } else { + scanState = stateOf(recorded, live); + } + if (worse(scanState, state).equals(scanState)) { + state = scanState; + recordedRows = recorded.getScanRows(); + liveRows = live.getScanRows(); + } + } + return new HboStructFreshness(state, recordedText, liveText, recordedRows, liveRows); + } + + /** + * Judge an entry against the catalog as a whole (used by {@code HBO SHOW STATISTICS} and + * {@code HBO DELETE STALE STATISTICS}, which cannot run the query of an entry): the recorded + * visible version and row count of every table against the current ones. + * + * <p>Row counts are only comparable when the scan read the whole table (no partition was pruned + * away), because only then the recorded rows are the rows of the table. + */ + public static HboStructFreshness of(String canonicalStructInfo) { + List<HboScanDescriptor> recordedScans = HboScanDescriptor.parseAll(canonicalStructInfo); + if (recordedScans.isEmpty()) { + return new HboStructFreshness(STATE_UNKNOWN, "", "", HboScanDescriptor.UNKNOWN, + HboScanDescriptor.UNKNOWN); + } + String state = STATE_LIVE; + // the current state of every table, de duplicated like the recorded side (a self join reads + // the same table twice, which would only repeat the same numbers) + Set<String> liveTexts = new LinkedHashSet<>(); + long recordedRows = HboScanDescriptor.UNKNOWN; + long liveRows = HboScanDescriptor.UNKNOWN; + for (HboScanDescriptor recorded : recordedScans) { + OlapTable table = resolveTable(recorded.getTable()); + if (table == null) { + if (worse(STATE_UNKNOWN, state).equals(STATE_UNKNOWN)) { + state = STATE_UNKNOWN; + } + liveTexts.add(recorded.getTable() + ":-"); + continue; + } + if (table.getIndexIdList().size() > 1 && !recorded.isPartitionSelectionComplete()) { + // the table has more than one index (a rollup) and the entry records only a partition + // count, so this statement cannot tell which index its row count came from: judging it + // against the base index could be wrong in both directions, so it is left to a query + liveTexts.add(recorded.getTable() + ":-"); + if (worse(STATE_UNKNOWN, state).equals(STATE_UNKNOWN)) { + state = STATE_UNKNOWN; + } + continue; + } + long currentVersion = visibleVersionOf(table); + // the rows of the whole table, computed the same way the recorded rows of a whole table + // scan were computed, so the two numbers describe the same thing + HboScanDescriptor.Rows liveRowsOfTable = HboScanDescriptor.rowsOf(table, + new ArrayList<>(table.getPartitionIds()), table.getBaseIndexId()); + long currentRows = liveRowsOfTable.getRows(); + HboScanDescriptor.RowsKind currentRowsKind = liveRowsOfTable.getKind(); + liveTexts.add(recorded.getTable() + ":v" + currentVersion + + (currentRows == HboScanDescriptor.UNKNOWN ? "" + : (currentRowsKind == HboScanDescriptor.RowsKind.MEASURED ? ",r" : ",e") + + currentRows)); + String scanState; + if (!recorded.hasVisibleVersion() && !recorded.hasScanRows()) { + // nothing was recorded for this scan: it applies as it always was, nothing to judge + scanState = STATE_UNKNOWN; + currentRows = HboScanDescriptor.UNKNOWN; + } else if (threshold() <= 0) { + if (currentVersion == HboScanDescriptor.UNKNOWN || !recorded.hasVisibleVersion()) { + // a temporary metadata failure must not look like a version mismatch: DELETE + // STALE would remove an entry which a query can still apply + scanState = STATE_UNKNOWN; + } else { + scanState = currentVersion == recorded.getVisibleVersion() ? STATE_LIVE : STATE_STALE; + } + currentRows = HboScanDescriptor.UNKNOWN; + } else if (recorded.hasVisibleVersion() && currentVersion == recorded.getVisibleVersion()) { + // the visible version is a property of the whole table, so an unchanged version means + // that nothing was loaded into any of its partitions: the recorded state is still the + // current one, whatever the entry was measured on. (The read side can not use this + // shortcut: it knows which partitions the entry was measured on and which ones the + // query reads, and those can differ under one unchanged version.) + scanState = STATE_LIVE; + } else if (!recorded.hasScanRows() || !recorded.isPartitionSelectionComplete()) { + // the entry records no row count at all (a struct literal of an older format, for + // example), or it records the rows of pruned partitions only, which are not the rows + // of the table: the state can only be judged by a query of that entry + scanState = STATE_UNKNOWN; + currentRows = HboScanDescriptor.UNKNOWN; + } else if (currentRows == HboScanDescriptor.UNKNOWN + || currentRowsKind != recorded.getRowsKind()) { + // the current row count cannot be read or was obtained differently than the recorded + // one, so the two are not comparable: the read side reports unknown for exactly this + // case and keeps applying the entry, and this scope must not disagree by calling it + // stale (DELETE STALE would then remove an entry which queries still use) + scanState = STATE_UNKNOWN; + currentRows = HboScanDescriptor.UNKNOWN; + } else if (recorded.getScanRows() == currentRows) { + scanState = STATE_LIVE; + } else { + scanState = changeRatio(recorded.getScanRows(), currentRows) <= threshold() + ? STATE_DRIFTED : STATE_STALE; + } + if (worse(scanState, state).equals(scanState) && scanState != STATE_LIVE) { + state = scanState; + recordedRows = scanState == STATE_UNKNOWN + ? HboScanDescriptor.UNKNOWN : recorded.getScanRows(); + liveRows = scanState == STATE_UNKNOWN ? HboScanDescriptor.UNKNOWN : currentRows; + } + } + return new HboStructFreshness(state, renderScans(recordedScans), String.join("; ", liveTexts), + recordedRows, liveRows); + } + + /** + * Judge a learned (profile collected) entry: the rows its input tables had when it was measured + * against the rows of the scans of the current query. The caller must only use this for an entry + * of a node whose recorded rows are the rows of the data it reads (a scan node: see + * {@code HboStatsCalculator.applyLearnedStats}). The input tables are aggregated by table name, because + * a learned entry describes the scans of the node it was published for in plan order while the + * struct info orders them canonically. + * + * <p>This is a one way guard: it only rejects an entry when a comparable pair of row counts + * proves that the data moved beyond the tolerance. Everything which cannot be compared (an + * injected entry without input statistics, a different number of tables, a row count of a + * different kind) leaves the entry applicable, so a learned lookup which worked before this + * check keeps working. + */ + public static HboStructFreshness ofLearnedEntry(List<PlanStatistics> recordedInputStatistics, + GroupStructInfo liveStructInfo) { + return ofLearnedEntry(recordedInputStatistics, liveStructInfo.getScans()); + } + + /** The same guard for a caller which has the scans of the current query only. */ + static HboStructFreshness ofLearnedEntry(List<PlanStatistics> recordedInputStatistics, + List<HboScanDescriptor> liveScans) { + Map<String, Long> recordedRowsByTable = new LinkedHashMap<>(); + if (recordedInputStatistics != null) { + for (PlanStatistics input : recordedInputStatistics) { + if (!(input instanceof ScanPlanStatistics)) { + continue; + } + PhysicalOlapScan scan = ((ScanPlanStatistics) input).getScan(); + if (scan == null || scan.getTable() == null) { + continue; + } + String table = scan.getTable().getNameWithFullQualifiers(); + // the input rows of a scan node are the rows it read, i.e. the number a catalog row + // count (which also ignores the predicates) can be compared with + recordedRowsByTable.merge(table, input.getInputRows(), Long::sum); + } + } + if (recordedRowsByTable.isEmpty()) { + // an injected learned entry carries no input table statistics: nothing to compare + return new HboStructFreshness(STATE_UNKNOWN, "", renderScans(liveScans), + HboScanDescriptor.UNKNOWN, HboScanDescriptor.UNKNOWN); + } + return ofLearnedRows(recordedRowsByTable, liveScans); + } + + /** + * The comparison of a learned entry: the rows it recorded for each input table against the rows + * of the scans of the current query. Only measured row counts are comparable with the recorded + * ones (a learned number is a measurement of a real run), and a table whose rows cannot be + * compared leaves the entry applicable. + */ + static HboStructFreshness ofLearnedRows(Map<String, Long> recordedRowsByTable, + List<HboScanDescriptor> liveScans) { + Map<String, long[]> liveByTable = new LinkedHashMap<>(); + Map<String, HboScanDescriptor.RowsKind> liveKinds = new LinkedHashMap<>(); + for (HboScanDescriptor scan : liveScans) { + long[] rows = liveByTable.computeIfAbsent(scan.getTable(), key -> new long[] {0}); + if (scan.hasScanRows()) { + rows[0] += scan.getScanRows(); + } else { + rows[0] = HboScanDescriptor.UNKNOWN; + } + HboScanDescriptor.RowsKind kind = liveKinds.get(scan.getTable()); + // a table which is scanned twice is only comparable when both scans agree on the kind + liveKinds.put(scan.getTable(), kind == null || kind == scan.getRowsKind() + ? scan.getRowsKind() : HboScanDescriptor.RowsKind.UNKNOWN); + } + String state = STATE_LIVE; + boolean compared = false; + long recordedRows = HboScanDescriptor.UNKNOWN; + long liveRows = HboScanDescriptor.UNKNOWN; + for (Map.Entry<String, Long> entry : recordedRowsByTable.entrySet()) { + long[] live = liveByTable.get(entry.getKey()); + long recorded = entry.getValue(); + if (live == null || recorded == HboScanDescriptor.UNKNOWN + || live[0] == HboScanDescriptor.UNKNOWN + || liveKinds.get(entry.getKey()) != HboScanDescriptor.RowsKind.MEASURED) { + // the learned numbers are measured row counts, so they are only compared with a + // measured one: a derived number could be off by a large factor + continue; + } + compared = true; + String tableState = recorded == live[0] ? STATE_LIVE + : changeRatio(recorded, live[0]) <= threshold() ? STATE_DRIFTED : STATE_STALE; + if (worse(tableState, state).equals(tableState)) { + state = tableState; + recordedRows = recorded; + liveRows = live[0]; + } + } + if (!compared) { + // nothing could be compared, so the entry is applied as it always was + return new HboStructFreshness(STATE_UNKNOWN, renderTables(recordedRowsByTable), + renderTables(liveByTable), HboScanDescriptor.UNKNOWN, HboScanDescriptor.UNKNOWN); + } + return new HboStructFreshness(state, renderTables(recordedRowsByTable), renderTables(liveByTable), + recordedRows, liveRows); + } + + /** Render the aggregated row counts of a learned entry, e.g. {@code hbo_test.t1:r1000}. */ + private static String renderTables(Map<String, ? extends Object> rowsByTable) { + List<String> rendered = new ArrayList<>(); + for (Map.Entry<String, ? extends Object> entry : rowsByTable.entrySet()) { + long rows = entry.getValue() instanceof long[] ? ((long[]) entry.getValue())[0] + : (Long) entry.getValue(); + // a learned number is a measurement of a real run, so it is rendered as one + rendered.add(entry.getKey() + (rows == HboScanDescriptor.UNKNOWN ? "" : ":r" + rows)); + } + return String.join("; ", rendered); + } + + /** + * The state of one scan of the current query against the same scan of an entry. + * See the class comment for the meaning of the states. + */ + private static String stateOf(HboScanDescriptor recorded, HboScanDescriptor live) { + if (!recorded.hasVisibleVersion() && !recorded.hasScanRows()) { + // the entry records nothing about this scan (e.g. a struct literal typed by hand) + return STATE_UNKNOWN; + } + if (threshold() <= 0) { + // strict mode: only the recorded data state may be reused - the semantics hbo had before + // the baseline existed, when the visible version was part of the key + return recorded.hasSameVersion(live) ? STATE_LIVE : STATE_STALE; + } + if (recorded.hasScanRows() && live.hasScanRows()) { + if (recorded.getRowsKind() != live.getRowsKind()) { + // the two numbers were obtained differently (one is a measurement of the partitions + // the scan selected, the other one is derived from the whole table), so they are not + // comparable: comparing them would reject entries whose data did not change at all, + // and the version alone can not say whether the two describe the same data (the same + // version covers every partition of the table). The state is therefore reported as + // unknown and the entry is applied as it always was. + return STATE_UNKNOWN; + } + // the rows are compared BEFORE the version: a different partition selection (or another + // scan of the same table) can read a completely different amount of data under the very + // same table version, so equal versions do not mean that the entry still fits + if (recorded.getScanRows() == live.getScanRows()) { + return STATE_LIVE; + } + return changeRatio(recorded.getScanRows(), live.getScanRows()) <= threshold() + ? STATE_DRIFTED : STATE_STALE; + } + // the recorded baseline was never a measurement, or the current row count cannot be read: + // there is nothing to judge, so the entry is applied as it always was + return STATE_UNKNOWN; + } + + /** How much the row count of a scan may change and still be reused; a non-positive value is strict. */ + private static double threshold() { + return Config.hbo_row_count_change_ratio; + } + + /** {@code |now - recorded| / recorded}; 1 when the recorded side is empty and the other is not. */ + private static double changeRatio(long recordedRows, long liveRows) { + if (recordedRows == 0) { + return liveRows == 0 ? 0 : 1; + } + return Math.abs(liveRows - recordedRows) / (double) recordedRows; + } + + /** {@code +3.0%} / {@code -20.0%}: how the current row count differs from the recorded one. */ + private static String changeText(long recordedRows, long liveRows) { + double change = recordedRows == 0 + ? (liveRows == 0 ? 0 : 1) : (liveRows - recordedRows) / (double) recordedRows; + return String.format(Locale.ROOT, "%+.1f%%", change * 100); + } + + /** The worse of two states (live, drifted, unknown, stale - in this order). */ + private static String worse(String left, String right) { + return severity(left) >= severity(right) ? left : right; + } + + private static int severity(String state) { + switch (state) { + case STATE_STALE: + return 3; + case STATE_UNKNOWN: + return 2; + case STATE_DRIFTED: + return 1; + default: + return 0; + } + } + + /** Render the baseline of a scan, e.g. {@code hbo_test.t1:v3,r1000,p1/5}. */ + private static String renderScans(List<HboScanDescriptor> scans) { + Set<String> rendered = new LinkedHashSet<>(); + for (HboScanDescriptor scan : scans) { + StringBuilder sb = new StringBuilder(scan.getTable()); + if (scan.hasVisibleVersion()) { + sb.append(":v").append(scan.getVisibleVersion()); + } + if (scan.hasScanRows()) { + sb.append(scan.getRowsKind() == HboScanDescriptor.RowsKind.MEASURED ? ",r" : ",e") + .append(scan.getScanRows()); + } + if (scan.getTotalPartitions() != HboScanDescriptor.UNKNOWN + && !scan.isPartitionSelectionComplete()) { + sb.append(",p").append(scan.getSelectedPartitions()).append('/') + .append(scan.getTotalPartitions()); + } + rendered.add(sb.toString()); + } + return String.join("; ", rendered); + } + + /** The current visible version of a table, or {@link HboScanDescriptor#UNKNOWN}. */ + private static long visibleVersionOf(OlapTable table) { + try { + return table.getVisibleVersion(); + } catch (RpcException | RuntimeException e) { + LOG.debug("failed to read the visible version of {}", table.getName(), e); + return HboScanDescriptor.UNKNOWN; + } + } + + /** Resolve {@code [catalog.]db.table} to the olap table, or null when it does not exist. */ + private static OlapTable resolveTable(String fullName) { + String[] parts = fullName.split("\\."); + try { + if (parts.length < 2 || parts.length > 3) { Review Comment: 认可，已修复（da26dd850df）：resolveTable 不再固定按 3 段切分，改为依次尝试可能的 catalog / database 组合（长的 database 名优先），nested namespace 下 internal.tenant.analytics.t 可以正确解析。 ########## fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java: ########## @@ -7182,6 +7185,146 @@ public LogicalPlan visitAdminSetFrontendConfig(DorisParser.AdminSetFrontendConfi applyToAll); } + @Override + public LogicalPlan visitHboSetStatistics(DorisParser.HboSetStatisticsContext ctx) { + checkHboStatementWords(ctx.hbo, ctx.statistics); + Double value = null; + String typeName = null; + String fingerprint = null; + String structCanonical = null; + String literalModeName = null; + for (DorisParser.HboSetParamContext param : ctx.hboSetParam()) { + if (param instanceof DorisParser.HboSetValueContext) { + if (value != null) { + throw new ParseException("VALUE is given twice in hbo set statistics statement"); + } + try { + value = Double.parseDouble(((DorisParser.HboSetValueContext) param).value.getText()); + } catch (NumberFormatException e) { + throw new ParseException("hbo statistics VALUE out of range: " + + ((DorisParser.HboSetValueContext) param).value.getText()); + } + } else if (param instanceof DorisParser.HboSetTypeContext) { + if (typeName != null) { + throw new ParseException("TYPE is given twice in hbo set statistics statement"); + } + typeName = ((DorisParser.HboSetTypeContext) param).typeName.getText(); + } else if (param instanceof DorisParser.HboSetStructContext) { + if (structCanonical != null) { + throw new ParseException("STRUCT is given twice in hbo set statistics statement"); + } + structCanonical = stripQuotes(((DorisParser.HboSetStructContext) param).structCanonical.getText()); Review Comment: 认可，已修复（da26dd850df）：STRUCT='...' 是 STRING_LITERAL，改为走 SqlLiteralUtils.parseStringLiteral()，与其它字符串字面量一致的转义解码。 ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/HboStatisticsCommand.java: ########## @@ -0,0 +1,330 @@ +// 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.trees.plans.commands; + +import org.apache.doris.catalog.Env; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.stats.GroupStructInfo; +import org.apache.doris.nereids.stats.HboPlanStatisticsManager; +import org.apache.doris.nereids.stats.HboPlanStatisticsManager.LiteralMode; +import org.apache.doris.nereids.stats.HboPlanStatisticsManager.PinnedType; +import org.apache.doris.nereids.stats.HboStatisticsStore; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; + +import com.google.common.hash.Hashing; + +import java.nio.charset.StandardCharsets; +import java.util.Locale; +import java.util.regex.Pattern; + +/** + * Manual HBO statistics management statements: + * <pre> + * HBO SET [PINNED|LEARNED] STATISTICS VALUE=<rows> [TYPE=<type>] [STRUCT='<canonical>'] + * FINGERPRINT='<fingerprint>' [LITERAL_MODE=<mode>] + * HBO DELETE [PINNED|LEARNED] STATISTICS FINGERPRINT='<fingerprint>' + * </pre> + * The optional scope selects which cache is written: {@code PINNED} (default) injects an entry that + * is authoritative over the learned entries, {@code LEARNED} injects into the learned cache so the + * learned lookup path can be exercised without a real profile publish. The injected value is bound + * to the fingerprint the user copied from EXPLAIN; which struct info that fingerprint belongs to is + * decided here (an explicit {@code LITERAL_MODE} wins, the default is the constant agnostic form + * every join / aggregation key uses) and persisted with the entry. + * + * <p>The struct info is not just a label: it carries the data state of every scan it reads + * ({@code S{db.t,v3,r1000,p1/5}}, see {@link org.apache.doris.nereids.stats.HboScanDescriptor}), + * which the read side compares with + * the data state of the query before applying the entry + * (see {@link org.apache.doris.nereids.stats.HboStructFreshness}). Copying + * the {@code struct=} value of EXPLAIN therefore also records what the injected value was measured + * on. + */ +public class HboStatisticsCommand extends Command { + + /** Operation kind. */ + public enum Op { + SET, + DELETE + } + + /** Which cache the statement targets. */ + public enum Scope { + PINNED, + LEARNED + } + + private static final Pattern FINGERPRINT_PATTERN = Pattern.compile("[0-9a-fA-F]{64}"); + + private final Op op; + private final Scope scope; + private final String fingerprint; + // a row count, or the fan-out factor of a PinnedType.JOIN_EXPANSION entry + private final double value; + private final PinnedType type; + private final String structCanonical; + // which struct info the entry is keyed by (WITH_LITERAL / NO_LITERAL), null means "derive" + private final String literalModeName; + + /** + * HboStatisticsCommand + * @param op SET or DELETE + * @param scopeName optional PINNED / LEARNED, null or empty means PINNED + * @param fingerprint hbo fingerprint (sha256 of the simplified group struct info) + * @param value injected output row count, or the fan-out factor for JOIN_EXPANSION + * @param typeName optional EXACT / FILTER_SMALL, null means EXACT + * @param structCanonical optional human-readable simplified struct info canonical string + * @param literalModeName optional WITH_LITERAL / NO_LITERAL, null means the constant agnostic + * (NO_LITERAL) form, which is also what the struct is folded to + */ + public HboStatisticsCommand(Op op, String scopeName, String fingerprint, double value, + String typeName, String structCanonical, String literalModeName) { + super(PlanType.HBO_STATISTICS_COMMAND); + this.op = op; + this.scope = parseScope(scopeName); + // sha256 hex fingerprints are lowercase everywhere (group struct info, read-side lookup); + // normalize user input so an uppercase fingerprint cannot silently miss its entry + this.fingerprint = fingerprint == null ? null : fingerprint.toLowerCase(Locale.ROOT); + this.value = value; + this.type = typeName == null ? PinnedType.EXACT : PinnedType.fromName(typeName); + this.structCanonical = structCanonical == null ? "" : structCanonical; + this.literalModeName = literalModeName; + } + + @Override + public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { + if (!Env.getCurrentEnv().getAccessManager() + .checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) { + throw new AnalysisException("Access denied: HBO statistics management requires ADMIN privilege"); + } + HboPlanStatisticsManager hboManager = Env.getCurrentEnv().getHboPlanStatisticsManager(); + // validate the fingerprint shape on both operations: a malformed DELETE would otherwise + // silently leave the pinned entry active (it can never match a stored fingerprint) + validateFingerprint(fingerprint); + if (type == null) { + throw new AnalysisException( + "invalid hbo statistics type, expect EXACT, FILTER_SMALL or JOIN_EXPANSION"); + } + if (scope == Scope.LEARNED && type != PinnedType.EXACT) { + // a learned entry only carries the row count; the other types are pinned-only concepts + throw new AnalysisException("TYPE is not supported for hbo learned statistics"); + } + if (scope == Scope.LEARNED && literalModeName != null) { + throw new AnalysisException("LITERAL_MODE is not supported for hbo learned statistics"); + } + if (op == Op.SET) { + // a learned entry has no key of its own (it is looked up by the fingerprint the read side + // computes), so its struct info is a label only: it is validated as pasted instead of + // being folded to the constant agnostic form + LiteralMode literalMode = scope == Scope.LEARNED ? LiteralMode.WITH_LITERAL + : resolveLiteralMode(literalModeName); + // a pinned entry is only displayable when it carries its struct info, and a struct info + // which does not belong to the fingerprint would make the SHOW output misleading + String canonical = validateStructCanonical(fingerprint, structCanonical, type, literalMode, + scope != Scope.LEARNED, scope != Scope.LEARNED); + if (!HboStatisticsStore.fitsStructColumn(canonical)) { + throw new AnalysisException("hbo statistics STRUCT is too long to be persisted without" + + " truncation (the scan baseline would be cut off), shorten the node it was" + + " copied from"); + } + if (type == PinnedType.JOIN_EXPANSION) { + // a fan-out factor relative to the larger input of the join: >= 1 means the join + // expands, < 1 means it filters (0.1 keeps 10% of that input), so only zero and + // negative values are meaningless + if (value <= 0) { + throw new AnalysisException( + "hbo join expansion must be greater than 0: " + value); + } + } else if (value < 0 || value != Math.floor(value) || value >= (double) Long.MAX_VALUE) { + // the VALUE of every other type is a row count; a double above Long.MAX_VALUE would + // not be rejected by the integer check but would silently saturate when it is cast + // (and persisted) as a long + throw new AnalysisException("hbo statistics VALUE must be a non-negative integer" + + " no greater than " + Long.MAX_VALUE + " for type " + type + ": " + value); + } + if (scope == Scope.LEARNED) { + hboManager.putLearnedPlanStatistics(fingerprint, (long) value, structCanonical); + } else if (type == PinnedType.JOIN_EXPANSION) { + hboManager.putPinnedExpansionStatistics(fingerprint, value, canonical); + } else { + hboManager.putPinnedPlanStatistics(fingerprint, (long) value, type, canonical, literalMode); + } + return; + } + switch (op) { + case DELETE: + if (scope == Scope.LEARNED) { + hboManager.removeLearnedPlanStatistics(fingerprint); + } else { + hboManager.removePinnedPlanStatistics(fingerprint); + } + break; + default: + throw new IllegalStateException("unexpected hbo statistics op: " + op); + } + } + + /** + * Which struct info the entry is keyed by. An explicit LITERAL_MODE wins (the pasted struct is + * folded accordingly); the default is the constant agnostic form, because that is the key the + * join / aggregation read paths use and it makes a filter injection survive a constant change. + */ + private static LiteralMode resolveLiteralMode(String modeName) throws AnalysisException { + if (modeName == null) { + return LiteralMode.NO_LITERAL; + } + for (LiteralMode mode : LiteralMode.values()) { + if (mode.name().equalsIgnoreCase(modeName)) { + return mode; + } + } + throw new AnalysisException( + "invalid hbo statistics LITERAL_MODE, expect WITH_LITERAL or NO_LITERAL: " + modeName); + } + + /** The canonical the entry is keyed by: the pasted struct, folded for the agnostic mode. */ + private static String canonicalFor(String structCanonical, LiteralMode literalMode) { + return literalMode == LiteralMode.NO_LITERAL + ? GroupStructInfo.toNoLiteral(structCanonical) : structCanonical; + } + + @Override + public <R, C> R accept(PlanVisitor<R, C> visitor, C context) { + return visitor.visitCommand(this, context); + } + + public Op getOp() { + return op; + } + + public Scope getScope() { + return scope; + } + + public String getFingerprint() { + return fingerprint; + } + + public long getRows() { + return (long) value; + } + + /** The row count, or the fan-out factor of a JOIN_EXPANSION entry. */ + public double getValue() { + return value; + } + + public PinnedType getPinnedType() { + return type; + } + + public String getStructCanonical() { + return structCanonical; + } + + private static Scope parseScope(String scopeName) { + if (scopeName == null || scopeName.isEmpty() || "pinned".equalsIgnoreCase(scopeName)) { + return Scope.PINNED; + } + if ("learned".equalsIgnoreCase(scopeName)) { + return Scope.LEARNED; + } + throw new IllegalArgumentException( + "invalid hbo statistics scope, expect PINNED or LEARNED: " + scopeName); + } + + /** + * Check that the pasted struct info describes the sub tree the fingerprint was taken from. The + * canonical form is accepted as it is printed by EXPLAIN, and also with every literal replaced + * by {@code lit(*)} (see {@link GroupStructInfo#toNoLiteral}) because the constant agnostic + * fingerprint of a filter node is looked up with exactly that struct info. + * + * @param required whether a missing struct info is an error (pinned entries are displayed by + * {@code HBO SHOW STATISTICS}, so they must carry one) + * @param keyedEntry whether the struct info is the key of the entry. A pinned entry is looked up + * by the sha256 of what the user pasted, so the struct info has to be a form + * the read side can produce; the struct info of a learned entry is a label + * only (its key is generated internally), so it is accepted as it is - which + * is also why a learned join / aggregation entry may carry its J-shape or + * A-shape struct even though no literal carrying form of those exists. + */ + private String validateStructCanonical(String targetFingerprint, String structCanonical, PinnedType type, + LiteralMode literalMode, boolean required, boolean keyedEntry) throws AnalysisException { + if (structCanonical.isEmpty()) { + if (required) { + throw new AnalysisException("hbo statistics STRUCT is required," + + " copy the struct= value of the target node from EXPLAIN"); + } + return ""; + } + // the type decides which struct info the entry can be keyed by at all + if (type == PinnedType.JOIN_EXPANSION && !structCanonical.startsWith("JE{")) { + throw new AnalysisException("TYPE=JOIN_EXPANSION is keyed by the join conditions," + + " its STRUCT must be the condition canonical: STRUCT='JE{...}'"); + } + if (type == PinnedType.FILTER_SMALL && !structCanonical.startsWith("F{")) { + throw new AnalysisException("TYPE=FILTER_SMALL can only guard a filter entry," + + " its STRUCT must be a filter root: STRUCT='F{...}(...)'"); + } + if (keyedEntry && type != PinnedType.JOIN_EXPANSION && structCanonical.startsWith("JE{")) { + // the row count of a join is pinned by the struct info of its group (J{...}); a + // condition canonical is only read by the expansion path, so such an entry could never + // be applied - report it instead of storing an entry which silently does nothing + throw new AnalysisException("a join row count entry is keyed by the group struct info," + + " its STRUCT must be the group canonical: STRUCT='J{...}(...)'. A JE{...}" + + " condition canonical is only read by TYPE=JOIN_EXPANSION"); + } + if (keyedEntry && type != PinnedType.JOIN_EXPANSION && literalMode == LiteralMode.WITH_LITERAL + && (structCanonical.startsWith("J{") || structCanonical.startsWith("A{"))) { + throw new AnalysisException("a join / aggregation entry has no literal carrying form" + + " (their canonical is always constant agnostic), use LITERAL_MODE=NO_LITERAL"); + } + if (keyedEntry && type == PinnedType.JOIN_EXPANSION && literalMode == LiteralMode.WITH_LITERAL) { + throw new AnalysisException("a join expansion entry is keyed by the constant agnostic" + + " condition canonical, use LITERAL_MODE=NO_LITERAL"); + } + String canonical = canonicalFor(structCanonical, literalMode); Review Comment: 认可，已修复（da26dd850df）：pinned 行数只在 filter / join / aggregate 节点查找，S{...} 作为 key 的 pinned 条目永远查不到，现在直接在 SET 校验阶段报错，避免存进去却静默不生效。 ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboUtils.java: ########## @@ -502,27 +477,56 @@ private static InputTableStatisticsInfo buildHboInputTableStatisticsInfo(Physica * @param planToIdMap planToIdMap * @param scanToFilterMap scanToFilterMap * @param curPlanNodeRuntimeStats curPlanNodeRuntimeStats + * @param nodeFingerprints snapshot of nereids plan node id -> hbo fingerprint taken during + * planning (only populated when the struct-info fingerprint is enabled) * @return plan statistics map */ public static Map<PlanNodeAndHash, PlanStatisticsWithInputInfo> genPlanStatisticsMap( Map<Integer, PhysicalPlan> idToPlanMap, Map<PhysicalPlan, Integer> planToIdMap, Map<RelationId, Set<Expression>> scanToFilterMap, - List<TPlanNodeRuntimeStatsItem> curPlanNodeRuntimeStats) { + List<TPlanNodeRuntimeStatsItem> curPlanNodeRuntimeStats, + Map<Integer, String> nodeFingerprints) { Map<PlanNodeAndHash, PlanStatisticsWithInputInfo> outputPlanStatisticsMap = new HashMap<>(); + int missingFingerprints = 0; + int skippedWithoutSnapshot = 0; for (TPlanNodeRuntimeStatsItem nodeStats : curPlanNodeRuntimeStats) { int nodeId = nodeStats.node_id; PhysicalPlan planNode = idToPlanMap.get(nodeId); - if (planNode != null) { - PlanStatistics curPlanStatistics = PlanStatistics.buildFromStatsItem( - nodeStats, planNode, scanToFilterMap); - InputTableStatisticsInfo inputTableStatisticsInfo = buildHboInputTableStatisticsInfo( - planNode, planToIdMap, scanToFilterMap, curPlanNodeRuntimeStats); - Optional<String> hash = inputTableStatisticsInfo.getHash(); - PlanNodeAndHash planNodeAndHash = new PlanNodeAndHash((AbstractPlan) planNode, hash); - PlanStatisticsWithInputInfo planHashWithInputInfo = new PlanStatisticsWithInputInfo( - nodeId, curPlanStatistics, inputTableStatisticsInfo); - outputPlanStatisticsMap.put(planNodeAndHash, planHashWithInputInfo); + if (planNode == null) { + continue; + } + PlanStatistics curPlanStatistics = PlanStatistics.buildFromStatsItem( + nodeStats, planNode, scanToFilterMap); + InputTableStatisticsInfo inputTableStatisticsInfo = buildHboInputTableStatisticsInfo( + planNode, planToIdMap, scanToFilterMap, curPlanNodeRuntimeStats); + Optional<String> hash; + if (nodeFingerprints != null && nodeFingerprints.containsKey(nodeId)) { + hash = Optional.ofNullable(nodeFingerprints.get(nodeId)); + } else { + // no planning-time snapshot for this node: skip it + skippedWithoutSnapshot++; + if (nodeFingerprints != null && !nodeFingerprints.isEmpty()) { + missingFingerprints++; + } + continue; } + if (!hash.isPresent()) { + continue; + } + PlanNodeAndHash planNodeAndHash = new PlanNodeAndHash((AbstractPlan) planNode, hash); + PlanStatisticsWithInputInfo planHashWithInputInfo = new PlanStatisticsWithInputInfo( + nodeId, curPlanStatistics, inputTableStatisticsInfo); + outputPlanStatisticsMap.put(planNodeAndHash, planHashWithInputInfo); + } + if (missingFingerprints > 0) { + LOG.warn("hbo struct fingerprint snapshot missing for {} of {} plan nodes", Review Comment: 认可，已修复（da26dd850df）：规划期快照现在会为被有意排除的节点（如 sort/topn，其 group 不受支持）记录 null fingerprint，发布路径只对快照里完全没有的节点告警。 -- 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]
