github-actions[bot] commented on code in PR #69322: URL: https://github.com/apache/doris/pull/69322#discussion_r4336321641
########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboJoinConditions.java: ########## @@ -0,0 +1,206 @@ +// 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.expressions.NullSafeEqual; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.plans.JoinType; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.algebra.Join; + +import com.google.common.hash.Hashing; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.TreeSet; + +/** + * 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>The join type and the residual (non equality) conditions take part in the key next to the hash + * (equality) conjuncts, because the injected factor replaces the output rows of the whole join: an + * inner join which matches nothing and a left outer join of the same columns produce very different + * row counts, and a selective residual predicate changes the output by orders of magnitude. The + * shape of both sub trees is still excluded, so one injected entry stays reusable by every query + * which joins the same columns with the same join type and the same residual conditions. + * + * <p>The canonical form is {@code JE{<join type>;EqualTo(...);...}} with the pairs and the residual + * conditions 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. A residual condition is rendered with its literal in both modes, + * because it decides how many partner rows a row finds ({@code b > c + 1} is not {@code b > c + 2}). + */ +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 conditions of a join, empty when the join has no expansion key at all. + * + * <p>An expansion entry replaces the output of a join for every query which matches it, and its + * fingerprint carries no data state to guard it - so the operands have to name the columns of the + * tables themselves. An operand which is not a direct slot of a catalog table (a computed column, an + * aggregate output, a cast) cannot name its table; such a join has no key, so it is skipped instead + * of being given a key which two different table pairs with the same aliases would share. A null safe + * equality ({@code <=>}) is a hash conjunct as well and takes part like a plain equality. + */ + public static Optional<String> canonicalOf(Join join) { + return keyParts(join).map(parts -> PREFIX + String.join(SEP, parts) + "}"); + } + + /** + * The key of the expansion entry of a join: the sha256 of its structural identity, empty when the + * join has no expansion key (see {@link #canonicalOf}). The canonical string is the readable form of + * the same parts: it is printed by EXPLAIN and stored as the struct of the entry, and the key is not + * derived from the text. + */ + public static Optional<String> fingerprintOf(Join join) { + return keyParts(join).map(parts -> { + ShapeKeyWriter key = new ShapeKeyWriter(); + key.tag(0); + for (String part : parts) { + key.text(part); + } + return Hashing.sha256().hashBytes(key.toByteArray()).toString(); + }); + } + + /** + * The identity of a join expansion: its type, the equality pairs, the residual conditions and the + * columns of the tables the join reads. The identity is the same while a group is estimated (its + * children are group references there) and while the chosen plan is printed, because the operands are + * read from the slots themselves. + */ + private static Optional<List<String>> keyParts(Join join) { + Set<String> operands = new TreeSet<>(); + List<String> pairs = new ArrayList<>(); + for (Expression conjunct : join.getHashJoinConjuncts()) { + if (!(conjunct instanceof EqualTo) && !(conjunct instanceof NullSafeEqual)) { + return Optional.empty(); + } + Optional<String> left = operandIdentity(conjunct.child(0)); + Optional<String> right = operandIdentity(conjunct.child(1)); + if (!left.isPresent() || !right.isPresent()) { + return Optional.empty(); + } + operands.add(left.get()); + operands.add(right.get()); + // equality is commutative: the operands are sorted so `a = b` and `b = a` collapse + String name = conjunct instanceof NullSafeEqual ? "NullSafeEqual" : "EqualTo"; + pairs.add(left.get().compareTo(right.get()) <= 0 + ? name + "(col(" + left.get() + "),col(" + right.get() + "))" + : name + "(col(" + right.get() + "),col(" + left.get() + "))"); + } + if (pairs.isEmpty()) { + return Optional.empty(); + } + pairs.sort(String::compareTo); + List<String> residuals = new ArrayList<>(); + for (Expression conjunct : join.getOtherJoinConjuncts()) { + // a residual decides how many partner rows a row finds (`b > u.b + 1` is not `u.b + 2`), so + // its literals are part of the key and are never folded - and every slot it reads has to name + // its table like the equality operands do + Set<String> residualOperands = new TreeSet<>(); + boolean supported = supportsResidual(conjunct, residualOperands); + if (!supported) { + return Optional.empty(); + } + operands.addAll(residualOperands); + residuals.add(GroupStructInfo.normalizeExpression(conjunct, + GroupStructInfo.LiteralMode.WITH_LITERAL)); + } + residuals.sort(String::compareTo); + List<String> parts = new ArrayList<>(1 + pairs.size() + residuals.size() + 1); + parts.add(String.valueOf(join.getJoinType())); + parts.addAll(pairs); + parts.addAll(residuals); + parts.add("operands:[" + String.join(",", operands) + "]"); + if (join.getJoinType() != JoinType.INNER_JOIN) { + // an outer join preserves one side, so the side a relation is on is part of the key: it names + // the columns each side produces, which are the same while a group is estimated and while the + // chosen plan is printed (the slots carry the tables they come from) + Plan joinPlan = (Plan) join; + parts.add("sides:[" + sideOperands(joinPlan.child(0)) + "|" + sideOperands(joinPlan.child(1)) + "]"); + } + return Optional.of(parts); + } + + /** The columns one side of an outer join produces, sorted: which side a relation is on is identity. */ + private static String sideOperands(Plan side) { + Set<String> identities = new TreeSet<>(); + for (Slot slot : side.getOutput()) { + if (slot instanceof SlotReference) { + operandIdentity(slot).ifPresent(identities::add); + } + } + return String.join(",", identities); + } + + /** Whether every slot of a residual condition names its table (collecting those identities). */ + private static boolean supportsResidual(Expression residual, Set<String> operands) { + boolean[] supported = {true}; + residual.foreachUp(child -> { + if (!(child instanceof SlotReference)) { + return; + } + Optional<String> identity = operandIdentity((Expression) child); + if (!identity.isPresent()) { + supported[0] = false; + } else { + operands.add(identity.get()); + } + }); + return supported[0]; + } + + /** + * The stable identity of the column an operand reads: the table it belongs to and the name of the + * column. The slot carries both, so the identity does not depend on the alias a query gave the + * relation, on the shape of its qualifier or on where the join is described. + */ + private static Optional<String> operandIdentity(Expression operand) { + if (!(operand instanceof SlotReference)) { + return Optional.empty(); + } + SlotReference slot = (SlotReference) operand; + if (!slot.getOriginalTable().isPresent() || !slot.getOriginalColumn().isPresent()) { + return Optional.empty(); + } + return Optional.of(HboScanDescriptor.qualifiedNameOf(slot.getOriginalTable().get()) + "." Review Comment: [P2] Keep each equality operand bound to its outer-join input `LEFT_OUTER_JOIN(a.k=b.v, Scan(t a), Scan(t b))` and the same join on `a.v=b.k`, selecting all four columns, get identical expansion keys here. For t(k,v)={(1,1),(2,1)}, they produce three versus two rows. Both operands become the original-table names t.k/t.v, the equality sorts them, and `sideOperands` sees the same {t.k,t.v} on both sides. A factor of 1.5 injected for the first therefore estimates three rows for the second as well, with no binding/freshness guard. Preserve the input/occurrence of each operand before canonicalizing an outer self-join, and cover this reversed-column pair. This is distinct from the existing different-table INNER JOIN thread: both inputs here read the same table. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboUtils.java: ########## @@ -502,27 +548,77 @@ 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<>(); + // the fingerprint a plan node was published under, so a second node with the same fingerprint + // can be folded into it (the map above is keyed by the plan node identity as well) + Map<String, PlanNodeAndHash> publishedKeyByHash = 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)) { + String snapshot = nodeFingerprints.get(nodeId); + if (GroupStructInfo.UNSUPPORTED_FINGERPRINT.equals(snapshot)) { + // the node is deliberately excluded from the hbo fingerprint (its group is + // unsupported), so its missing fingerprint is not a diagnostic gap + continue; + } + hash = Optional.ofNullable(snapshot); + } 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); + PlanNodeAndHash publishedKey = publishedKeyByHash.get(hash.get()); + if (publishedKey == null) { + publishedKeyByHash.put(hash.get(), planNodeAndHash); + outputPlanStatisticsMap.put(planNodeAndHash, new PlanStatisticsWithInputInfo( + nodeId, curPlanStatistics, inputTableStatisticsInfo)); + continue; + } + // a second runtime node with the same fingerprint (the two scans of a self join, for + // instance): the provider stores one entry per fingerprint, so the input statistics of both + // nodes are folded here - writing both would let the last write discard the other node's + // rows, and the drift comparison of the published entry expects the rows of every scan + PlanStatisticsWithInputInfo published = outputPlanStatisticsMap.get(publishedKey); + outputPlanStatisticsMap.put(publishedKey, new PlanStatisticsWithInputInfo( + published.getId(), published.getPlanStatistics(), Review Comment: [P2] Keep each learned output count with the inputs that produced it `UnionAll(Filter(k<10, Scan(t)), Filter(k<1000, Scan(t)))` over the same unaliased key table containing k=0..999 can publish two scans under the same S fingerprint. The filters are embedded into their scan operators. If the narrow scan is visited first, this retains its output=10, while `InputTableStatisticsInfo.merge` chooses the other scan's larger input=1000 together with its predicate and RelationId. On the next identical query, `computeFilter`'s scan-key fallback matches that retained k<1000 predicate and freshness accepts 1000 against the table's 1000 rows, then applies the unrelated output=10 to the 1000-row branch. Keep output and corresponding inputs together, using separate recent-run entries for different predicate/input signatures. Unlike the existing ScanRows-population thread, this mismatched pair passes both matching and freshness. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboStatisticsStore.java: ########## @@ -0,0 +1,295 @@ +// 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.Database; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.Config; +import org.apache.doris.common.FeConstants; +import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.statistics.repository.ResultRow; +import org.apache.doris.statistics.util.StatisticsUtil; + +import com.google.common.collect.ImmutableList; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.ArrayList; +import java.util.List; + +/** + * Persistence of manually injected (pinned) hbo statistics into the internal database + * ({@code __internal_schema.hbo_statistics}), controlled by + * {@code Config.hbo_persist_pinned_to_internal_db}. + * + * <p>The in-memory {@link HboPlanStatisticsManager} stays authoritative on the read path; + * SET/DELETE are written through synchronously (best effort, failures only logged) and a FE + * loads the whole table into memory lazily on first use, so pinned entries survive a FE + * restart. All FEs share the table through the backend storage. + */ +public class HboStatisticsStore { + private static final Logger LOG = LogManager.getLogger(HboStatisticsStore.class); + + private static final String INTERNAL_DB = FeConstants.INTERNAL_DB_NAME; + private static final String TABLE = "hbo_statistics"; + private static final String FULL_QUALIFIED = + InternalCatalog.INTERNAL_CATALOG_NAME + "." + INTERNAL_DB + "." + TABLE; + /** max length (bytes) of the struct_info varchar column */ + private static final int STRUCT_MAX_BYTES = 65533; + + /** creation time of an entry, as a datetime(3) column like the other internal tables */ + private static final String CREATE_TIME_COLUMN = "create_time"; + /** fan-out factor of a PinnedType.JOIN_EXPANSION entry, NULL for a row count entry */ + private static final String EXPANSION_COLUMN = "expansion"; + /** which struct info the entry is keyed by (with_literal / no_literal) */ + private static final String LITERAL_MODE_COLUMN = "literal_mode"; + /** the columns this version writes and reads, checked once per FE (see verifySchema) */ + private static final List<String> COLUMNS = ImmutableList.of("fingerprint", "row_count", "stats_type", + LITERAL_MODE_COLUMN, "struct_info", EXPANSION_COLUMN, CREATE_TIME_COLUMN); + // the DDL and the schema check are done once per FE: the table can only change by a FE upgrade, + // and both cost a round trip to the internal table (a forwarded DDL on a follower FE) + private static volatile boolean tableCreated = false; + private static volatile boolean schemaVerified = false; + + private HboStatisticsStore() { + } + + /** + * Whether a canonical struct info fits the {@code struct_info} column without truncation. + * A truncated struct info would keep its fingerprint but lose its scan tokens (the baseline is + * printed after the conditions), which silently disables both the freshness verdict and the + * relation pre filter for that entry, so such an entry must not be accepted at all. + */ + public static boolean fitsStructColumn(String structCanonical) { + return structCanonical == null + || structCanonical.getBytes(java.nio.charset.StandardCharsets.UTF_8).length <= STRUCT_MAX_BYTES; + } + + /** Create the table if it does not exist, and check that its schema is the current one. */ + public static void ensureTable() throws Exception { + if (!tableCreated) { + StatisticsUtil.execUpdate(createDdl()); + tableCreated = true; + } + if (!schemaVerified) { + verifySchema(); + } + } + + /** + * Check that the table has every column this version reads and writes, by looking at its metadata + * in the FE catalog. A query would be cheaper to write but wrong to interpret: it fails both when + * the table has an old schema and when the internal schema is simply not ready yet, and the two + * need opposite reactions (drop the table vs wait). + * + * <p>A table created by an older FE version (e.g. with a {@code create_time_ms} column) cannot be + * written by this version, and because it is created with {@code CREATE TABLE IF NOT EXISTS} it + * stays broken; the check therefore logs the fix and forgets that the table was "created", so + * that the DDL runs again - and recreates the table - after the operator dropped it. Without + * that, persistence would stay dead until the FE restarts. + */ + private static void verifySchema() throws Exception { + Database db = Env.getCurrentEnv().getInternalCatalog().getDbNullable(INTERNAL_DB); + TableIf table = db == null ? null : db.getTableNullable(TABLE); + if (table == null) { + // the DDL succeeded but the table is not visible yet (a follower waiting for the journal, + // or an internal schema which is not ready): retry later, do not blame the table + throw new IllegalStateException("the internal table " + FULL_QUALIFIED + " is not visible yet"); + } + List<String> missing = new ArrayList<>(); + for (String column : COLUMNS) { + if (table.getColumn(column) == null) { + missing.add(column); + } + } + if (!missing.isEmpty()) { + tableCreated = false; + LOG.warn("the internal table {} has no {} column: it was created by an older FE version and" + + " has to be dropped so that it is recreated (DROP TABLE {})", + FULL_QUALIFIED, missing, FULL_QUALIFIED); + throw new IllegalStateException("the internal table " + FULL_QUALIFIED + " has an old schema"); + } + schemaVerified = true; + } + + /** + * Upsert one pinned entry (UNIQUE KEY fingerprint replaces on conflict). + */ + public static void persist(String fingerprint, long rows, + HboPlanStatisticsManager.PinnedType type, String structCanonical, double expansion, + HboPlanStatisticsManager.LiteralMode literalMode, long createTimeMs) { + try { + ensureTable(); + HboPlanStatisticsManager.PinnedType statsType = type == null + ? HboPlanStatisticsManager.PinnedType.EXACT : type; + // an empty struct_info records no data state: the freshness comparison reports unknown for such + // an entry and the pinned lookup index has no relation key for it, which is why the pinned scope + // of HBO SET requires the canonical string (see HboStatisticsCommand) + String struct = truncateUtf8(structCanonical == null ? "" : structCanonical, STRUCT_MAX_BYTES); + // the fingerprint kind is only known after the entry was applied once, so the stored + // value stays UNKNOWN (it is reported by HBO SHOW from the in-memory entry) + String sql = "INSERT INTO " + FULL_QUALIFIED + + " (`fingerprint`, `row_count`, `stats_type`, `" + LITERAL_MODE_COLUMN + "`, `struct_info`," + + " `" + EXPANSION_COLUMN + "`, `" + CREATE_TIME_COLUMN + "`) VALUES ('" + + StatisticsUtil.escapeSQL(fingerprint) + "', " + rows + ", '" + + StatisticsUtil.escapeSQL(statsType.name().toLowerCase(java.util.Locale.ROOT)) + "', '" + + StatisticsUtil.escapeSQL((literalMode == null + ? HboPlanStatisticsManager.LiteralMode.NO_LITERAL : literalMode) + .name().toLowerCase(java.util.Locale.ROOT)) + + "', '" + StatisticsUtil.escapeSQL(struct) + "', " + + (expansion > 0 ? String.valueOf(expansion) : "NULL") + + ", '" + createTimeLiteral(createTimeMs) + "')"; + StatisticsUtil.execUpdate(sql); + } catch (Exception t) { + LOG.warn("failed to persist hbo pinned statistics for fingerprint {}", fingerprint, t); + } + } + + /** + * Remove the pinned entry a removal judged: the row is deleted only while its create time is not + * newer than {@code generation}, so a row another FE wrote after the judge was read survives. + * + * @return true when the row was removed (or the table had no such row), false when the + * best-effort removal failed (e.g. internal schema not ready yet) + */ + public static boolean delete(String fingerprint, long generation) { + try { + ensureTable(); + // only the row the removal judged is deleted: a row another FE wrote afterwards is a newer + // SET and must survive, because this FE cannot order its removal against that write + String sql = "DELETE FROM " + FULL_QUALIFIED + " WHERE `fingerprint` = '" + + StatisticsUtil.escapeSQL(fingerprint) + "' AND `" + CREATE_TIME_COLUMN + + "` <= '" + createTimeLiteral(generation) + "'"; + StatisticsUtil.execUpdate(sql); Review Comment: [P2] Use a write revision that distinguishes replacements across FEs `create_time` comes from `System.currentTimeMillis()` before the FE-local lock, so it is not a unique generation across FEs. Two concurrent SETs for one fingerprint can capture the same millisecond; after B's INSERT replaces A's row, a conditional stale deletion on A still matches B's acknowledged replacement through this `<=` predicate. Clock skew can likewise give a later replacement a smaller timestamp, and `tombstoneCovers` will suppress that replacement during a pending load too. This violates the stated guarantee that a newer SET on another FE survives removal of the judged entry, even when every SQL operation succeeds. Persist a distinct revision/CAS token for the entry and target that revision when a known entry is removed, rather than ordering independent writes by local wall-clock time. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/GroupStructInfo.java: ########## @@ -0,0 +1,3047 @@ +// 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.TableIf; +import org.apache.doris.nereids.memo.Group; +import org.apache.doris.nereids.memo.GroupExpression; +import org.apache.doris.nereids.properties.OrderKey; +import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.And; +import org.apache.doris.nereids.trees.expressions.ArrayItemReference; +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.Match; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.NullSafeEqual; +import org.apache.doris.nereids.trees.expressions.Or; +import org.apache.doris.nereids.trees.expressions.OrderExpression; +import org.apache.doris.nereids.trees.expressions.SearchExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.WindowExpression; +import org.apache.doris.nereids.trees.expressions.functions.Function; +import org.apache.doris.nereids.trees.expressions.functions.Udf; +import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Lambda; +import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral; +import org.apache.doris.nereids.trees.expressions.literal.DateLiteral; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; +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.algebra.Aggregate; +import org.apache.doris.nereids.trees.plans.algebra.Except; +import org.apache.doris.nereids.trees.plans.algebra.Filter; +import org.apache.doris.nereids.trees.plans.algebra.Generate; +import org.apache.doris.nereids.trees.plans.algebra.Intersect; +import org.apache.doris.nereids.trees.plans.algebra.Join; +import org.apache.doris.nereids.trees.plans.algebra.Limit; +import org.apache.doris.nereids.trees.plans.algebra.OlapScan; +import org.apache.doris.nereids.trees.plans.algebra.Project; +import org.apache.doris.nereids.trees.plans.algebra.Repeat; +import org.apache.doris.nereids.trees.plans.algebra.SetOperation; +import org.apache.doris.nereids.trees.plans.algebra.Sort; +import org.apache.doris.nereids.trees.plans.algebra.TopN; +import org.apache.doris.nereids.trees.plans.algebra.Union; +import org.apache.doris.nereids.trees.plans.algebra.Window; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate; +import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan; +import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows; +import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEAnchor; +import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer; +import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer; +import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute; +import org.apache.doris.nereids.trees.plans.physical.PhysicalFileScan; +import org.apache.doris.nereids.trees.plans.physical.PhysicalGenerate; +import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate; +import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate; +import org.apache.doris.nereids.util.MutableState; +import org.apache.doris.qe.ConnectContext; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.hash.Hashing; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * 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. + * + * <p>This description is always built by the FE from the plan it is looking at, and every string it can + * produce is one of its renderings: {@link #getCanonical(LiteralMode)} is the text EXPLAIN prints and a + * user may paste into {@code HBO SET STATISTICS ... STRUCT='...'}, {@link #getShape(LiteralMode)} is the + * same text without the data state of the scans, and {@link #getFingerprint(LiteralMode)} is the sha256 + * of that shape. Nothing is ever parsed to produce a fingerprint - in particular the text a user pasted + * is only compared with the fingerprint next to it, never taken as its source. + */ +public class GroupStructInfo { + /** + * 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; + /** + * Snapshot value which marks a plan node that is deliberately excluded from the hbo + * fingerprint (its group is unsupported, e.g. a sort / topn). It is not a fingerprint and is + * never hashed: the publish path uses it to tell an expected gap from a lost snapshot. A + * concurrent map can not store a null marker, so an empty string is used. + */ + public static final String UNSUPPORTED_FINGERPRINT = ""; + + /** + * The prefix of the reason a node whose kind the description does not support is reported with, and + * the two halves of the reason a node with an unsupported number of children is reported with. They + * are constants so that the corpus can bind the frozen skip texts to the formula which produces them + * instead of restating the prefix. + */ + static final String UNSUPPORTED_REASON_PREFIX = "unsupported="; + static final String ARITY_REASON_PREFIX = "arity="; + static final String ARITY_REASON_MIDDLE = " of "; + + private static final List<HboScanDescriptor> INVALID_SCANS = Collections.emptyList(); + + /** + * Characters which structure the canonical form and therefore can not appear raw inside a + * literal value (see {@link #encodeLiteralValue}). + */ + private static final String RESERVED_LITERAL_CHARS = "\u0000\\%(){}[],;:|"; + + private static final char[] HEX_DIGITS = "0123456789ABCDEF".toCharArray(); + + /** + * How many groups this query described. A group is described once: the description is memoized on + * the group and composed by its parents, so this number only grows when a new group is reached. + */ + private static final AtomicInteger describedGroupCount = new AtomicInteger(); + + /** + * How many descriptions were walked while materializing a string. A description whose string is + * already materialized is copied instead of being walked again, so this number does not grow for + * the sub tree of an already materialized child. + */ + private static final AtomicInteger materializedGroupCount = new AtomicInteger(); + + /** Shared invalid instance: the group content is not supported by the simplified struct info. */ + public static final GroupStructInfo INVALID = new GroupStructInfo(false, "", null, INVALID_SCANS); + + private static final String SEP = ";"; + + /** + * The header letters whose arguments are part of the identity of their node (see + * {@link Kind#literalsAreIdentity()}): the fold of a pasted struct info copies those headers verbatim, + * because folding one of their literals would make the sha256 of the text differ from the fingerprint + * EXPLAIN printed. The set is derived from the kinds which render the text, so a kind whose literals + * stop being identity cannot be forgotten here. + */ + private static final String PROTECTED_HEADERS = protectedHeaders(); + + /** + * The markers of the headers whose condition collection is sorted per literal mode, i.e. the ones the + * fold of a pasted struct info has to re-sort after it folded their literals (see + * {@code recanonicalizeFoldedConditions}): a filter, an aggregation, a join (which also de-duplicates) + * and the grouping sets of a repeat. + */ + private static final String[] CONDITION_HEADER_MARKERS = {"F{", "A{gb:", "J{", "R{gs:["}; + + /** + * The functions whose value depends on the session time zone although neither their arguments nor + * their result are a date: an epoch value is read in that zone ({@code hour_from_unixtime(t.ts) = 0} + * keeps 1000 rows in one session and 1 row in another although its argument is the same integer). + * A function whose argument or result is a date or a datetime is covered by its type instead (see + * {@link #usesSessionTimeZone}). + */ + private static final List<String> EPOCH_TIME_ZONE_FUNCTIONS = Arrays.asList( + "hour_from_unixtime", "minute_from_unixtime", "second_from_unixtime", + "microsecond_from_unixtime", "unix_timestamp"); + + private static final String NO_LOOKUP_PATH_REASON = + "storage layer aggregate: the read side applies the row count of the aggregation above it"; + + private static final String PARTIAL_STAGE_REASON = + "local (partial) aggregation stage: the read side applies the row count of the final" + + " aggregation above it"; + + 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. It is the primary key of a scan, of a constant-free + * filter shape and of the key the publish path generates for a join / aggregate root; every + * root which is annotated also carries a literal carrying form.</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 + } + + /** + * Whether the description could be built. It is invalid - and the memo caches that answer, so every + * node above the sub tree gets no fingerprint for the whole query - when one of the following holds + * for the sub tree it covers: + * <ul> + * <li>it contains an operator the canonical form does not describe, which the annotation prints as + * {@code unsupported=<Operator>}: a table valued function relation ({@code numbers}, + * {@code s3}), an apply, a recursive union, a partition top-n, ...;</li> + * <li>it reads more scans than {@code hbo_max_scans_per_group}: + * {@code subtree scans=N > hbo_max_scans_per_group=M};</li> + * <li>a table name carries a struct info delimiter ({@code ,} or {@code }}), so the printed token + * could not be read back: {@code delimiter in table name <table>};</li> + * <li>the memo does not look like a canonical form can express it: a unary operator whose group + * expression has another arity, a CTE consumer which has no memo edge to its producer, a + * reused sub tree whose description is not available;</li> + * <li>the data state of a scan cannot be read (an external table whose metadata is not available, + * a cloud rpc failure): {@code no data state for <table> (<cause>)}.</li> + * </ul> + * A node whose group cannot be resolved at all is not described either, which the annotation reports + * as {@code no group for <Operator>}. + */ + private final boolean valid; + /** + * The tree of this description: the information of its own group and the descriptions of the groups + * below it, all as objects. Nothing is rendered while it is built - {@link #render} is the one place + * which knows the grammar - and every node remembers the forms it has rendered, so a parent which is + * rendered later composes the text of a child instead of walking its sub tree again. + */ + private final Node root; + /** 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; + // the forms materialized on first request: the read side only needs a fingerprint, while the + // explain annotation needs the canonical string, so building all of them up front would be wasted + private String canonicalWithLiteral; + private String canonicalNoLiteral; + private String shapeWithLiteral; + private String shapeNoLiteral; + private String fingerprintWithLiteral; + private String fingerprintNoLiteral; + + private GroupStructInfo(boolean valid, String skipReason, Node root, + List<HboScanDescriptor> scans) { + this.valid = valid; + this.skipReason = skipReason; + this.root = root; + this.scans = scans; + } + + private static GroupStructInfo ofNode(Node root) { + List<HboScanDescriptor> scans = new ArrayList<>(); + collectScans(root, scans); + return new GroupStructInfo(true, "", root, Collections.unmodifiableList(scans)); + } + + private static GroupStructInfo invalidStructInfo(String skipReason) { + return skipReason.isEmpty() + ? INVALID + : new GroupStructInfo(false, skipReason, null, INVALID_SCANS); + } + + private static String protectedHeaders() { + StringBuilder letters = new StringBuilder(); + for (Kind kind : Kind.values()) { + if (kind.literalsAreIdentity()) { + letters.append(kind.headerLetters()); + } + } + return letters.toString(); + } + + /** 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; + } + + /** The node this description is made of; null when the description is invalid. */ + private Node root() { + return root; + } + + /** + * 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). A sub + * tree which the canonical string writes twice - a child which is referenced more than once - keeps + * its token in the text twice while its tables are read once, so its descriptors appear here once; + * a caller which compares this list with the tokens of a text has to expect that difference (the + * learned freshness comparison folds the tokens of a text per table for the same reason). Empty when + * the struct info is invalid. + */ + public List<HboScanDescriptor> getScans() { + return scans; + } + + /** The canonical string of one literal mode, with the data state of every scan. */ + public String getCanonical(LiteralMode mode) { + boolean noLiteral = mode == LiteralMode.NO_LITERAL; + String materialized = noLiteral ? canonicalNoLiteral : canonicalWithLiteral; + if (materialized != null) { + return materialized; + } + materialized = root == null ? "" : render(root, mode, false); + if (noLiteral) { + canonicalNoLiteral = materialized; + } else { + canonicalWithLiteral = materialized; + } + return materialized; + } + + /** + * The fingerprint input of one literal mode: the canonical string with the data state of every scan + * removed ({@code S{db.t}}). It is rendered from the same tree as the canonical string, with the scan + * tokens reduced to their identity, and never parsed out of the text, so no text inside a table name + * or a literal can be mistaken for a scan token. + */ + public String getShape(LiteralMode mode) { + boolean noLiteral = mode == LiteralMode.NO_LITERAL; + String materialized = noLiteral ? shapeNoLiteral : shapeWithLiteral; + if (materialized != null) { + return materialized; + } + materialized = root == null ? "" : render(root, mode, true); + if (noLiteral) { + shapeNoLiteral = materialized; + } else { + shapeWithLiteral = materialized; + } + return materialized; + } + + /** + * The sha256 of {@link #shapeKey(LiteralMode)}: the key an hbo entry is registered under. + * + * <p>The preimage is the structural form of this description, not the canonical string: every field + * is written with its length, so a table name, an alias or a literal which contains a character of + * the printed grammar cannot change which field it belongs to. The canonical string is the readable + * projection of the same tree ({@link #getCanonical}), and a pasted struct info is never hashed. + */ + public String getFingerprint(LiteralMode mode) { + boolean noLiteral = mode == LiteralMode.NO_LITERAL; + String materialized = noLiteral ? fingerprintNoLiteral : fingerprintWithLiteral; + if (materialized != null) { + return materialized; + } + materialized = sha256(shapeKey(mode)); + if (noLiteral) { + fingerprintNoLiteral = materialized; + } else { + fingerprintWithLiteral = materialized; + } + return materialized; + } + + private static String sha256(byte[] value) { + return Hashing.sha256().hashBytes(value).toString(); + } + + /** The structural identity of this description in one literal mode (see {@link #getFingerprint}). */ + private byte[] shapeKey(LiteralMode mode) { + ShapeKeyWriter out = new ShapeKeyWriter(); + if (root == null) { + // an invalid description has no tree: its key is a marker of its own + out.tag(-1); + return out.toByteArray(); + } + out.tag(0); + appendShapeKey(out, root, mode); + return out.toByteArray(); + } + + /** + * Write one node of the structural identity: the kind, the fields the canonical string prints for it + * (each as one length prefixed text) and the children. A field is never split by the text of another + * one, which the printed grammar cannot promise for a value which carries its characters. + */ + private static void appendShapeKey(ShapeKeyWriter out, Node node, LiteralMode mode) { + if (node.kind == Kind.PASSTHROUGH && node.children.size() == 1) { + // a node which only composes its child renders no text of its own, so a plan which keeps such + // a node and the same plan with it inlined are one description: the identity must not name + // the node, otherwise the two forms of one plan would be keyed apart + appendShapeKey(out, node.children.get(0).node, mode); + return; + } + out.tag(node.kind.ordinal()); + switch (node.kind) { + case SCAN: + case FILE_SCAN: + appendScanKey(out, node.scan); + return; + case STORAGE_AGGREGATE: + out.text(String.valueOf(node.storageAggregate.getAggOp())); + out.flag(node.scan != null); + if (node.scan != null) { + appendScanKey(out, node.scan); + } + return; + case FILTER: + out.text(sortedConditions(node.conditions, mode)); + break; + case AGGREGATE: + out.text(sortedIdentityConditions(node.conditions)); + break; + case JOIN: + out.text(node.flattenable ? "inner" : String.valueOf(node.joinType)); + out.flag(node.markJoin); + out.text(distinctIdentityConditions(node.conditions)); + break; + case TOPN: { + StringBuilder header = new StringBuilder(); + appendTopNHeader(header, node, mode); + out.text(header.toString()); + break; + } + case LIMIT: + out.text(bound("L", node.limit, node.offset)); + break; + case WINDOW: { + StringBuilder header = new StringBuilder(); + appendWindowHeader(header, node); + out.text(header.toString()); + break; + } + case GENERATE: { + StringBuilder header = new StringBuilder(); + appendGenerateHeader(header, node); + out.text(header.toString()); + break; + } + case REPEAT: { + StringBuilder header = new StringBuilder(); + appendRepeatHeader(header, node, mode); + out.text(header.toString()); + break; + } + case SET_OPERATION: + appendSetOperationKey(out, node); + break; + case PASSTHROUGH: + break; + default: + throw new IllegalStateException("unknown node kind " + node.kind); + } + out.count(node.children.size()); + for (Child child : node.children) { + // a child which is referenced more than once renders the text of the first reference, so the + // identity names the sub tree only, not how often the plan references it + out.count(child.mapping.size()); + for (Expression mapped : child.mapping) { + out.text(renderExpression(mapped, mode)); + } + appendShapeKey(out, child.node, mode); + } + } + + /** + * The identity of a scan, one field at a time: a table name which carries the characters of the + * printed token ({@code #ti:}, a comma) cannot stand for another field because each field has its own + * length here. + */ + private static void appendScanKey(ShapeKeyWriter out, HboScanDescriptor descriptor) { + out.text(descriptor.getTable()); + out.number(descriptor.getTableId()); + out.text(descriptor.getRelationIdentity()); + out.text(descriptor.getPartitionIdentity()); + out.text(descriptor.getTabletIdentity()); + out.text(descriptor.getIndexIdentity()); + } + + /** The header fields of a set operation, which decide which rows its branches combine. */ + private static void appendSetOperationKey(ShapeKeyWriter out, Node node) { + SetOperation setOperation = node.setOperation; + out.text(node.setOperationPlan instanceof Union ? "U" + : node.setOperationPlan instanceof Intersect ? "I" : "E"); + out.text(String.valueOf(setOperation.getQualifier()).toLowerCase(Locale.ROOT)); + out.count(setOperation.getOutputs().size()); + for (NamedExpression output : setOperation.getOutputs()) { + out.text(identityText(output)); + } + List<List<NamedExpression>> constantExprsList = setOperation instanceof Union + ? ((Union) setOperation).getConstantExprsList() : Collections.emptyList(); + out.count(constantExprsList.size()); + for (List<NamedExpression> branch : constantExprsList) { + out.count(branch.size()); + for (NamedExpression constant : branch) { + out.text(identityText(constant)); + } + } + } + + /** + * 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) { + Group group = planNode.getGroupExpression().map(GroupExpression::getOwnerGroup).orElse(null); + if (hasNoHboLookupPath(planNode, group)) { + return Optional.empty(); + } + return fingerprintOfGroup(group, 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) { + Group group = resolveGroup(planNode, groupsById); + if (hasNoHboLookupPath(planNode, group)) { + return Optional.empty(); + } + return fingerprintOfGroup(group, mode); + } + + private static Optional<String> fingerprintOfGroup(Group group, LiteralMode mode) { + GroupStructInfo structInfo = group == null ? INVALID : group.getOrComputeHboStructInfo(); + return structInfo.isValid() ? Optional.of(structInfo.getFingerprint(mode)) : Optional.empty(); + } + + /** + * Resolve the description 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}). Empty when the group content is unsupported. + */ + public static Optional<GroupStructInfo> structInfoOfPlanNode(AbstractPlan planNode, + Map<Integer, Group> groupsById) { + Group group = resolveGroup(planNode, groupsById); + if (hasNoHboLookupPath(planNode, group)) { + // the annotation asks for the struct info first: without this guard the node would advertise + // a key which no read path ever looks up (see hasNoHboLookupPath) + return Optional.empty(); + } + return structInfoOfGroup(group); + } + + private static Optional<GroupStructInfo> structInfoOfGroup(Group group) { + if (group == null) { + return Optional.empty(); + } + GroupStructInfo structInfo = group.getOrComputeHboStructInfo(); + return structInfo.isValid() ? Optional.of(structInfo) : Optional.empty(); + } + + /** + * The description of a group: computed once per query and memoized on the group (see + * {@link Group#getOrComputeHboStructInfo}), so that a parent composes the description of its + * children instead of describing their sub trees again. + */ + public static GroupStructInfo of(Group group) { + return describeGroup(group); + } + + public static GroupStructInfo describeGroup(Group group) { + return describeGroup(group, new Ctx()); + } + + private static GroupStructInfo describeGroup(Group group, Ctx ctx) { + if (!ctx.visited.add(group)) { + // the same group is reached twice (a CTE which is consumed twice, a scan which feeds two + // consumers, a shared sub tree): a tree cannot express the reuse, so the sub tree is + // described again - the description is the one of a plan which writes that sub tree twice + // (a self join for instance), and the read side of a query builds the same description, so + // an entry recorded from one form applies to the other. The sub tree was read once, so the + // repeated reference composes its canonical form but not its scans again (the drift guard + // compares the scan states of the tables a query actually reads). + GroupStructInfo memoized = group.getHboStructInfoOrNull(); + if (memoized != null) { + return memoized; + } + invalid(ctx, "sharing=group " + group.getGroupId().asInt() + " reached twice"); + return INVALID; + } + return describeVisited(group, ctx); + } + + /** + * The description of a group which the caller already marked visited (the join chain marks the + * groups it descends through itself, so that a group is never added twice within one build). + */ + private static GroupStructInfo describeVisited(Group group, Ctx ctx) { + GroupStructInfo memoized = group.getHboStructInfoOrNull(); + if (memoized != null) { + return memoized; + } + GroupStructInfo computed = build(group, ctx); + group.setHboStructInfo(computed); + return computed; + } + + /** + * The first logical expression of a group, or null when it holds none: {@link + * Group#getFirstLogicalExpression()} asserts that there is one, so a group which only holds + * physical expressions (an enforcer group, or a group a rule filled with physical plans) cannot be + * asked that way. + */ + private static GroupExpression firstLogicalExpression(Group group) { + List<GroupExpression> logical = group.getLogicalExpressions(); + return logical.isEmpty() ? null : logical.get(0); + } + + /** Why a group has nothing to describe: its chosen id and how many expressions it holds. */ + private static String describeGroupState(Group group) { + return group.getGroupId().asInt() + + " (logical=" + group.getLogicalExpressions().size() + + ", physical=" + group.getPhysicalExpressions().size() + + ", enforcers=" + group.getEnforcers().size() + ")"; + } + + /** + * The expression which is described for a group. The first logical expression is used, and the logical + * form is what the key of a group has to be made of: the read side looks an entry up while the plan is + * still being optimized, i.e. before any expression is chosen, and a physical plan may hold nodes the + * logical plan abstracts away (the two staged aggregation of a {@code count(*)}, for instance), which + * would move the key of every such group and make a pinned entry miss. The expression the plan chooser + * picked is deliberately not consulted for the same reason - a key must not depend on when it is + * computed. The method is package visible so that a test can pin that the recorded choice of the plan + * chooser is not consulted. + */ + static GroupExpression getRepresentativeExpression(Group group) { + GroupExpression logical = firstLogicalExpression(group); + if (logical != null) { + return logical; + } + // Some groups have no logical plan at all, for instance the group of a + // {@link PhysicalStorageLayerAggregate} (an aggregation answered from the storage layer is a + // catalog relation a rule inserts, there is no logical aggregation left to describe) or a group + // which only holds an enforcer. Such a group is described through the physical expression it + // holds: it is the only form it has, so both sides - the read side and the annotation - see the + // same one, and giving the whole struct info up would hide every node above it. + List<GroupExpression> physical = group.getPhysicalExpressions(); + if (!physical.isEmpty()) { + return physical.get(0); + } + // an enforcer group keeps its expression in the enforcer list, not among the physical ones + return group.getEnforcers().isEmpty() ? null : group.getEnforcers().keySet().iterator().next(); + } + + /** + * Describe one group by composing the descriptions of its children (which are memoized, so each + * group of a plan is described exactly once per query), in both literal modes at once. + */ + private static GroupStructInfo build(Group group, Ctx ctx) { + describedGroupCount.incrementAndGet(); + HboStructSummary summary = group.getOrComputeHboStructSummary(); + int maxScansPerGroup = hboMaxScansPerGroup(); + if (maxScansPerGroup > 0 && summary.getScanCount() > maxScansPerGroup) { + // The canonical string of a group describes its whole sub tree, 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 invalidStructInfo("subtree scans=" + summary.getScanCount() + + " > hbo_max_scans_per_group=" + maxScansPerGroup); + } + GroupExpression logical = firstLogicalExpression(group); + if (logical != null) { + return describeExpression(group, logical, ctx); + } + GroupExpression representative = getRepresentativeExpression(group); + if (representative == null) { + // a group without any expression (no logical one, no physical one, no enforcer): there is + // nothing to describe + return invalidStructInfo("no expression in group " + describeGroupState(group)); + } + return describeExpression(group, representative, ctx); + } + + /** Describe one expression of a group: the body of {@link #build} without the choice of it. */ + private static GroupStructInfo describeExpression(Group group, GroupExpression ge, Ctx ctx) { + Node root; + try { + root = buildNode(ge.getPlan(), ge, ctx); + if (root != null) { + applyColumnDefinitions(ge.getPlan(), root); + } + } catch (RuntimeException e) { + // memo content which the simplified struct info cannot describe (an unexpected plan node, a + // cast of a named expression which is not the alias of a window expression): treat the group as + // invalid and fall back to the legacy behavior instead of failing the optimizer on the hot + // path. Rendering cannot throw for a description which was built: it only concatenates. + LOG.debug("failed to compute hbo struct info for group {}", group.getGroupId(), e); + return invalidStructInfo(ctx.reason); + } + if (!ctx.valid || root == null) { + return invalidStructInfo(ctx.reason); + } + return ofNode(root); + } + + /** + * Bind the columns the children of a node compute to the expression which produces them, and + * substitute a reference to such a column in the expressions of this node. + * + * <p>A project is transparent for structure matching and an aggregation keeps only its grouping + * keys in its header, so a column whose value they compute is <em>named</em> by every node above + * them and <em>defined</em> by none: {@code Aggregate(group by x) -> Project(a % 2 AS x) -> + * Scan(t)} and the same plan with {@code % 3} render one {@code A{gb:col(x)}} although their group + * counts differ, and a pinned row count of one is then applied to the other. The definition is + * therefore carried up: the expressions of this node are rewritten so that a column defined by a + * child renders as the expression which computes it, and the map is handed to the nodes above. + */ + private static void applyColumnDefinitions(Plan plan, Node node) { + Map<ExprId, Expression> childDefinitions = childColumnDefinitions(node); + resolveNodeExpressions(node, childDefinitions); + if (plan instanceof Project) { + node.columnDefinitions = projectColumnDefinitions((Project) plan, childDefinitions); + } else if (plan instanceof Aggregate) { + node.columnDefinitions = aggregateColumnDefinitions((Aggregate<?>) plan, childDefinitions); + } else { + node.columnDefinitions = childDefinitions; + } + } + + /** + * The columns an aggregation computes. Its header keeps the grouping keys - the number of groups, + * which is the row count of the node, depends on them alone - so the values its aggregate + * functions produce are defined nowhere: {@code Filter(s = 1) -> Aggregate(group by k, MIN(v) AS + * s)} and the same tree with {@code MAX(v) AS s} render one filter key although one k group whose + * values are 1 and 2 keeps zero and one rows. A definition is the aggregate function under the + * name of the alias which exposes it, resolved against the columns of the child like the header of + * the aggregation itself. + */ + private static Map<ExprId, Expression> aggregateColumnDefinitions(Aggregate<?> aggregate, + Map<ExprId, Expression> childDefinitions) { + Map<ExprId, Expression> definitions = new LinkedHashMap<>(); + for (NamedExpression output : aggregate.getOutputExpressions()) { + if (output instanceof Alias) { + definitions.put(output.getExprId(), + resolveColumnDefinitions(((Alias) output).child(), childDefinitions)); + } + } + return Collections.unmodifiableMap(definitions); + } + + /** The definitions the children of a node expose: a join forwards the columns of both sides. */ + private static Map<ExprId, Expression> childColumnDefinitions(Node node) { + if (node.children.size() == 1) { + return node.children.get(0).node.columnDefinitions; + } + if (node.children.isEmpty()) { + return Collections.emptyMap(); + } + Map<ExprId, Expression> definitions = new LinkedHashMap<>(); + for (Child child : node.children) { + definitions.putAll(child.node.columnDefinitions); Review Comment: [P2] Preserve null-extension in inherited column definitions Consider `Aggregate(gb=x) -> LEFT_OUTER_JOIN(t.k=u.k) -> [Scan(t), Project(u.k,coalesce(u.v,0) AS x)->Scan(u)]`, versus computing that same x in a project above the join. With t.k={1,2} and u={(k=1,v=0)}, the first has groups {0,NULL} (two rows); the second has {0} (one row). This merge forwards the right-side x definition unchanged across null-extension, so `applyColumnDefinitions` rewrites both grouping keys to the same Coalesce expression. Transparent projects then give both plans identical keys in both literal modes, and identical scan baselines pass freshness, allowing the other plan's pinned/learned group count to apply. Retain the nullable-side evaluation boundary when propagating definitions (or reject unsupported definitions), and add the paired plans. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboPlanInfoProvider.java: ########## @@ -147,35 +315,62 @@ public void handle(Field field, String confVal) throws Exception { } /** - * Reference the above UpdateConfig comments. + * Reference the above UpdateConfig comments. Every per-query cache of this provider is rebuilt + * (they are all sized and expired by the same two configs), and the entries of a running query + * are carried over so that a runtime change cannot drop the plan info of a query which is still + * in flight. */ public static synchronized void updateConfig() { HboPlanStatisticsManager hboManger = Env.getCurrentEnv().getHboPlanStatisticsManager(); if (hboManger == null) { return; } - HboPlanInfoProvider planInfoProvider = hboManger.getHboPlanInfoProvider(); - if (planInfoProvider == null) { + HboPlanInfoProvider provider = hboManger.getHboPlanInfoProvider(); + if (provider == null) { return; } + int cacheNum = Config.hbo_plan_info_cache_num; + int pinnedCacheNum = Config.hbo_pinned_stats_cache_num; + long expireSeconds = Config.expire_hbo_plan_info_cache_in_fe_second; - Cache<String, Map<Integer, PhysicalPlan>> idToPlanCache = buildHboIdToPlanCache( - Config.hbo_plan_info_cache_num, - Config.expire_hbo_plan_info_cache_in_fe_second - ); - Cache<String, Map<PhysicalPlan, Integer>> planToIdCache = buildHboPlanToIdCache( - Config.hbo_plan_info_cache_num, - Config.expire_hbo_plan_info_cache_in_fe_second - ); - Cache<String, Map<RelationId, Set<Expression>>> scanToFilterCache = buildHboScanToFilterCache( - Config.hbo_plan_info_cache_num, - Config.expire_hbo_plan_info_cache_in_fe_second - ); - idToPlanCache.putAll(planInfoProvider.idToPlanCache.asMap()); - planInfoProvider.idToPlanCache = idToPlanCache; - planToIdCache.putAll(planInfoProvider.planToIdCache.asMap()); - planInfoProvider.planToIdCache = planToIdCache; - scanToFilterCache.putAll(planInfoProvider.scanToFilterCache.asMap()); - planInfoProvider.scanToFilterCache = scanToFilterCache; + Cache<String, Map<Integer, PhysicalPlan>> idToPlanCache = + buildHboIdToPlanCache(cacheNum, expireSeconds); + idToPlanCache.putAll(provider.idToPlanCache.asMap()); + provider.idToPlanCache = idToPlanCache; + + Cache<String, Map<PhysicalPlan, Integer>> planToIdCache = + buildHboPlanToIdCache(cacheNum, expireSeconds); + planToIdCache.putAll(provider.planToIdCache.asMap()); + provider.planToIdCache = planToIdCache; + + Cache<String, Map<RelationId, Set<Expression>>> scanToFilterCache = + buildHboScanToFilterCache(cacheNum, expireSeconds); + scanToFilterCache.putAll(provider.scanToFilterCache.asMap()); + provider.scanToFilterCache = scanToFilterCache; Review Comment: [P3] Rebuild the relation-table cache when its configuration changes `relationTableIdCache` is constructed with `hbo_plan_info_cache_num` and `expire_hbo_plan_info_cache_in_fe_second`, but this callback never rebuilds it while updating every sibling cache. For example, starting with both limits disabled and then setting positive bounds through ADMIN leaves this new cache unbounded and non-expiring for subsequent filtered queries until FE restart; reducing a positive size/TTL likewise keeps the original limits. Both settings are mutable and invoke this callback. Rebuild the relation-to-table cache with the updated size/expiry and carry over its entries, as done for the adjacent scan-filter cache. -- 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]
