englefly commented on code in PR #67549: URL: https://github.com/apache/doris/pull/67549#discussion_r4238199712
########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboStructFreshness.java: ########## @@ -0,0 +1,579 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.stats; + +import org.apache.doris.catalog.DatabaseIf; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.Config; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan; +import org.apache.doris.rpc.RpcException; +import org.apache.doris.statistics.hbo.PlanStatistics; +import org.apache.doris.statistics.hbo.ScanPlanStatistics; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +/** + * Whether a hbo entry may still be applied: the recorded data state of an entry (the baseline of + * every scan of its struct info, see {@link HboScanDescriptor}) compared with the data state of the + * query which wants to use it. + * + * <p>A hbo fingerprint does not contain any data state any more, so an entry keeps matching while a + * table grows. The read side therefore has to decide whether the injected row count still describes + * the data it was measured on: + * <ul> + * <li>{@link #STATE_LIVE}: the data the entry was measured on is still the current one, the entry + * is applied;</li> + * <li>{@link #STATE_DRIFTED}: the data moved, but by at most + * {@code Config.hbo_row_count_change_ratio} of the recorded row count (10% by default): the + * entry is applied, which is the point of the tolerance - a table which merely keeps growing + * (e.g. a few new rows in a partition) does not invalidate a measured row count;</li> + * <li>{@link #STATE_STALE}: the data moved further than that: the entry is <b>not</b> applied, and + * the query falls back to the optimizer estimation;</li> + * <li>{@link #STATE_UNKNOWN}: nothing can be compared (the entry records no row count, or the + * current one cannot be read / was obtained a different way), so there is nothing to judge: the + * entry is applied as it always was.</li> + * </ul> + * The recorded row count is never scaled: an entry either applies with the value the user measured + * or does not apply at all ("soft strategy" - a user who needs the exact recorded state in any case + * should not inject hbo statistics for that query). Setting + * {@code hbo_row_count_change_ratio} to 0 or less switches the verdict back to the strict semantics + * hbo had before the baseline existed: the recorded visible version has to still be the current one. + * + * <p>The rows are compared <b>before</b> the version, and only against a number which was obtained + * the same way ({@link HboScanDescriptor.RowsKind}): the visible version is a property of the table, + * so two different partition selections of one table share it while reading completely different + * amounts of data, and a measured row count must not be compared with one that was derived from the + * table average. Whenever nothing can be compared - a hand written struct literal, an entry whose + * rows were never measured, a current row count which cannot be read, or two numbers of different + * kinds - the state is {@link #STATE_UNKNOWN} and the entry is applied: a transient failure to read + * the catalog must not drop an entry the user injected. The version is only decisive in strict mode + * ({@code hbo_row_count_change_ratio <= 0}). + * + * <p>A scan is judged by the rows of its <b>selected</b> partitions, so pruning decides which data + * the entry depends on: an unrelated partition which grows does not invalidate a pruned entry. + * {@code HBO SHOW STATISTICS} and {@code HBO DELETE STALE STATISTICS} cannot run the query of an + * entry, so there the recorded state is compared with the table as a whole; an entry whose struct + * prunes partitions can therefore only be judged by rows when it was recorded as a whole table scan + * (its state is reported as unknown otherwise, and such an entry is kept unless the user asks for + * {@code OLDER_THAN}). + * + * <p>Learned (profile collected) entries carry the input table statistics of the run they were + * measured in, so they are judged by the same tolerance over those recorded rows; an entry injected + * by {@code HBO SET LEARNED STATISTICS} carries no such statistics and is applied as it always was. + */ +public class HboStructFreshness { + /** The recorded data state is still the current one. */ + public static final String STATE_LIVE = "live"; + /** The data moved, but within the tolerance of {@code Config.hbo_row_count_change_ratio}. */ + public static final String STATE_DRIFTED = "drifted"; + /** The data moved further than the tolerance: the entry must not be applied. */ + public static final String STATE_STALE = "stale"; + /** The entry records no data state (or it cannot be read any more): nothing to judge. */ + public static final String STATE_UNKNOWN = "unknown"; + + private static final Logger LOG = LogManager.getLogger(HboStructFreshness.class); + + private final String state; + // the recorded baseline and the current state, both rendered for display + private final String recorded; + private final String live; + // the row counts of the scan which decided the state, {@link HboScanDescriptor#UNKNOWN} when the + // state was decided by something else (e.g. a version which is not readable) + private final long recordedRows; + private final long liveRows; + + private HboStructFreshness(String state, String recorded, String live, long recordedRows, long liveRows) { + this.state = state; + this.recorded = recorded; + this.live = live; + this.recordedRows = recordedRows; + this.liveRows = liveRows; + } + + public String getState() { + return state; + } + + public boolean isStale() { + return STATE_STALE.equals(state); + } + + public boolean isUnknown() { + return STATE_UNKNOWN.equals(state); + } + + public boolean isDrifted() { + return STATE_DRIFTED.equals(state); + } + + /** The data state the entry was recorded in, e.g. {@code hbo_test.t1:v3,r1000,p1/5}. */ + public String getRecorded() { + return recorded; + } + + /** The data state the entry is compared with (empty when it cannot be read). */ + public String getLive() { + return live; + } + + /** + * The verdict as one token, printed after {@code used=} / {@code skipped=} in the explain + * annotation, e.g. {@code drifted(rows=1030,rec=1000,+3.0%)}. + */ + public String getSummary() { + if (!STATE_DRIFTED.equals(state) && !STATE_STALE.equals(state)) { + return state; + } + if (recordedRows == HboScanDescriptor.UNKNOWN || liveRows == HboScanDescriptor.UNKNOWN) { + return state; + } + return state + "(rows=" + liveRows + ",rec=" + recordedRows + "," + + changeText(recordedRows, liveRows) + ")"; + } + + /** The {@code now} part of {@code HBO SHOW STATISTICS}: the state read from the catalog. */ + public String getLiveDetail() { + if (live.isEmpty()) { + return "-"; + } + if (recordedRows == HboScanDescriptor.UNKNOWN || liveRows == HboScanDescriptor.UNKNOWN) { + return live; + } + return live + "," + changeText(recordedRows, liveRows); + } + + /** + * Whether the entry records any data state at all. An entry which does not (a struct literal + * typed by hand, or one copied from a plan whose tables were never reported) is always applied, + * so a caller does not have to compute the state of the current query to find that out. + */ + public static boolean hasRecordedDataState(String recordedCanonical) { + for (HboScanDescriptor scan : HboScanDescriptor.parseAll(recordedCanonical)) { + if (scan.hasVisibleVersion() || scan.hasScanRows()) { + return true; + } + } + return false; + } + + /** + * Judge an entry against the current query: the data state recorded in {@code recordedCanonical} + * (the struct info of the entry) against the state of {@code liveCanonical} (the struct info the + * read side just computed for the group). This is the decision of the read side. + */ + public static HboStructFreshness between(String recordedCanonical, GroupStructInfo liveStructInfo) { + return between(recordedCanonical, liveStructInfo.getScans()); + } + + /** The same comparison for a caller which has the scans of the current query only. */ + static HboStructFreshness between(String recordedCanonical, List<HboScanDescriptor> liveScans) { + List<HboScanDescriptor> recordedScans = HboScanDescriptor.parseAll(recordedCanonical); + String liveText = renderScans(liveScans); + if (recordedScans.isEmpty()) { + // an entry without a baseline (a hand written struct literal): nothing to judge + return new HboStructFreshness(STATE_UNKNOWN, "", liveText, + HboScanDescriptor.UNKNOWN, HboScanDescriptor.UNKNOWN); + } + if (recordedScans.size() != liveScans.size()) { + // the entry describes a different number of scans than the group: cannot be compared + return new HboStructFreshness(STATE_UNKNOWN, renderScans(recordedScans), liveText, + HboScanDescriptor.UNKNOWN, HboScanDescriptor.UNKNOWN); + } + String state = STATE_LIVE; + String recordedText = renderScans(recordedScans); + + long recordedRows = HboScanDescriptor.UNKNOWN; + long liveRows = HboScanDescriptor.UNKNOWN; + for (int i = 0; i < recordedScans.size(); i++) { + HboScanDescriptor recorded = recordedScans.get(i); + HboScanDescriptor live = liveScans.get(i); + String scanState; + if (!recorded.getTable().equals(live.getTable())) { + // the same fingerprint describes the same scans in the same order, so this cannot + // happen for an entry of this group: refuse to judge rather than guess + scanState = STATE_UNKNOWN; + } else { + scanState = stateOf(recorded, live); + } + if (worse(scanState, state).equals(scanState)) { + state = scanState; + recordedRows = recorded.getScanRows(); + liveRows = live.getScanRows(); + } + } + return new HboStructFreshness(state, recordedText, liveText, recordedRows, liveRows); + } + + /** + * Judge an entry against the catalog as a whole (used by {@code HBO SHOW STATISTICS} and + * {@code HBO DELETE STALE STATISTICS}, which cannot run the query of an entry): the recorded + * visible version and row count of every table against the current ones. + * + * <p>Row counts are only comparable when the scan read the whole table (no partition was pruned + * away), because only then the recorded rows are the rows of the table. + */ + public static HboStructFreshness of(String canonicalStructInfo) { + List<HboScanDescriptor> recordedScans = HboScanDescriptor.parseAll(canonicalStructInfo); + if (recordedScans.isEmpty()) { + return new HboStructFreshness(STATE_UNKNOWN, "", "", HboScanDescriptor.UNKNOWN, + HboScanDescriptor.UNKNOWN); + } + String state = STATE_LIVE; + // the current state of every table, de duplicated like the recorded side (a self join reads + // the same table twice, which would only repeat the same numbers) + Set<String> liveTexts = new LinkedHashSet<>(); + long recordedRows = HboScanDescriptor.UNKNOWN; + long liveRows = HboScanDescriptor.UNKNOWN; + for (HboScanDescriptor recorded : recordedScans) { + OlapTable table = resolveTable(recorded.getTable()); + if (table == null) { + if (worse(STATE_UNKNOWN, state).equals(STATE_UNKNOWN)) { + state = STATE_UNKNOWN; + } + liveTexts.add(recorded.getTable() + ":-"); + continue; + } + if (table.getIndexIdList().size() > 1 && !recorded.isPartitionSelectionComplete()) { + // the table has more than one index (a rollup) and the entry records only a partition + // count, so this statement cannot tell which index its row count came from: judging it + // against the base index could be wrong in both directions, so it is left to a query + liveTexts.add(recorded.getTable() + ":-"); + if (worse(STATE_UNKNOWN, state).equals(STATE_UNKNOWN)) { + state = STATE_UNKNOWN; + } + continue; + } + long currentVersion = visibleVersionOf(table); + // the rows of the whole table, computed the same way the recorded rows of a whole table + // scan were computed, so the two numbers describe the same thing + HboScanDescriptor.Rows liveRowsOfTable = HboScanDescriptor.rowsOf(table, + new ArrayList<>(table.getPartitionIds()), table.getBaseIndexId()); Review Comment: 认可。catalog 作用域在“整表扫描 + 多 index”时绕过上面的 index 守卫，直接用 base index 行数比，于是 rollup 的 baseline 会被判 STALE 并被 DELETE STALE 删掉。修法：整表扫描且表有多 index 时也返回 UNKNOWN（读侧仍可用），我补提交。 ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboStructFreshness.java: ########## @@ -0,0 +1,579 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.stats; + +import org.apache.doris.catalog.DatabaseIf; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.Config; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan; +import org.apache.doris.rpc.RpcException; +import org.apache.doris.statistics.hbo.PlanStatistics; +import org.apache.doris.statistics.hbo.ScanPlanStatistics; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +/** + * Whether a hbo entry may still be applied: the recorded data state of an entry (the baseline of + * every scan of its struct info, see {@link HboScanDescriptor}) compared with the data state of the + * query which wants to use it. + * + * <p>A hbo fingerprint does not contain any data state any more, so an entry keeps matching while a + * table grows. The read side therefore has to decide whether the injected row count still describes + * the data it was measured on: + * <ul> + * <li>{@link #STATE_LIVE}: the data the entry was measured on is still the current one, the entry + * is applied;</li> + * <li>{@link #STATE_DRIFTED}: the data moved, but by at most + * {@code Config.hbo_row_count_change_ratio} of the recorded row count (10% by default): the + * entry is applied, which is the point of the tolerance - a table which merely keeps growing + * (e.g. a few new rows in a partition) does not invalidate a measured row count;</li> + * <li>{@link #STATE_STALE}: the data moved further than that: the entry is <b>not</b> applied, and + * the query falls back to the optimizer estimation;</li> + * <li>{@link #STATE_UNKNOWN}: nothing can be compared (the entry records no row count, or the + * current one cannot be read / was obtained a different way), so there is nothing to judge: the + * entry is applied as it always was.</li> + * </ul> + * The recorded row count is never scaled: an entry either applies with the value the user measured + * or does not apply at all ("soft strategy" - a user who needs the exact recorded state in any case + * should not inject hbo statistics for that query). Setting + * {@code hbo_row_count_change_ratio} to 0 or less switches the verdict back to the strict semantics + * hbo had before the baseline existed: the recorded visible version has to still be the current one. + * + * <p>The rows are compared <b>before</b> the version, and only against a number which was obtained + * the same way ({@link HboScanDescriptor.RowsKind}): the visible version is a property of the table, + * so two different partition selections of one table share it while reading completely different + * amounts of data, and a measured row count must not be compared with one that was derived from the + * table average. Whenever nothing can be compared - a hand written struct literal, an entry whose + * rows were never measured, a current row count which cannot be read, or two numbers of different + * kinds - the state is {@link #STATE_UNKNOWN} and the entry is applied: a transient failure to read + * the catalog must not drop an entry the user injected. The version is only decisive in strict mode + * ({@code hbo_row_count_change_ratio <= 0}). + * + * <p>A scan is judged by the rows of its <b>selected</b> partitions, so pruning decides which data + * the entry depends on: an unrelated partition which grows does not invalidate a pruned entry. + * {@code HBO SHOW STATISTICS} and {@code HBO DELETE STALE STATISTICS} cannot run the query of an + * entry, so there the recorded state is compared with the table as a whole; an entry whose struct + * prunes partitions can therefore only be judged by rows when it was recorded as a whole table scan + * (its state is reported as unknown otherwise, and such an entry is kept unless the user asks for + * {@code OLDER_THAN}). + * + * <p>Learned (profile collected) entries carry the input table statistics of the run they were + * measured in, so they are judged by the same tolerance over those recorded rows; an entry injected + * by {@code HBO SET LEARNED STATISTICS} carries no such statistics and is applied as it always was. + */ +public class HboStructFreshness { + /** The recorded data state is still the current one. */ + public static final String STATE_LIVE = "live"; + /** The data moved, but within the tolerance of {@code Config.hbo_row_count_change_ratio}. */ + public static final String STATE_DRIFTED = "drifted"; + /** The data moved further than the tolerance: the entry must not be applied. */ + public static final String STATE_STALE = "stale"; + /** The entry records no data state (or it cannot be read any more): nothing to judge. */ + public static final String STATE_UNKNOWN = "unknown"; + + private static final Logger LOG = LogManager.getLogger(HboStructFreshness.class); + + private final String state; + // the recorded baseline and the current state, both rendered for display + private final String recorded; + private final String live; + // the row counts of the scan which decided the state, {@link HboScanDescriptor#UNKNOWN} when the + // state was decided by something else (e.g. a version which is not readable) + private final long recordedRows; + private final long liveRows; + + private HboStructFreshness(String state, String recorded, String live, long recordedRows, long liveRows) { + this.state = state; + this.recorded = recorded; + this.live = live; + this.recordedRows = recordedRows; + this.liveRows = liveRows; + } + + public String getState() { + return state; + } + + public boolean isStale() { + return STATE_STALE.equals(state); + } + + public boolean isUnknown() { + return STATE_UNKNOWN.equals(state); + } + + public boolean isDrifted() { + return STATE_DRIFTED.equals(state); + } + + /** The data state the entry was recorded in, e.g. {@code hbo_test.t1:v3,r1000,p1/5}. */ + public String getRecorded() { + return recorded; + } + + /** The data state the entry is compared with (empty when it cannot be read). */ + public String getLive() { + return live; + } + + /** + * The verdict as one token, printed after {@code used=} / {@code skipped=} in the explain + * annotation, e.g. {@code drifted(rows=1030,rec=1000,+3.0%)}. + */ + public String getSummary() { + if (!STATE_DRIFTED.equals(state) && !STATE_STALE.equals(state)) { + return state; + } + if (recordedRows == HboScanDescriptor.UNKNOWN || liveRows == HboScanDescriptor.UNKNOWN) { + return state; + } + return state + "(rows=" + liveRows + ",rec=" + recordedRows + "," + + changeText(recordedRows, liveRows) + ")"; + } + + /** The {@code now} part of {@code HBO SHOW STATISTICS}: the state read from the catalog. */ + public String getLiveDetail() { + if (live.isEmpty()) { + return "-"; + } + if (recordedRows == HboScanDescriptor.UNKNOWN || liveRows == HboScanDescriptor.UNKNOWN) { + return live; + } + return live + "," + changeText(recordedRows, liveRows); + } + + /** + * Whether the entry records any data state at all. An entry which does not (a struct literal + * typed by hand, or one copied from a plan whose tables were never reported) is always applied, + * so a caller does not have to compute the state of the current query to find that out. + */ + public static boolean hasRecordedDataState(String recordedCanonical) { + for (HboScanDescriptor scan : HboScanDescriptor.parseAll(recordedCanonical)) { + if (scan.hasVisibleVersion() || scan.hasScanRows()) { + return true; + } + } + return false; + } + + /** + * Judge an entry against the current query: the data state recorded in {@code recordedCanonical} + * (the struct info of the entry) against the state of {@code liveCanonical} (the struct info the + * read side just computed for the group). This is the decision of the read side. + */ + public static HboStructFreshness between(String recordedCanonical, GroupStructInfo liveStructInfo) { + return between(recordedCanonical, liveStructInfo.getScans()); + } + + /** The same comparison for a caller which has the scans of the current query only. */ + static HboStructFreshness between(String recordedCanonical, List<HboScanDescriptor> liveScans) { + List<HboScanDescriptor> recordedScans = HboScanDescriptor.parseAll(recordedCanonical); + String liveText = renderScans(liveScans); + if (recordedScans.isEmpty()) { + // an entry without a baseline (a hand written struct literal): nothing to judge + return new HboStructFreshness(STATE_UNKNOWN, "", liveText, + HboScanDescriptor.UNKNOWN, HboScanDescriptor.UNKNOWN); + } + if (recordedScans.size() != liveScans.size()) { + // the entry describes a different number of scans than the group: cannot be compared + return new HboStructFreshness(STATE_UNKNOWN, renderScans(recordedScans), liveText, + HboScanDescriptor.UNKNOWN, HboScanDescriptor.UNKNOWN); + } + String state = STATE_LIVE; + String recordedText = renderScans(recordedScans); + + long recordedRows = HboScanDescriptor.UNKNOWN; + long liveRows = HboScanDescriptor.UNKNOWN; + for (int i = 0; i < recordedScans.size(); i++) { + HboScanDescriptor recorded = recordedScans.get(i); + HboScanDescriptor live = liveScans.get(i); + String scanState; + if (!recorded.getTable().equals(live.getTable())) { + // the same fingerprint describes the same scans in the same order, so this cannot + // happen for an entry of this group: refuse to judge rather than guess + scanState = STATE_UNKNOWN; + } else { + scanState = stateOf(recorded, live); + } + if (worse(scanState, state).equals(scanState)) { + state = scanState; + recordedRows = recorded.getScanRows(); + liveRows = live.getScanRows(); + } + } + return new HboStructFreshness(state, recordedText, liveText, recordedRows, liveRows); + } + + /** + * Judge an entry against the catalog as a whole (used by {@code HBO SHOW STATISTICS} and + * {@code HBO DELETE STALE STATISTICS}, which cannot run the query of an entry): the recorded + * visible version and row count of every table against the current ones. + * + * <p>Row counts are only comparable when the scan read the whole table (no partition was pruned + * away), because only then the recorded rows are the rows of the table. + */ + public static HboStructFreshness of(String canonicalStructInfo) { + List<HboScanDescriptor> recordedScans = HboScanDescriptor.parseAll(canonicalStructInfo); + if (recordedScans.isEmpty()) { + return new HboStructFreshness(STATE_UNKNOWN, "", "", HboScanDescriptor.UNKNOWN, + HboScanDescriptor.UNKNOWN); + } + String state = STATE_LIVE; + // the current state of every table, de duplicated like the recorded side (a self join reads + // the same table twice, which would only repeat the same numbers) + Set<String> liveTexts = new LinkedHashSet<>(); + long recordedRows = HboScanDescriptor.UNKNOWN; + long liveRows = HboScanDescriptor.UNKNOWN; + for (HboScanDescriptor recorded : recordedScans) { + OlapTable table = resolveTable(recorded.getTable()); + if (table == null) { + if (worse(STATE_UNKNOWN, state).equals(STATE_UNKNOWN)) { + state = STATE_UNKNOWN; + } + liveTexts.add(recorded.getTable() + ":-"); + continue; + } + if (table.getIndexIdList().size() > 1 && !recorded.isPartitionSelectionComplete()) { + // the table has more than one index (a rollup) and the entry records only a partition + // count, so this statement cannot tell which index its row count came from: judging it + // against the base index could be wrong in both directions, so it is left to a query + liveTexts.add(recorded.getTable() + ":-"); + if (worse(STATE_UNKNOWN, state).equals(STATE_UNKNOWN)) { + state = STATE_UNKNOWN; + } + continue; + } + long currentVersion = visibleVersionOf(table); + // the rows of the whole table, computed the same way the recorded rows of a whole table + // scan were computed, so the two numbers describe the same thing + HboScanDescriptor.Rows liveRowsOfTable = HboScanDescriptor.rowsOf(table, + new ArrayList<>(table.getPartitionIds()), table.getBaseIndexId()); + long currentRows = liveRowsOfTable.getRows(); + HboScanDescriptor.RowsKind currentRowsKind = liveRowsOfTable.getKind(); + liveTexts.add(recorded.getTable() + ":v" + currentVersion + + (currentRows == HboScanDescriptor.UNKNOWN ? "" + : (currentRowsKind == HboScanDescriptor.RowsKind.MEASURED ? ",r" : ",e") + + currentRows)); + String scanState; + if (!recorded.hasVisibleVersion() && !recorded.hasScanRows()) { + // nothing was recorded for this scan: it applies as it always was, nothing to judge + scanState = STATE_UNKNOWN; + currentRows = HboScanDescriptor.UNKNOWN; + } else if (threshold() <= 0) { + scanState = recorded.hasVisibleVersion() && currentVersion == recorded.getVisibleVersion() + ? STATE_LIVE : STATE_STALE; + currentRows = HboScanDescriptor.UNKNOWN; + } else if (recorded.hasVisibleVersion() && currentVersion == recorded.getVisibleVersion()) { + // the visible version is a property of the whole table, so an unchanged version means + // that nothing was loaded into any of its partitions: the recorded state is still the + // current one, whatever the entry was measured on. (The read side can not use this + // shortcut: it knows which partitions the entry was measured on and which ones the + // query reads, and those can differ under one unchanged version.) + scanState = STATE_LIVE; + } else if (!recorded.hasScanRows() || !recorded.isPartitionSelectionComplete()) { + // the entry records no row count at all (a struct literal of an older format, for + // example), or it records the rows of pruned partitions only, which are not the rows + // of the table: the state can only be judged by a query of that entry + scanState = STATE_UNKNOWN; + currentRows = HboScanDescriptor.UNKNOWN; + } else if (currentRows == HboScanDescriptor.UNKNOWN + || currentRowsKind != recorded.getRowsKind()) { + // the current row count cannot be read or was obtained differently than the recorded + // one: the version is the only comparable signal + scanState = recorded.hasVisibleVersion() && currentVersion == recorded.getVisibleVersion() Review Comment: 已在本轮修复：catalog 作用域遇到行数口径不同/读不到时不再退回 version 判定，改为 unknown，与读侧（unknown → 照旧应用）一致，避免 DELETE STALE 删掉查询仍在用的条目。 ########## 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(), + rows.getRows(), rows.getKind(), selectedPartitionIds.size(), + table.getPartitionNames().size()); + } + + /** + * The rows the given partitions of a table hold, computed the way the optimizer computes the + * input size of a scan (see {@code StatsCalculator.getOlapTableRowCount} / + * {@code getSelectedPartitionRowCount}): the row count of a partition as the table statistics + * report it, the rows loaded since the last analysis as the fallback for the whole table, and - + * for a partition whose own row count is not available (e.g. right after a load, before the + * backend reported it again) - the average of the table. + * + * <p>An entry records this number and the read side computes it again, so both sides describe + * the same quantity (the size of the input the row count was estimated on) instead of a + * physical row count which would be read from a different place on each side. + * {@link #UNKNOWN} is returned only when the table has no row count at all (never analyzed and + * never reported by a backend). + */ + public static Rows rowsOf(OlapTable table, List<Long> partitionIds, long indexId) { + long rows = 0; + long unknownPartitions = 0; + for (long partitionId : partitionIds) { + long partitionRows = reportedRowsOf(table, partitionId, indexId); + if (partitionRows == UNKNOWN) { + unknownPartitions++; + } else { + rows += partitionRows; + } + } + if (unknownPartitions == 0) { + return new Rows(rows, RowsKind.MEASURED); + } + long tableRows = tableRowsOf(table, indexId); + if (tableRows == UNKNOWN) { + return Rows.UNKNOWN_ROWS; + } + // a partition whose row count is unknown holds at least one row, the rest is the average of + // the table (the same rule the optimizer uses for the same situation). The result is an + // estimate and is marked as one, so that the drift verdict never compares it with a measured + // row count. + return new Rows(rows + Math.max(unknownPartitions, + tableRows * unknownPartitions / table.getPartitionNum()), RowsKind.ESTIMATED); + } + + /** + * The row count a backend reported for one partition, or {@link #UNKNOWN} when no backend + * reported it (or when a newer load invalidated the report): only such a number is a + * measurement of the partition, everything else is derived by {@link #rowsOf}. + */ + private static long reportedRowsOf(OlapTable table, long partitionId, long indexId) { + // the strict accessor is the only one which distinguishes an index no backend reported yet + // (-1) from an empty one (0): the non strict one maps both to 0, which would make a never + // reported partition look like a measurement of zero rows + return table.getRowCountForPartitionIndex(partitionId, indexId, true); + } + + /** The rows of a whole table, with the rows loaded since the last analysis as the fallback. */ + public static long tableRowsOf(OlapTable table, long indexId) { + long rows = table.getRowCountForIndex(indexId, true); + if (rows != UNKNOWN) { + return rows; + } + TableStatsMeta tableMeta = Env.getCurrentEnv().getAnalysisManager() + .findTableStatsStatus(table.getId()); + if (tableMeta == null || tableMeta.userInjected) { + return tableMeta == null ? UNKNOWN : tableMeta.getRowCount(indexId); + } + long analyzedRows = tableMeta.getRowCount(indexId); + return analyzedRows == UNKNOWN ? UNKNOWN + : analyzedRows + tableMeta.getBaseIndexDeltaRowCount(table); + } + + /** + * Parse the content of a scan token, as printed by {@link #render()}. Missing parts (an unknown + * number, or a selection which covers every partition) are {@link #UNKNOWN} / all partitions. + */ + public static HboScanDescriptor parse(String header) { + String[] parts = header.split(","); + long version = UNKNOWN; + long rows = UNKNOWN; + RowsKind kind = RowsKind.UNKNOWN; + long selected = UNKNOWN; + long total = UNKNOWN; + for (int i = 1; i < parts.length; i++) { + String part = parts[i]; + if (part.startsWith("v")) { + version = parseLong(part.substring(1)); + } else if (part.startsWith("r")) { + rows = parseLong(part.substring(1)); + kind = RowsKind.MEASURED; + } else if (part.startsWith("e")) { + rows = parseLong(part.substring(1)); + kind = RowsKind.ESTIMATED; + } else if (part.startsWith("p")) { + int slash = part.indexOf('/'); + if (slash > 0) { + selected = parseLong(part.substring(1, slash)); + total = parseLong(part.substring(slash + 1)); + } + } + } + if (selected == UNKNOWN) { + // no partition part: the scan selects every partition of the table + selected = total; + } + return new HboScanDescriptor(parts[0], version, rows, kind, selected, total); + } + + /** The content of the scan token of this scan: {@code db.t,v3,r1000,p1/5}. */ + public String render() { + StringBuilder sb = new StringBuilder(table); + if (visibleVersion != UNKNOWN) { + sb.append(",v").append(visibleVersion); + } + // a measured row count is marked 'r', an estimated one 'e': the verdict only ever compares + // two numbers of the same kind (see HboStructFreshness) + if (rowsKind == RowsKind.MEASURED) { + sb.append(",r").append(scanRows); + } else if (rowsKind == RowsKind.ESTIMATED) { + sb.append(",e").append(scanRows); + } + if (totalPartitions != UNKNOWN && selectedPartitions != totalPartitions) { + sb.append(",p").append(selectedPartitions).append('/').append(totalPartitions); + } + return sb.toString(); + } + + /** Every scan of a canonical struct info, in the order the tokens appear in it. */ + public static List<HboScanDescriptor> parseAll(String canonicalStructInfo) { + List<HboScanDescriptor> scans = new ArrayList<>(); + if (canonicalStructInfo == null || canonicalStructInfo.isEmpty()) { + return scans; + } + int index = 0; + while ((index = nextScanTokenStart(canonicalStructInfo, index)) >= 0) { + int end = canonicalStructInfo.indexOf('}', index); + if (end < 0) { + break; + } + scans.add(parse(canonicalStructInfo.substring(index + 2, end))); + index = end + 1; + } + return scans; + } + + /** + * The index of the next scan token at or after {@code from}, or -1. A scan token is an + * {@code S} group, which always follows the start of the string, an opening parenthesis or a + * separator, so a column whose name contains {@code S} cannot be mistaken for one. + */ + public static int nextScanTokenStart(String canonicalStructInfo, int from) { + int index = from; + while ((index = canonicalStructInfo.indexOf("S{", index)) >= 0) { + if (index == 0 || "();[".indexOf(canonicalStructInfo.charAt(index - 1)) >= 0) { Review Comment: 认可，这是文本式 token 定位的已知弱点：nextScanTokenStart 允许 S{ 出现在 ( 之后，而 lit(S{fake}:STRING) 正好满足。修法：token 扫描跳过 lit(...) 内部（按括号配对），或让解析器直接产出关系键而不复用文本扫描。不影响 key 语义，只影响被接受的输入集合，我补提交。 ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/GroupStructInfo.java: ########## @@ -0,0 +1,902 @@ +// 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) { + return INVALID; Review Comment: 已在本轮修复：GroupStructInfo.of 在 !ctx.valid 时改为 ctx.retryable ? TRANSIENT_FAILURE : INVALID，云上 version RPC 失败不会再被 memo group 缓存成 INVALID 而让整个查询用不上 hbo。 ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboStatisticsStore.java: ########## @@ -0,0 +1,276 @@ +// 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() { + } + + /** 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; + String struct = truncateUtf8(structCanonical == null ? "" : structCanonical, STRUCT_MAX_BYTES); Review Comment: 已在本轮修复：HBO SET 现在拒绝超过 struct_info 列长度（65533 字节）的 STRUCT 并报错，不再截断持久化——截断会保留 fingerprint 却丢掉尾部 scan token，从而同时废掉 freshness 与关系预筛。 ########## fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboStatisticsStore.java: ########## @@ -0,0 +1,276 @@ +// 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() { + } + + /** 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; + 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 a pinned entry. + * + * @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) { + try { + ensureTable(); + String sql = "DELETE FROM " + FULL_QUALIFIED + " WHERE `fingerprint` = '" + + StatisticsUtil.escapeSQL(fingerprint) + "'"; + StatisticsUtil.execUpdate(sql); + return true; + } catch (Exception t) { + LOG.warn("failed to delete hbo pinned statistics for fingerprint {}", fingerprint, t); + return false; + } + } + + /** + * Load all pinned entries from the internal table. + * + * @return the loaded entries, or {@code null} when the load failed (e.g. the internal schema + * is not ready yet); the caller decides when to retry + */ + public static List<HboPlanStatisticsManager.PinnedHboStatistics> loadAll() { + List<HboPlanStatisticsManager.PinnedHboStatistics> result = new ArrayList<>(); + try { + ensureTable(); + List<ResultRow> rows = StatisticsUtil.execStatisticQueryOrThrow( + "SELECT `fingerprint`, `row_count`, `stats_type`, `struct_info`, `" + EXPANSION_COLUMN + + "`, `" + LITERAL_MODE_COLUMN + "`, `" + CREATE_TIME_COLUMN + "` FROM " + + FULL_QUALIFIED); + for (ResultRow row : rows) { + try { + HboPlanStatisticsManager.PinnedType type = + HboPlanStatisticsManager.PinnedType.fromName(row.get(2)); + result.add(new HboPlanStatisticsManager.PinnedHboStatistics( + row.get(0), + Long.parseLong(row.get(1)), + type == null ? HboPlanStatisticsManager.PinnedType.EXACT : type, + row.get(3) == null ? "" : row.get(3), + parseExpansion(row.get(4)), + parseLiteralMode(row.get(5)), + parseCreateTime(row.get(6)))); + } catch (RuntimeException e) { + LOG.warn("skip malformed hbo pinned statistics row {}", row, e); + } + } + return result; + } catch (Exception t) { + LOG.warn("failed to load hbo pinned statistics from internal table", t); + return null; + } + } + + private static String createDdl() { + // follow the internal-table convention (see InternalSchemaInitializer) so that CREATE + // never fails on clusters whose min_replication_num_per_tablet exceeds 1 + int replication = Math.max(1, Config.min_replication_num_per_tablet); + return "CREATE TABLE IF NOT EXISTS `" + InternalCatalog.INTERNAL_CATALOG_NAME + "`.`" + INTERNAL_DB + + "`.`" + TABLE + "` (\n" + + " `fingerprint` varchar(64) NOT NULL COMMENT \"\",\n" + + " `row_count` bigint NOT NULL COMMENT \"\",\n" + + " `stats_type` varchar(32) NOT NULL COMMENT \"\",\n" + + " `literal_mode` varchar(16) NOT NULL COMMENT \"\",\n" + + " `struct_info` varchar(" + STRUCT_MAX_BYTES + ") NULL COMMENT \"\",\n" + + " `expansion` double NULL COMMENT \"\",\n" + + " `create_time` datetime(3) NOT NULL COMMENT \"\"\n" + + ") ENGINE = olap\n" + + "UNIQUE KEY(`fingerprint`)\n" + + "COMMENT \"Doris internal hbo pinned statistics table, DO NOT MODIFY IT\"\n" + + "DISTRIBUTED BY HASH(`fingerprint`)\n" + + "BUCKETS 1\n" + + "PROPERTIES (\"replication_num\" = \"" + replication + "\")"; + } + + /** Render a creation time as the datetime literal the internal table stores. */ + private static String createTimeLiteral(long createTimeMs) { + return DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS").format( Review Comment: 认可。写库用 writer FE 的 ZoneId.systemDefault() 生成 DATETIME，读库按 loading FE 的时区解释，跨时区会偏移，直接影响 OLDER_THAN。修法：持久化 epoch 毫秒（或统一按 UTC 写读并兼容解释历史行）。需要动内部表语义，建议单独提交并补一条改 ZoneId 的用例。 -- 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]
