[feature](HBO) Add history-based optimizer statistics with structural struct-info keys

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2332,6 +2332,66 @@ public class Config extends ConfigBase {
                     + "cache entries.")
     public static int hbo_plan_info_cache_num = 1000;
 
+    /**
+     * Threshold of the "extremely small filter estimate" guard used by pinned entries injected with
+     * {@code TYPE FILTER_SMALL}: such an entry is only applied while the optimizer's own filter
+     * estimate E satisfies {@code E <= 1} or {@code E <= inputRows * hbo_filter_small_ratio}, i.e.
+     * while the pathological regime that motivated the injection still holds. Hot mutable.
+     */
+    @ConfField(mutable = true, description = "The default setting is 0.001. Relative threshold below which "
+            + "the optimizer's own filter estimate counts as 'extremely small' for HBO FILTER_SMALL "
+            + "pinned entries.")
+    public static double hbo_filter_small_ratio = 0.001;
+
+    /**
+     * Tolerance of a hbo entry against data growth. A hbo key no longer contains the visible version
+     * or the row count of its tables, so an entry keeps matching while a table grows; the read side
+     * applies an entry while the row count of the data it was measured on changed by at most this
+     * ratio (|now - recorded| / recorded). Beyond it the entry is not applied and the query falls
+     * back to the optimizer estimation. A non-positive value disables the tolerance, i.e. an entry
+     * is only applied while the recorded data state is exactly the current one. Hot mutable.
+     */
+    @ConfField(mutable = true, description = "The default setting is 0.1. Relative row count change of the data "
+            + "an injected HBO entry was measured on, up to which the entry may still be applied; 0 or "
+            + "less means the recorded data state has to match exactly.")
+    public static double hbo_row_count_change_ratio = 0.1;
+
+    /**
+     * Maximum number of manually injected (pinned) hbo statistics entries kept per FE. When the
+     * limit is exceeded the least recently used pinned entry is evicted (LRU); pinned entries are
+     * otherwise not expired automatically. A non-positive value disables the LRU bound
+     * (unbounded), matching the other hbo caches. Takes effect at FE start (not hot-mutable).
+     */
+    @ConfField(description = "The default setting is 5000. Maximum number of manually injected "
+            + "(pinned) hbo statistics entries kept per FE; exceeding the limit evicts the least "
+            + "recently used entry; a non-positive value disables the bound (unbounded). Takes "
+            + "effect at FE start (not hot-mutable).")
+    public static int hbo_pinned_stats_cache_num = 5000;
+
+    /**
+     * Persist manually injected (pinned) hbo statistics into the internal database table
+     * __internal_schema.hbo_statistics, so that they survive FE restarts. In-memory pinned
+     * statistics stay authoritative; changes are written through synchronously (best effort)
+     * and each FE loads the table in the background at start-up (HboPinnedStatisticsLoader).
+     * Persistence applies to SET/DELETE issued
+     * while this config is true: entries SET while it is false are not persisted, and entries
+     * DELETEd while it is false keep their stored rows. A FE loads the table into memory at most
+     * once per process (in the background at start-up, retried until it succeeds), so rows deleted
+     * while it is false reappear only for a FE whose one-time load is still pending (e.g. right after
+     * FE start). On that load, existing in-memory entries win over the stored
+     * snapshot; deletions issued while this config is true during the pending load are honored —
+     * rows of theirs that the load sees are best-effort removed from the table again, and they
+     * reappear after a FE restart only if that removal also fails. In general, rows whose
+     * best-effort removal from the table fails (during or after the one-time load, e.g. because
+     * the internal table is temporarily unreachable) reappear after a FE restart; re-issue the
+     * DELETE once the table is reachable. A FE never refreshes entries SET by other FEs during
+     * its lifetime.
+     */
+    @ConfField(mutable = true, description = "The default setting is true. When true, HBO SET/DELETE STATISTICS "
+            + "entries are persisted into __internal_schema.hbo_statistics and reloaded after FE "
+            + "restart.")
+    public static boolean hbo_persist_pinned_to_internal_db = true;
+
     /**
      * Maximum number of events to poll in each RPC.
      */

fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java

@@ -2112,6 +2112,10 @@ protected void startNonMasterDaemonThreads() {
         workloadRuntimeStatusMgr.start();
         admissionControl.start();
         splitSourceManager.start();
+
+        // loads the pinned hbo statistics this FE persisted in the internal database (a no-op when
+        // the feature is off); it runs in the background so that no user query pays for the load
+        hboPlanStatisticsManager.startPinnedStatisticsLoader();
     }
 
     private void transferToNonMaster(FrontendNodeType newType) {

fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java

@@ -390,18 +390,19 @@ public void publishHboPlanStatistics(String queryId, List<TPlanNodeRuntimeStatsI
             Map<Integer, PhysicalPlan> idToPlanMap = planInfoProvider.getIdToPlanMap(queryId);
             Map<PhysicalPlan, Integer> planToIdMap = planInfoProvider.getPlanToIdMap(queryId);
             Map<RelationId, Set<Expression>> scanToFilterMap = planInfoProvider.getScanToFilterMap(queryId);
+            Map<Integer, String> nodeFingerprints = planInfoProvider.getNodeIdToFingerprintMap(queryId);
 
             if (!idToPlanMap.isEmpty() && idToPlanMap.size() == planToIdMap.size()) {
                 Map<PlanNodeAndHash, PlanStatisticsWithInputInfo> curPlanStatistics = HboUtils.genPlanStatisticsMap(
-                        idToPlanMap, planToIdMap, scanToFilterMap, curPlanNodeRuntimeStats);
+                        idToPlanMap, planToIdMap, scanToFilterMap, curPlanNodeRuntimeStats, nodeFingerprints);
                 Map<PlanNodeAndHash, RecentRunsPlanStatistics> recentRunsPlanStatisticsMap =
                         hboPlanStatisticsProvider.getHboPlanStats(
                                 curPlanStatistics.keySet().stream().collect(Collectors.toList()));
 
                 // update plan statistics
                 Map<PlanNodeAndHash, RecentRunsPlanStatistics> newPlanStatistics = curPlanStatistics.entrySet().stream()
                         .filter(entry -> entry.getKey().getHash().isPresent()
-                                && entry.getValue().getInputTableInfo().getInputTableStatistics().isPresent())
+                                && entry.getValue().getInputTableInfo().hasInputTables())
                         .collect(Collectors.toMap(
                                 Map.Entry::getKey,
                                 entry -> {

fe/fe-core/src/main/java/org/apache/doris/common/profile/RuntimeProfile.java

@@ -822,6 +822,36 @@ public void write(DataOutput output) throws IOException {
         Text.writeString(output, GsonUtils.GSON.toJson(this));
     }
 
+    /** The rows a scan read, as its nested counter profiles report them (see the caller). */
+    private void fillInputRowsFromChildProfiles(TPlanNodeRuntimeStatsItem item) {
+        if (item.getInputRows() != 0) {
+            return;
+        }
+        childLock.readLock().lock();
+        try {
+            for (Pair<RuntimeProfile, Boolean> pair : childList) {
+                RuntimeProfile child = pair.first;
+                if (!"CustomCounters".equals(child.getName()) && !"Scanner".equals(child.getName())) {
+                    continue;
+                }
+                for (Map.Entry<String, Counter> entry : child.counterMap.entrySet()) {
+                    if (!(entry.getValue() instanceof AggCounter)) {
+                        continue;
+                    }
+                    String counterName = entry.getKey();
+                    if ("RowsRead".equals(counterName) || "ScanRows".equals(counterName)) {
+                        item.setInputRows(((AggCounter) entry.getValue()).sum.getValue());
+                        if (item.getInputRows() != 0) {
+                            return;
+                        }
+                    }
+                }
+            }
+        } finally {
+            childLock.readLock().unlock();
+        }
+    }
+
     private TPlanNodeRuntimeStatsItem toTPlanNodeRuntimeStatsItem() {
         TPlanNodeRuntimeStatsItem item = new TPlanNodeRuntimeStatsItem();
         item.setNodeId(this.nodeId());
@@ -835,6 +865,17 @@ private TPlanNodeRuntimeStatsItem toTPlanNodeRuntimeStatsItem() {
                     case "RowsProduced":
                         item.setOutputRows(value.sum.getValue());
                         break;
+                    case "RowsRead":
+                    case "ScanRows":
+                        // a scan operator reports the rows it read under these names; a learned entry has
+                        // to record the rows a scan read, and "InputRows" is only registered for sink
+                        // operators, so a scan which records zero here would be judged stale on the next
+                        // identical query (the learned comparison reads the recorded scan rows). The
+                        // case of "InputRows" below wins when a node reports both.
+                        if (item.getInputRows() == 0) {
+                            item.setInputRows(value.sum.getValue());
+                        }
+                        break;
                     case "InputRows":
                         item.setInputRows(value.sum.getValue());
                         if (isBuildSinkOperator) {
@@ -864,6 +905,12 @@ private TPlanNodeRuntimeStatsItem toTPlanNodeRuntimeStatsItem() {
             }
         }
 
+        // a scan operator registers the rows it read on its "CustomCounters" child profile
+        // (`ADD_COUNTER(custom_profile(), "ScanRows", ...)`), while the operator node itself reports
+        // none: without them the published learned entry records zero rows for its input table, and the
+        // freshness comparison has nothing to compare
+        fillInputRowsFromChildProfiles(item);
+
         // Handle runtime profile structure
         /*
 - RuntimeFilterInfo: sum 0, avg 0, max 0, min 0