englefly commented on code in PR #67549:
URL: https://github.com/apache/doris/pull/67549#discussion_r4240221848
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboStatsCalculator.java:
##########
@@ -27,17 +29,36 @@
import org.apache.doris.nereids.trees.plans.algebra.Aggregate;
import org.apache.doris.nereids.trees.plans.algebra.Filter;
import org.apache.doris.nereids.trees.plans.algebra.Join;
-import org.apache.doris.statistics.hbo.PlanStatistics;
+import org.apache.doris.nereids.trees.plans.algebra.OlapScan;
import org.apache.doris.statistics.hbo.RecentRunsPlanStatistics;
+import org.apache.doris.statistics.hbo.RecentRunsPlanStatisticsEntry;
import org.apache.doris.statistics.model.Statistics;
+import java.util.Locale;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
/**
* StatsCalculator by using hbo plan stats. to do estimation.
*/
public class HboStatsCalculator extends StatsCalculator {
+ /** Pinned lookup order for filter roots: exact form first, then the
constant agnostic form. */
+ private static final GroupStructInfo.LiteralMode[] FILTER_LOOKUP_MODES = {
+ GroupStructInfo.LiteralMode.WITH_LITERAL,
+ GroupStructInfo.LiteralMode.NO_LITERAL,
+ };
+
+ /**
+ * Lookup modes of a join / aggregation root. Their canonical form is
never keyed with literals
+ * ({@code HBO SET STATISTICS} rejects a {@code J{...}} / {@code A{...}}
struct in WITH_LITERAL
+ * mode), so trying that form would only cost a second struct-info
computation per group without
+ * any chance to match.
+ */
+ private static final GroupStructInfo.LiteralMode[] JOIN_AGG_LOOKUP_MODES =
{
Review Comment:
认可这是设计缺陷：J/A 的 key 目前只有 NO_LITERAL 形式，filter 里的常量差异会撞到同一个 join/aggregate
key。修复需要把 J/A 的 key 扩展为携带 literal 的形式（key 结构 + SET 校验一起改），本轮未做，计划在下一个 commit 处理。
##########
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 +
")";
Review Comment:
认可：expression canonical 目前只用类名与 children，同名参数的 Java/Python UDF 会撞 key。需要在
canonical 中加入函数标识（name/database/signature/id）。本轮未做，计划下一步处理。
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboPlanStatisticsManager.java:
##########
@@ -38,4 +187,415 @@ public HboPlanStatisticsProvider
getHboPlanStatisticsProvider() {
public HboPlanInfoProvider getHboPlanInfoProvider() {
return hboPlanInfoProvider;
}
+
+ /**
+ * Inject (or overwrite) a pinned statistics entry applied unconditionally.
+ * @param fingerprint hbo fingerprint (simplified group struct info sha256)
+ * @param rows output row count that overrides the optimizer estimation
+ * @param structCanonical optional human-readable simplified struct info
canonical string
+ */
+ public void putPinnedPlanStatistics(String fingerprint, long rows, String
structCanonical) {
+ putPinnedPlanStatistics(fingerprint, rows, PinnedType.EXACT,
structCanonical, LiteralMode.NO_LITERAL);
+ }
+
+ /**
+ * Inject (or overwrite) a pinned statistics entry.
+ * @param type how the injected row count is applied (unconditional vs
filter-small guard)
+ * @param structCanonical optional human-readable simplified struct info
canonical string
+ */
+ public void putPinnedPlanStatistics(String fingerprint, long rows,
PinnedType type, String structCanonical,
+ LiteralMode literalMode) {
+ putPinnedStatistics(fingerprint, rows, type, structCanonical, 0,
literalMode);
+ }
+
+ /**
+ * Inject (or overwrite) a pinned statistics entry.
+ *
+ * @param expansion fan-out factor, only for {@link
PinnedType#JOIN_EXPANSION} entries
+ */
+ private void putPinnedStatistics(String fingerprint, long rows, PinnedType
type, String structCanonical,
+ double expansion, LiteralMode literalMode) {
+ // LRU bounded by Config.hbo_pinned_stats_cache_num; pinned entries
are otherwise never
+ // expired automatically and are only removed by HBO DELETE STATISTICS
or eviction
+ long createTimeMs = System.currentTimeMillis();
+ synchronized (pinnedLoadLock) {
+ pinnedPlanStatistics.put(fingerprint,
+ new PinnedHboStatistics(fingerprint, rows, type,
structCanonical, expansion, literalMode,
+ createTimeMs));
+ if (type != PinnedType.JOIN_EXPANSION) {
+ // only row count entries are looked up by relation key: a
join expansion entry is
+ // keyed by the join conditions, is rejected by
getPinnedPlanStatistics, and its
+ // scanless canonical would turn the pre filter into a no-op
for every group
+ registerRelationKey(structCanonical);
+ }
+ // a SET after a failed DELETE re-creates the entry: drop the
deletion intent so a
+ // pending load does not skip the re-created row
+ pendingLoadTombstones.remove(fingerprint);
+ // persist under the same lock as DELETE's DB removal, so
concurrent same-key
+ // SET/DELETE serialize their DB writes in the same order as their
memory updates
+ // and, when the best-effort writes succeed, the stored row
matches the last memory
+ // writer (a failed write only logs a warning and may reappear
after a FE restart)
+ if (persistenceEnabled()) {
+ HboStatisticsStore.persist(fingerprint, rows, type,
structCanonical, expansion, literalMode,
+ createTimeMs);
+ }
+ }
+ }
+
+ /**
+ * The pinned row-count entry of a sub tree. Entries of another kind (a
join expansion factor
+ * keyed by the join conditions) are never returned here, so the two read
paths can not pick up
+ * each other's entry.
+ *
+ * <p>This is the planning hot path, so it only reads the in-memory cache:
the persisted entries
+ * of this FE are loaded by {@link HboPinnedStatisticsLoader} in the
background. Doing it here
+ * would run a DDL and a full table scan on a user query's planning
thread, and the internal
+ * statement it runs would re-enter this manager (see the class javadoc).
+ */
+ public Optional<PinnedHboStatistics> getPinnedPlanStatistics(String
fingerprint) {
+ return
Optional.ofNullable(pinnedPlanStatistics.getIfPresent(fingerprint))
+ .filter(pinned -> !pinned.isExpansion());
+ }
+
+ /**
+ * Whether this FE holds any pinned entry at all (of any kind). The
planning path looks an entry
+ * up by the hbo fingerprint of a plan node, and that fingerprint is
expensive to build (a whole
+ * memo subtree traversal); an empty cache means every such lookup is a
guaranteed miss, so the
+ * caller can skip building the fingerprint. Like {@link
#getPinnedPlanStatistics(String)} this
+ * only reads the in-memory cache and never triggers the background loader.
+ */
+ public boolean hasAnyPinnedStatistics() {
+ return !pinnedPlanStatistics.asMap().isEmpty();
+ }
+
+ /**
+ * Whether a pinned entry could possibly apply to a sub tree which reads
exactly {@code
+ * relationKey} (see {@link
GroupStructInfo.HboStructSummary#getRelationKey()}).
+ *
+ * <p>An entry is keyed by the fingerprint of its struct info, and two
equal struct infos
+ * necessarily read the same tables, so an entry whose relation key
differs can not match: the
+ * caller can skip building the (expensive) fingerprint. The index is an
over approximation -
+ * keys are only ever added, so a removed or evicted entry may keep its
key - which can only make
+ * a lookup happen, never suppress one.
+ */
+ public boolean mayHavePinnedEntryForRelations(String relationKey) {
+ return pinnedEntriesWithoutRelationKey.get() > 0 ||
pinnedRelationKeys.contains(relationKey);
+ }
+
+ /** Register the relation key of an entry's struct info in the lookup
index. */
+ private void registerRelationKey(String structCanonical) {
+ List<HboScanDescriptor> scans =
HboScanDescriptor.parseAll(structCanonical);
+ if (scans.isEmpty()) {
+ // an entry which records no scan at all can not be filtered by a
relation key
+ pinnedEntriesWithoutRelationKey.incrementAndGet();
+ return;
+ }
+ List<String> tables = new ArrayList<>(scans.size());
+ for (HboScanDescriptor scan : scans) {
+ tables.add(scan.getTable());
+ }
+ Collections.sort(tables);
+ pinnedRelationKeys.add(String.join(",", tables));
+ }
+
+ /**
+ * Remove a pinned statistics entry. Learned entries of the same key (if
any) are left
+ * untouched here; callers may invalidate them explicitly.
+ */
+ public void removePinnedPlanStatistics(String fingerprint) {
+ synchronized (pinnedLoadLock) {
+ pinnedPlanStatistics.invalidate(fingerprint);
+ if (persistenceEnabled()) {
+ // the DB removal runs under the lock so a lazy load can never
SELECT the row in
+ // the window between the memory invalidation and its DB
removal (M-1 R1)
+ if (!HboStatisticsStore.delete(fingerprint) &&
!hboPinnedLoaded) {
+ // the row is still stored and a pending/retrying load
would resurrect it
+ // into memory: remember the deletion intent (M-1 R2)
+ pendingLoadTombstones.add(fingerprint);
+ }
+ }
+ }
+ }
+
+ /**
+ * Inject (or overwrite) the fan-out factor of a set of join equality
conditions
+ * ({@code HBO SET STATISTICS ... TYPE=JOIN_EXPANSION}). It is a pinned
entry like every other
+ * hbo statistics entry - same table, same lifecycle - only its value is a
factor instead of a
+ * row count and its key is the join condition fingerprint.
+ *
+ * @param fingerprint sha256 of the canonical equality-condition set
+ * @param expansion measured fan-out factor relative to the larger input
of the join (0.1 keeps
+ * 10% of the larger input, 1000 fans out to 1000 times
it), greater than 0
+ * @param condCanonical the canonical equality-condition set the factor
was measured for
+ */
+ public void putPinnedExpansionStatistics(String fingerprint, double
expansion, String condCanonical) {
+ // a condition key never carries a literal by itself: it is the
constant agnostic form
+ putPinnedStatistics(fingerprint, 0, PinnedType.JOIN_EXPANSION,
condCanonical, expansion,
+ LiteralMode.NO_LITERAL);
+ }
+
+ /**
+ * The pinned join expansion entry of a set of equality conditions, if one
was injected.
+ */
+ public Optional<PinnedHboStatistics> getPinnedExpansion(String
fingerprint) {
+ return
Optional.ofNullable(pinnedPlanStatistics.getIfPresent(fingerprint))
+ .filter(PinnedHboStatistics::isExpansion);
+ }
+
+ /**
+ * Inject a learned entry (used by {@code HBO SET LEARNED STATISTICS}) so
that the learned
+ * lookup path can be exercised without waiting for a real profile
publish. The injected entry
+ * carries no input table statistics and therefore matches by fingerprint
alone.
+ */
+ public void putLearnedPlanStatistics(String fingerprint, long rows, String
structCanonical) {
+ PlanStatistics planStatistics = new PlanStatistics(0, rows, rows,
rows, rows,
+ rows, rows, rows, rows, 0, 0, 1);
+ RecentRunsPlanStatistics runs = new RecentRunsPlanStatistics(
+ Lists.newArrayList(new
RecentRunsPlanStatisticsEntry(planStatistics, Lists.newArrayList())));
+ hboPlanStatisticsProvider.putHboPlanStatsByFingerprint(fingerprint,
runs);
+ if (structCanonical != null && !structCanonical.isEmpty()) {
+ learnedStructCanonical.put(fingerprint, structCanonical);
+ }
+ }
+
+ /** The struct info a learned entry was injected with, when it was
injected with one. */
+ public Optional<String> getLearnedStructCanonical(String fingerprint) {
+ return
Optional.ofNullable(learnedStructCanonical.getIfPresent(fingerprint));
+ }
+
+ /** Remove a learned entry by fingerprint. */
+ public void removeLearnedPlanStatistics(String fingerprint) {
+ hboPlanStatisticsProvider.removeHboPlanStats(fingerprint);
+ learnedStructCanonical.invalidate(fingerprint);
+ }
+
+ /**
+ * Remove every pinned entry which can never be applied again: an entry
records the data state it
+ * was measured in (the baseline of the scans of its struct info), and
once a table moved beyond
+ * {@code Config.hbo_row_count_change_ratio} (or was dropped) the entry is
not applied any more
+ * (see {@link HboStructFreshness}); it only keeps a cache slot until it
is removed here. Entries
+ * which are merely {@code drifted} (the data moved, but within the
tolerance) are kept: they are
+ * still applied.
+ *
+ * @param olderThanMillis when non negative, entries whose state cannot be
determined (their
+ * struct info records no comparable state, e.g. it
prunes partitions or
+ * does not resolve to existing tables) are removed
as well, but only when
+ * they are older than this many milliseconds; a
negative value keeps them
+ * @return the number of removed entries
+ */
+ public int deleteStalePinnedPlanStatistics(long olderThanMillis) {
+ long now = System.currentTimeMillis();
+ int removed = 0;
+ for (PinnedHboStatistics pinned :
getAllPinnedPlanStatistics().values()) {
+ // an expansion entry is keyed by join conditions and carries no
data state, so it can
+ // neither go stale nor be judged by age: OLDER_THAN must not
silently drop it
+ if (pinned.isExpansion()) {
+ continue;
+ }
+ HboStructFreshness freshness =
HboStructFreshness.of(pinned.getStructCanonical());
+ boolean obsolete = freshness.isStale()
+ || (olderThanMillis >= 0 && freshness.isUnknown()
+ && now - pinned.getCreateTime() > olderThanMillis);
+ if (obsolete) {
+ removePinnedPlanStatistics(pinned.getFingerprint());
+ removed++;
+ }
+ }
+ return removed;
+ }
+
+ /**
+ * Every pinned entry of this FE, after making sure the persisted ones
have been loaded.
+ *
+ * <p>Unlike the planning path ({@link #getPinnedPlanStatistics}) this is
only used by the admin
+ * statements ({@code HBO SHOW} / {@code HBO DELETE STALE STATISTICS}),
which are expected to wait
+ * for the entries of the internal table and to report what is stored.
Loading here cannot
+ * re-enter the way it could before the load moved to {@link
HboPinnedStatisticsLoader}: the
+ * internal statement a load runs is planned by this FE, and its planning
reads the pinned cache
+ * without triggering a load any more.
+ */
+ public Map<String, PinnedHboStatistics> getAllPinnedPlanStatistics() {
+ ensurePinnedLoaded();
+ return Collections.unmodifiableMap(pinnedPlanStatistics.asMap());
+ }
+
+ /**
+ * Start loading the persisted pinned statistics of this FE in the
background. Called once by
+ * {@code Env} while the daemon threads are started; a FE which never
loads them (the internal
+ * schema is not ready, or persistence is turned off) simply plans without
the persisted entries.
+ */
+ public void startPinnedStatisticsLoader() {
+ synchronized (pinnedLoadLock) {
+ if (loader != null && loader.isAlive()) {
+ // the daemon threads are started again when a FE is promoted;
a second loader would
+ // only double the retry attempts
+ return;
+ }
+ loader = new HboPinnedStatisticsLoader(this);
+ loader.start();
+ }
+ LOG.info("started the hbo pinned statistics loader");
+ }
+
+ /** Whether the persisted pinned statistics of this FE have been loaded
(or are not needed). */
+ public boolean isPinnedStatisticsLoaded() {
+ return hboPinnedLoaded || !FeConstants.enableInternalSchemaDb;
+ }
+
+ /** Whether the loader has nothing left to do. */
+ boolean needsPinnedStatisticsLoad() {
+ return FeConstants.enableInternalSchemaDb && !hboPinnedLoaded;
+ }
+
+ /** Load the persisted pinned statistics once; a failure is retried by the
loader. */
+ void loadPinnedStatistics() {
+ ensurePinnedLoaded();
+ }
+
+ private void ensurePinnedLoaded() {
+ if (hboPinnedLoaded || !persistenceEnabled()) {
+ return;
+ }
+ synchronized (pinnedLoadLock) {
+ if (hboPinnedLoaded || !persistenceEnabled()) {
+ return;
+ }
+ long now = System.currentTimeMillis();
+ if (now - lastLoadFailedMs < LOAD_RETRY_INTERVAL_MS) {
+ return;
+ }
+ List<PinnedHboStatistics> loaded = HboStatisticsStore.loadAll();
+ if (loaded == null) {
+ lastLoadFailedMs = now;
+ return;
+ }
+ List<String> suppressedByTombstone = new ArrayList<>();
+ for (PinnedHboStatistics pinned : loaded) {
+ // entries injected while persistence was off, or concurrently
with the load,
+ // must win over the stored snapshot, and entries deleted
while the load was
+ // still pending must not be resurrected: the in-memory cache
stays authoritative
+ if (pendingLoadTombstones.contains(pinned.getFingerprint())) {
+ suppressedByTombstone.add(pinned.getFingerprint());
+ } else {
+
pinnedPlanStatistics.asMap().putIfAbsent(pinned.getFingerprint(), pinned);
+ if (pinned.getType() != PinnedType.JOIN_EXPANSION) {
+ registerRelationKey(pinned.getStructCanonical());
+ }
+ }
+ }
+ hboPinnedLoaded = true;
+ if (!suppressedByTombstone.isEmpty()) {
+ // the load succeeded, so the internal table is reachable:
retry the failed DB
+ // removals (idempotent) for exactly the rows that were seen
in the loaded
+ // snapshot, so they do not reappear after this FE restarts.
Rows another FE
+ // created after the snapshot are not touched; rows it created
before the
+ // snapshot may be removed (see class javadoc)
+ int failed = 0;
+ for (String fingerprint : suppressedByTombstone) {
+ if (!HboStatisticsStore.delete(fingerprint)) {
Review Comment:
认可（P1）：DELETE 回放时 fingerprint-only 墓碑会压掉它之后更晚的 SET。需要给墓碑带上时间/版本，只压制更早的
SET。本轮未做，计划下一步处理。
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboPlanStatisticsManager.java:
##########
@@ -38,4 +187,415 @@ public HboPlanStatisticsProvider
getHboPlanStatisticsProvider() {
public HboPlanInfoProvider getHboPlanInfoProvider() {
return hboPlanInfoProvider;
}
+
+ /**
+ * Inject (or overwrite) a pinned statistics entry applied unconditionally.
+ * @param fingerprint hbo fingerprint (simplified group struct info sha256)
+ * @param rows output row count that overrides the optimizer estimation
+ * @param structCanonical optional human-readable simplified struct info
canonical string
+ */
+ public void putPinnedPlanStatistics(String fingerprint, long rows, String
structCanonical) {
+ putPinnedPlanStatistics(fingerprint, rows, PinnedType.EXACT,
structCanonical, LiteralMode.NO_LITERAL);
+ }
+
+ /**
+ * Inject (or overwrite) a pinned statistics entry.
+ * @param type how the injected row count is applied (unconditional vs
filter-small guard)
+ * @param structCanonical optional human-readable simplified struct info
canonical string
+ */
+ public void putPinnedPlanStatistics(String fingerprint, long rows,
PinnedType type, String structCanonical,
+ LiteralMode literalMode) {
+ putPinnedStatistics(fingerprint, rows, type, structCanonical, 0,
literalMode);
+ }
+
+ /**
+ * Inject (or overwrite) a pinned statistics entry.
+ *
+ * @param expansion fan-out factor, only for {@link
PinnedType#JOIN_EXPANSION} entries
+ */
+ private void putPinnedStatistics(String fingerprint, long rows, PinnedType
type, String structCanonical,
+ double expansion, LiteralMode literalMode) {
+ // LRU bounded by Config.hbo_pinned_stats_cache_num; pinned entries
are otherwise never
+ // expired automatically and are only removed by HBO DELETE STATISTICS
or eviction
+ long createTimeMs = System.currentTimeMillis();
+ synchronized (pinnedLoadLock) {
+ pinnedPlanStatistics.put(fingerprint,
+ new PinnedHboStatistics(fingerprint, rows, type,
structCanonical, expansion, literalMode,
+ createTimeMs));
+ if (type != PinnedType.JOIN_EXPANSION) {
+ // only row count entries are looked up by relation key: a
join expansion entry is
+ // keyed by the join conditions, is rejected by
getPinnedPlanStatistics, and its
+ // scanless canonical would turn the pre filter into a no-op
for every group
+ registerRelationKey(structCanonical);
+ }
+ // a SET after a failed DELETE re-creates the entry: drop the
deletion intent so a
+ // pending load does not skip the re-created row
+ pendingLoadTombstones.remove(fingerprint);
+ // persist under the same lock as DELETE's DB removal, so
concurrent same-key
+ // SET/DELETE serialize their DB writes in the same order as their
memory updates
+ // and, when the best-effort writes succeed, the stored row
matches the last memory
+ // writer (a failed write only logs a warning and may reappear
after a FE restart)
+ if (persistenceEnabled()) {
+ HboStatisticsStore.persist(fingerprint, rows, type,
structCanonical, expansion, literalMode,
+ createTimeMs);
+ }
+ }
+ }
+
+ /**
+ * The pinned row-count entry of a sub tree. Entries of another kind (a
join expansion factor
+ * keyed by the join conditions) are never returned here, so the two read
paths can not pick up
+ * each other's entry.
+ *
+ * <p>This is the planning hot path, so it only reads the in-memory cache:
the persisted entries
+ * of this FE are loaded by {@link HboPinnedStatisticsLoader} in the
background. Doing it here
+ * would run a DDL and a full table scan on a user query's planning
thread, and the internal
+ * statement it runs would re-enter this manager (see the class javadoc).
+ */
+ public Optional<PinnedHboStatistics> getPinnedPlanStatistics(String
fingerprint) {
+ return
Optional.ofNullable(pinnedPlanStatistics.getIfPresent(fingerprint))
+ .filter(pinned -> !pinned.isExpansion());
+ }
+
+ /**
+ * Whether this FE holds any pinned entry at all (of any kind). The
planning path looks an entry
+ * up by the hbo fingerprint of a plan node, and that fingerprint is
expensive to build (a whole
+ * memo subtree traversal); an empty cache means every such lookup is a
guaranteed miss, so the
+ * caller can skip building the fingerprint. Like {@link
#getPinnedPlanStatistics(String)} this
+ * only reads the in-memory cache and never triggers the background loader.
+ */
+ public boolean hasAnyPinnedStatistics() {
+ return !pinnedPlanStatistics.asMap().isEmpty();
+ }
+
+ /**
+ * Whether a pinned entry could possibly apply to a sub tree which reads
exactly {@code
+ * relationKey} (see {@link
GroupStructInfo.HboStructSummary#getRelationKey()}).
+ *
+ * <p>An entry is keyed by the fingerprint of its struct info, and two
equal struct infos
+ * necessarily read the same tables, so an entry whose relation key
differs can not match: the
+ * caller can skip building the (expensive) fingerprint. The index is an
over approximation -
+ * keys are only ever added, so a removed or evicted entry may keep its
key - which can only make
+ * a lookup happen, never suppress one.
+ */
+ public boolean mayHavePinnedEntryForRelations(String relationKey) {
+ return pinnedEntriesWithoutRelationKey.get() > 0 ||
pinnedRelationKeys.contains(relationKey);
+ }
+
+ /** Register the relation key of an entry's struct info in the lookup
index. */
+ private void registerRelationKey(String structCanonical) {
+ List<HboScanDescriptor> scans =
HboScanDescriptor.parseAll(structCanonical);
+ if (scans.isEmpty()) {
+ // an entry which records no scan at all can not be filtered by a
relation key
+ pinnedEntriesWithoutRelationKey.incrementAndGet();
+ return;
+ }
+ List<String> tables = new ArrayList<>(scans.size());
+ for (HboScanDescriptor scan : scans) {
+ tables.add(scan.getTable());
+ }
+ Collections.sort(tables);
+ pinnedRelationKeys.add(String.join(",", tables));
+ }
+
+ /**
+ * Remove a pinned statistics entry. Learned entries of the same key (if
any) are left
+ * untouched here; callers may invalidate them explicitly.
+ */
+ public void removePinnedPlanStatistics(String fingerprint) {
+ synchronized (pinnedLoadLock) {
+ pinnedPlanStatistics.invalidate(fingerprint);
+ if (persistenceEnabled()) {
+ // the DB removal runs under the lock so a lazy load can never
SELECT the row in
+ // the window between the memory invalidation and its DB
removal (M-1 R1)
+ if (!HboStatisticsStore.delete(fingerprint) &&
!hboPinnedLoaded) {
+ // the row is still stored and a pending/retrying load
would resurrect it
+ // into memory: remember the deletion intent (M-1 R2)
+ pendingLoadTombstones.add(fingerprint);
+ }
+ }
+ }
+ }
+
+ /**
+ * Inject (or overwrite) the fan-out factor of a set of join equality
conditions
+ * ({@code HBO SET STATISTICS ... TYPE=JOIN_EXPANSION}). It is a pinned
entry like every other
+ * hbo statistics entry - same table, same lifecycle - only its value is a
factor instead of a
+ * row count and its key is the join condition fingerprint.
+ *
+ * @param fingerprint sha256 of the canonical equality-condition set
+ * @param expansion measured fan-out factor relative to the larger input
of the join (0.1 keeps
+ * 10% of the larger input, 1000 fans out to 1000 times
it), greater than 0
+ * @param condCanonical the canonical equality-condition set the factor
was measured for
+ */
+ public void putPinnedExpansionStatistics(String fingerprint, double
expansion, String condCanonical) {
+ // a condition key never carries a literal by itself: it is the
constant agnostic form
+ putPinnedStatistics(fingerprint, 0, PinnedType.JOIN_EXPANSION,
condCanonical, expansion,
+ LiteralMode.NO_LITERAL);
+ }
+
+ /**
+ * The pinned join expansion entry of a set of equality conditions, if one
was injected.
+ */
+ public Optional<PinnedHboStatistics> getPinnedExpansion(String
fingerprint) {
+ return
Optional.ofNullable(pinnedPlanStatistics.getIfPresent(fingerprint))
+ .filter(PinnedHboStatistics::isExpansion);
+ }
+
+ /**
+ * Inject a learned entry (used by {@code HBO SET LEARNED STATISTICS}) so
that the learned
+ * lookup path can be exercised without waiting for a real profile
publish. The injected entry
+ * carries no input table statistics and therefore matches by fingerprint
alone.
+ */
+ public void putLearnedPlanStatistics(String fingerprint, long rows, String
structCanonical) {
+ PlanStatistics planStatistics = new PlanStatistics(0, rows, rows,
rows, rows,
+ rows, rows, rows, rows, 0, 0, 1);
+ RecentRunsPlanStatistics runs = new RecentRunsPlanStatistics(
+ Lists.newArrayList(new
RecentRunsPlanStatisticsEntry(planStatistics, Lists.newArrayList())));
+ hboPlanStatisticsProvider.putHboPlanStatsByFingerprint(fingerprint,
runs);
+ if (structCanonical != null && !structCanonical.isEmpty()) {
+ learnedStructCanonical.put(fingerprint, structCanonical);
+ }
+ }
+
+ /** The struct info a learned entry was injected with, when it was
injected with one. */
+ public Optional<String> getLearnedStructCanonical(String fingerprint) {
+ return
Optional.ofNullable(learnedStructCanonical.getIfPresent(fingerprint));
+ }
+
+ /** Remove a learned entry by fingerprint. */
+ public void removeLearnedPlanStatistics(String fingerprint) {
+ hboPlanStatisticsProvider.removeHboPlanStats(fingerprint);
+ learnedStructCanonical.invalidate(fingerprint);
+ }
+
+ /**
+ * Remove every pinned entry which can never be applied again: an entry
records the data state it
+ * was measured in (the baseline of the scans of its struct info), and
once a table moved beyond
+ * {@code Config.hbo_row_count_change_ratio} (or was dropped) the entry is
not applied any more
+ * (see {@link HboStructFreshness}); it only keeps a cache slot until it
is removed here. Entries
+ * which are merely {@code drifted} (the data moved, but within the
tolerance) are kept: they are
+ * still applied.
+ *
+ * @param olderThanMillis when non negative, entries whose state cannot be
determined (their
+ * struct info records no comparable state, e.g. it
prunes partitions or
+ * does not resolve to existing tables) are removed
as well, but only when
+ * they are older than this many milliseconds; a
negative value keeps them
+ * @return the number of removed entries
+ */
+ public int deleteStalePinnedPlanStatistics(long olderThanMillis) {
+ long now = System.currentTimeMillis();
+ int removed = 0;
+ for (PinnedHboStatistics pinned :
getAllPinnedPlanStatistics().values()) {
+ // an expansion entry is keyed by join conditions and carries no
data state, so it can
+ // neither go stale nor be judged by age: OLDER_THAN must not
silently drop it
+ if (pinned.isExpansion()) {
+ continue;
+ }
+ HboStructFreshness freshness =
HboStructFreshness.of(pinned.getStructCanonical());
+ boolean obsolete = freshness.isStale()
+ || (olderThanMillis >= 0 && freshness.isUnknown()
+ && now - pinned.getCreateTime() > olderThanMillis);
+ if (obsolete) {
+ removePinnedPlanStatistics(pinned.getFingerprint());
+ removed++;
+ }
+ }
+ return removed;
+ }
+
+ /**
+ * Every pinned entry of this FE, after making sure the persisted ones
have been loaded.
+ *
+ * <p>Unlike the planning path ({@link #getPinnedPlanStatistics}) this is
only used by the admin
+ * statements ({@code HBO SHOW} / {@code HBO DELETE STALE STATISTICS}),
which are expected to wait
+ * for the entries of the internal table and to report what is stored.
Loading here cannot
+ * re-enter the way it could before the load moved to {@link
HboPinnedStatisticsLoader}: the
+ * internal statement a load runs is planned by this FE, and its planning
reads the pinned cache
+ * without triggering a load any more.
+ */
+ public Map<String, PinnedHboStatistics> getAllPinnedPlanStatistics() {
+ ensurePinnedLoaded();
+ return Collections.unmodifiableMap(pinnedPlanStatistics.asMap());
+ }
+
+ /**
+ * Start loading the persisted pinned statistics of this FE in the
background. Called once by
+ * {@code Env} while the daemon threads are started; a FE which never
loads them (the internal
+ * schema is not ready, or persistence is turned off) simply plans without
the persisted entries.
+ */
+ public void startPinnedStatisticsLoader() {
+ synchronized (pinnedLoadLock) {
+ if (loader != null && loader.isAlive()) {
+ // the daemon threads are started again when a FE is promoted;
a second loader would
+ // only double the retry attempts
+ return;
+ }
+ loader = new HboPinnedStatisticsLoader(this);
+ loader.start();
+ }
+ LOG.info("started the hbo pinned statistics loader");
+ }
+
+ /** Whether the persisted pinned statistics of this FE have been loaded
(or are not needed). */
+ public boolean isPinnedStatisticsLoaded() {
+ return hboPinnedLoaded || !FeConstants.enableInternalSchemaDb;
+ }
+
+ /** Whether the loader has nothing left to do. */
+ boolean needsPinnedStatisticsLoad() {
+ return FeConstants.enableInternalSchemaDb && !hboPinnedLoaded;
+ }
+
+ /** Load the persisted pinned statistics once; a failure is retried by the
loader. */
+ void loadPinnedStatistics() {
+ ensurePinnedLoaded();
+ }
+
+ private void ensurePinnedLoaded() {
+ if (hboPinnedLoaded || !persistenceEnabled()) {
+ return;
+ }
+ synchronized (pinnedLoadLock) {
+ if (hboPinnedLoaded || !persistenceEnabled()) {
+ return;
+ }
+ long now = System.currentTimeMillis();
+ if (now - lastLoadFailedMs < LOAD_RETRY_INTERVAL_MS) {
+ return;
+ }
+ List<PinnedHboStatistics> loaded = HboStatisticsStore.loadAll();
+ if (loaded == null) {
+ lastLoadFailedMs = now;
+ return;
+ }
+ List<String> suppressedByTombstone = new ArrayList<>();
+ for (PinnedHboStatistics pinned : loaded) {
+ // entries injected while persistence was off, or concurrently
with the load,
+ // must win over the stored snapshot, and entries deleted
while the load was
+ // still pending must not be resurrected: the in-memory cache
stays authoritative
+ if (pendingLoadTombstones.contains(pinned.getFingerprint())) {
+ suppressedByTombstone.add(pinned.getFingerprint());
+ } else {
+
pinnedPlanStatistics.asMap().putIfAbsent(pinned.getFingerprint(), pinned);
Review Comment:
认可：有界缓存的 loadAll 没有保持“当前活动条目”，hbo_pinned_stats_cache_num 很小时重启后可能丢掉在用条目。需要在
loadAll 里按创建时间保留活动条目。本轮未做，计划下一步处理。
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboScanDescriptor.java:
##########
@@ -0,0 +1,342 @@
+// 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.Env;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.nereids.trees.plans.algebra.OlapScan;
+import org.apache.doris.rpc.RpcException;
+import org.apache.doris.statistics.analysis.TableStatsMeta;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The data state of one scan of a hbo struct info: the table it reads, the
table visible version,
+ * the number of rows the scan reads (the sum of the row counts of its
selected partitions) and which
+ * part of the table partitions it selects.
+ *
+ * <p>These numbers are the <b>baseline</b> a hbo entry is judged against. The
scan token of a
+ * canonical struct info carries them as an annotation
+ * ({@code S{db.t,v3,r1000,p1/5}}), so a user copies the state the entry was
measured in together
+ * with the struct info, and the read side compares them with the state of the
current query to
+ * decide whether the injected row count may still be applied (see {@link
HboStructFreshness}).
+ *
+ * <p>None of these numbers takes part in the fingerprint:
+ * {@link GroupStructInfo#stripScanBaseline} removes the whole baseline of
every scan token before
+ * the sha256 is taken, so a table which simply keeps growing neither changes
the key of an entry nor
+ * makes it unmatchable; whether the entry may still be applied is a read side
decision.
+ *
+ * <p>A number which could not be determined is omitted from the token (and
parsed back as
+ * {@link #UNKNOWN}), e.g. the row count of a table whose row count was never
reported by any
+ * backend: such an entry cannot be compared by rows any more and falls back
to the version check.
+ */
+public class HboScanDescriptor {
+ /** A number which could not be determined; a row count of 0 is a known,
meaningful value. */
+ public static final long UNKNOWN = -1;
+
+ private final String table;
+ private final long visibleVersion;
+ private final long scanRows;
+ private final RowsKind rowsKind;
+ private final long selectedPartitions;
+ private final long totalPartitions;
+
+ HboScanDescriptor(String table, long visibleVersion, long scanRows,
RowsKind rowsKind,
+ long selectedPartitions, long totalPartitions) {
+ this.table = table;
+ this.visibleVersion = visibleVersion;
+ this.scanRows = scanRows;
+ this.rowsKind = rowsKind;
+ this.selectedPartitions = selectedPartitions;
+ this.totalPartitions = totalPartitions;
+ }
+
+ /**
+ * How the row count of a scan was obtained.
+ */
+ public enum RowsKind {
+ /**
+ * Every selected partition reported its own row count, so the number
is the physical size of
+ * the data the scan reads.
+ */
+ MEASURED,
+ /**
+ * The row count of at least one selected partition is not available,
so the number is
+ * derived from the table (its last reported row count, or the
statistics of the last
+ * analysis plus the rows loaded since). Two such numbers are
comparable with each other, but
+ * a number of this kind must never be compared with a {@link
#MEASURED} one: the derivation
+ * can be off by a large factor, e.g. for the partitions of a skewed
table.
+ */
+ ESTIMATED,
+ /** No row count at all (the table was neither analyzed nor reported
by a backend). */
+ UNKNOWN
+ }
+
+ /** A row count and how it was obtained. */
+ public static class Rows {
+ private static final Rows UNKNOWN_ROWS = new Rows(UNKNOWN,
RowsKind.UNKNOWN);
+
+ private final long rows;
+ private final RowsKind kind;
+
+ Rows(long rows, RowsKind kind) {
+ this.rows = rows;
+ this.kind = kind;
+ }
+
+ public long getRows() {
+ return rows;
+ }
+
+ public RowsKind getKind() {
+ return kind;
+ }
+
+ public boolean isKnown() {
+ return kind != RowsKind.UNKNOWN;
+ }
+ }
+
+ /** The state of a scan node: which table it reads, and how much of it. */
+ public static HboScanDescriptor of(OlapScan scan) throws RpcException {
+ OlapTable table = scan.getTable();
+ List<Long> selectedPartitionIds = scan.getSelectedPartitionIds();
+ Rows rows = rowsOf(table, selectedPartitionIds,
scan.getSelectedIndexId());
+ return new HboScanDescriptor(table.getNameWithFullQualifiers(),
table.getVisibleVersion(),
Review Comment:
认可：scan key 只有库表名与基线，没有 OlapTable ID，drop + recreate 同名表会误命中。需要把 tableId 纳入
scan 标识与 canonical。这是格式变更，本轮未做，计划下一步处理。
--
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]
