refactor: derive tableFilterExpr from rules, show db.table names in MatchedTables

- Make tableFilterExpr transient (not persisted), computed from
  tableFilterRules via canonicalize() as single source of truth
- Add computeTableFilterExpr() helper, called in constructor and
  rebuildOnTablesFilter()
- Remove setTableFilterExpr() from Builder
- Change currentTableIds from Set<Long> to Map<Long, String> mapping
  table ID to 'db.table' qualified name
- Update getMatchedTablesString() to show sorted db.table names
  (e.g., 'ods.orders, ods.products') instead of numeric IDs
- Update resolveTableIds() in CacheHotspotManager to return
  Map<Long, String> and update all callers
- Add 3 new tests, update all existing tests for new API

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: bobhan1

fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java

@@ -69,7 +69,6 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -873,9 +872,7 @@ public long createJob(WarmUpClusterCommand stmt) throws AnalysisException {
                         pr.pattern = rule.getRawPattern();
                         persistedRules.add(pr);
                     }
-                    String tableFilterExpr = CloudWarmUpJob.canonicalize(persistedRules);
-                    builder.setTableFilterExpr(tableFilterExpr)
-                            .setTableFilterRules(persistedRules);
+                    builder.setTableFilterRules(persistedRules);
                 }
             } else {
                 builder.setSyncMode(SyncMode.ONCE);
@@ -885,9 +882,9 @@ public long createJob(WarmUpClusterCommand stmt) throws AnalysisException {
             // For event-driven jobs with ON TABLES, rebuild filter and resolve initial table IDs
             if (warmUpJob.hasTableFilter()) {
                 warmUpJob.rebuildOnTablesFilter();
-                Set<Long> initialTableIds = resolveTableIds(warmUpJob.getOnTablesFilter());
-                warmUpJob.setCurrentTableIds(initialTableIds);
-                if (initialTableIds.isEmpty()) {
+                Map<Long, String> initialTableIdNames = resolveTableIds(warmUpJob.getOnTablesFilter());
+                warmUpJob.setCurrentTableIdNames(initialTableIdNames);
+                if (initialTableIdNames.isEmpty()) {
                     LOG.warn("ON TABLES rules do not match any existing table for job {}", jobId);
                 }
             }
@@ -952,8 +949,8 @@ public void replayCloudWarmUpJob(CloudWarmUpJob cloudWarmUpJob) throws Exception
         // Rebuild transient ON TABLES filter from persisted rules
         if (cloudWarmUpJob.hasTableFilter()) {
             cloudWarmUpJob.rebuildOnTablesFilter();
-            Set<Long> tableIds = resolveTableIds(cloudWarmUpJob.getOnTablesFilter());
-            cloudWarmUpJob.setCurrentTableIds(tableIds);
+            Map<Long, String> tableIdNames = resolveTableIds(cloudWarmUpJob.getOnTablesFilter());
+            cloudWarmUpJob.setCurrentTableIdNames(tableIdNames);
         }
 
         if (cloudWarmUpJob.isDone()) {
@@ -974,11 +971,11 @@ public void replayCloudWarmUpJob(CloudWarmUpJob cloudWarmUpJob) throws Exception
     }
 
     /**
-     * Resolve glob-based ON TABLES filter to a set of matching table IDs
+     * Resolve glob-based ON TABLES filter to a map of matching table ID → "db.table" name
      * by iterating all databases and tables in the internal catalog.
      */
-    public Set<Long> resolveTableIds(OnTablesFilter filter) {
-        Set<Long> result = new HashSet<>();
+    public Map<Long, String> resolveTableIds(OnTablesFilter filter) {
+        Map<Long, String> result = new HashMap<>();
         if (filter == null) {
             return result;
         }
@@ -992,7 +989,7 @@ public Set<Long> resolveTableIds(OnTablesFilter filter) {
             }
             for (TableIf tableIf : dbIf.getTables()) {
                 if (filter.shouldWarmUp(dbName, tableIf.getName())) {
-                    result.add(tableIf.getId());
+                    result.put(tableIf.getId(), dbName + "." + tableIf.getName());
                 }
             }
         }
@@ -1009,14 +1006,14 @@ public void refreshAllTableFilters() {
                 continue;
             }
             try {
-                Set<Long> newTableIds = resolveTableIds(job.getOnTablesFilter());
+                Map<Long, String> newTableIdNames = resolveTableIds(job.getOnTablesFilter());
                 Set<Long> oldTableIds = job.getCurrentTableIds();
-                if (!newTableIds.equals(oldTableIds)) {
-                    job.setCurrentTableIds(newTableIds);
+                if (!newTableIdNames.keySet().equals(oldTableIds)) {
+                    job.setCurrentTableIdNames(newTableIdNames);
                     LOG.info("refreshed table filter for job {}: {} -> {} tables",
                             job.getJobId(),
                             oldTableIds == null ? 0 : oldTableIds.size(),
-                            newTableIds.size());
+                            newTableIdNames.size());
                 }
             } catch (Exception e) {
                 LOG.warn("failed to refresh table filter for job {}", job.getJobId(), e);

fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java

@@ -58,7 +58,6 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -143,14 +142,14 @@ public enum SyncEvent {
     @SerializedName(value = "syncEvent")
     protected SyncEvent syncEvent;
 
-    @SerializedName(value = "tableFilterExpr")
-    protected String tableFilterExpr = "";
-
     @SerializedName(value = "tableFilterRules")
     protected List<PersistedTableFilterRule> tableFilterRules = new ArrayList<>();
 
+    // Computed from tableFilterRules via canonicalize(); not persisted.
+    private transient String tableFilterExpr = "";
     private transient OnTablesFilter onTablesFilter;
-    private transient Set<Long> currentTableIds = new HashSet<>();
+    // Maps table ID → "db.table" qualified name for matched tables.
+    private transient Map<Long, String> currentTableIdNames = new HashMap<>();
 
     /**
      * Serializable rule for GSON persistence.
@@ -182,7 +181,6 @@ public static class Builder {
         private SyncMode syncMode = SyncMode.ONCE;
         private SyncEvent syncEvent;
         private long syncInterval;
-        private String tableFilterExpr = "";
         private List<PersistedTableFilterRule> tableFilterRules = new ArrayList<>();
 
         public Builder() {}
@@ -222,11 +220,6 @@ public Builder setSyncInterval(long syncInterval) {
             return this;
         }
 
-        public Builder setTableFilterExpr(String tableFilterExpr) {
-            this.tableFilterExpr = tableFilterExpr;
-            return this;
-        }
-
         public Builder setTableFilterRules(List<PersistedTableFilterRule> tableFilterRules) {
             this.tableFilterRules = tableFilterRules;
             return this;
@@ -249,8 +242,8 @@ private CloudWarmUpJob(Builder builder) {
         this.syncMode = builder.syncMode;
         this.syncEvent = builder.syncEvent;
         this.syncInterval = builder.syncInterval;
-        this.tableFilterExpr = builder.tableFilterExpr;
         this.tableFilterRules = builder.tableFilterRules;
+        this.tableFilterExpr = computeTableFilterExpr();
         this.createTimeMs = System.currentTimeMillis();
     }
 
@@ -443,11 +436,10 @@ public List<String> getJobInfo() {
     }
 
     private String getMatchedTablesString() {
-        if (currentTableIds == null || currentTableIds.isEmpty()) {
+        if (currentTableIdNames == null || currentTableIdNames.isEmpty()) {
             return "";
         }
-        return currentTableIds.stream()
-                .map(String::valueOf)
+        return currentTableIdNames.values().stream()
                 .sorted()
                 .collect(Collectors.joining(", "));
     }
@@ -526,12 +518,29 @@ public OnTablesFilter getOnTablesFilter() {
         return onTablesFilter;
     }
 
+    /**
+     * Returns the set of currently matched table IDs.
+     */
     public Set<Long> getCurrentTableIds() {
-        return currentTableIds;
+        return currentTableIdNames.keySet();
     }
 
-    public void setCurrentTableIds(Set<Long> tableIds) {
-        this.currentTableIds = tableIds;
+    /**
+     * Sets the current matched table ID-to-name mapping.
+     */
+    public void setCurrentTableIdNames(Map<Long, String> idNames) {
+        this.currentTableIdNames = idNames;
+    }
+
+    /**
+     * Compute the canonical table filter expression from persisted rules.
+     * Returns empty string when no table filter rules exist.
+     */
+    private String computeTableFilterExpr() {
+        if (tableFilterRules == null || tableFilterRules.isEmpty()) {
+            return "";
+        }
+        return canonicalize(tableFilterRules);
     }
 
     /**
@@ -565,13 +574,14 @@ public static String canonicalize(List<PersistedTableFilterRule> rules) {
     }
 
     /**
-     * Rebuild the transient OnTablesFilter from persisted tableFilterRules.
+     * Rebuild the transient OnTablesFilter and tableFilterExpr from persisted tableFilterRules.
      * Called after deserialization (EditLog replay, FE restart).
      */
     public void rebuildOnTablesFilter() {
         if (tableFilterRules == null || tableFilterRules.isEmpty()) {
             return;
         }
+        this.tableFilterExpr = computeTableFilterExpr();
         List<OnTablesFilter.TableFilterRule> rules = tableFilterRules.stream()
                 .map(r -> new OnTablesFilter.TableFilterRule(
                         "INCLUDE".equals(r.ruleType)
@@ -821,13 +831,13 @@ private void runEventDrivenJob() throws Exception {
                 }
                 request.setEvent(event);
                 if (hasTableFilter()) {
-                    request.setTableIds(currentTableIds != null
-                            ? new ArrayList<>(currentTableIds) : new ArrayList<>());
+                    request.setTableIds(currentTableIdNames != null
+                            ? new ArrayList<>(currentTableIdNames.keySet()) : new ArrayList<>());
                 }
                 LOG.debug("send warm up request to BE {} ({}). job_id={}, event={}, "
                                 + "request_type=SET_JOB(EVENT), table_ids_count={}",
                         entry.getKey(), getBackendEndpoint(entry.getKey()), jobId, syncEvent,
-                        hasTableFilter() ? currentTableIds.size() : "all");
+                        hasTableFilter() ? currentTableIdNames.size() : "all");
                 TWarmUpTabletsResponse response = entry.getValue().warmUpTablets(request);
                 if (response.getStatus().getStatusCode() != TStatusCode.OK) {
                     if (!response.getStatus().getErrorMsgs().isEmpty()) {