[Fix](Query Stats) Add QueryStatsRecorder for column-level query and filter - Part1

fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java

@@ -85,6 +85,7 @@
 import org.apache.doris.qe.ResultSet;
 import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.qe.VariableMgr;
+import org.apache.doris.statistics.query.QueryStatsRecorder;
 import org.apache.doris.statistics.util.StatisticsUtil;
 import org.apache.doris.thrift.TQueryCacheParam;
 
@@ -613,6 +614,7 @@ protected void splitFragments(PhysicalPlan resultPlan) {
             return;
         }
         PlanFragment root = physicalPlanTranslator.translatePlan(physicalPlan);
+        QueryStatsRecorder.record(physicalPlan, statementContext);
         if (statementContext.getConnectContext().getExecutor() != null) {
             statementContext.getConnectContext().getExecutor().getSummaryProfile()
                     .setNereidsTranslateTime(TimeUtils.getStartTimeMs());

fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java

@@ -319,6 +319,7 @@ public enum TableFrom {
     // When true, data will be collected to a single node to avoid generating too many small files
     private boolean useGatherForIcebergRewrite = false;
     private boolean hasNestedColumns;
+    private boolean queryStatsRecorded = false;
 
     private final Set<CTEId> mustInlineCTE = new HashSet<>();
     private final Set<String> usedAIResourceNames = new LinkedHashSet<>();
@@ -1163,6 +1164,14 @@ public boolean isInsert() {
         return isInsert;
     }
 
+    public boolean isQueryStatsRecorded() {
+        return queryStatsRecorded;
+    }
+
+    public void setQueryStatsRecorded() {
+        queryStatsRecorded = true;
+    }
+
     public Optional<Map<TableIf, Set<Expression>>> getMvRefreshPredicates() {
         return mvRefreshPredicates;
     }

fe/fe-core/src/main/java/org/apache/doris/statistics/query/QueryStatsRecorder.java

@@ -0,0 +1,189 @@
+// 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.statistics.query;
+
+import org.apache.doris.analysis.StatementBase;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.common.Config;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.glue.LogicalPlanAdapter;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ExprId;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.commands.Command;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalDeferMaterializeOlapScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalFilter;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
+import org.apache.doris.qe.ConnectContext;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * Records column-level query and filter hit stats from the Nereids physical plan.
+ */
+public class QueryStatsRecorder {
+    private static final Logger LOG = LogManager.getLogger(QueryStatsRecorder.class);
+
+    private QueryStatsRecorder() {}
+
+    public static void record(PhysicalPlan plan, StatementContext stmtContext) {
+        if (!shouldRecord(stmtContext)) {
+            return;
+        }
+        if (stmtContext.isQueryStatsRecorded()) {
+            return;
+        }
+        stmtContext.setQueryStatsRecorded();
+        try {
+            Map<ExprId, PhysicalOlapScan> exprIdToScan = new HashMap<>();
+            Map<String, StatsDelta> deltas = new LinkedHashMap<>();
+            walkPlan(plan, exprIdToScan, deltas);
+            if (exprIdToScan.isEmpty()) {
+                return;
+            }
+            for (Slot slot : plan.getOutput()) {
+                SlotReference sr = unwrapSlotRef(slot);
+                if (sr == null) {
+                    continue;
+                }
+                PhysicalOlapScan sourceScan = exprIdToScan.get(sr.getExprId());
+                if (sourceScan == null) {
+                    continue;
+                }
+                StatsDelta delta = getOrCreateDelta(deltas, sourceScan);
+                if (delta == null) {
+                    continue;
+                }
+                sr.getOriginalColumn().ifPresent(col -> {
+                    if (!col.getName().isEmpty()) {
+                        delta.addQueryStats(col.getName());
+                    }
+                });
+            }
+            for (StatsDelta delta : deltas.values()) {
+                if (!delta.empty()) {
+                    Env.getCurrentEnv().getQueryStats().addStats(delta);
+                }
+            }
+        } catch (Exception e) {
+            LOG.warn("Failed to record query stats", e);
+        }
+    }
+
+    // Package-private for testing.
+    static boolean shouldRecord(StatementContext ctx) {
+        if (!Config.enable_query_hit_stats) {
+            return false;
+        }
+        // Skip internal queries such as auto-analyze.
+        ConnectContext connectContext = ctx.getConnectContext();
+        if (connectContext != null && connectContext.getState().isInternal()) {
+            return false;
+        }
+        StatementBase stmt = ctx.getParsedStatement();
+        if (stmt == null || stmt.isExplain()) {
+            return false;
+        }
+        if (stmt instanceof LogicalPlanAdapter
+                && ((LogicalPlanAdapter) stmt).getLogicalPlan() instanceof Command) {
+            return false;
+        }
+        return true;
+    }
+
+    private static void walkPlan(Plan plan,
+            Map<ExprId, PhysicalOlapScan> exprIdToScan,
+            Map<String, StatsDelta> deltas) {
+        if (plan instanceof PhysicalOlapScan) {
+            PhysicalOlapScan scan = (PhysicalOlapScan) plan;
+            for (Slot slot : scan.getOutput()) {
+                exprIdToScan.put(slot.getExprId(), scan);
+            }
+            return;
+        }
+        if (plan instanceof PhysicalDeferMaterializeOlapScan) {
+            PhysicalOlapScan inner =
+                    ((PhysicalDeferMaterializeOlapScan) plan).getPhysicalOlapScan();
+            for (Slot slot : plan.getOutput()) {
+                exprIdToScan.put(slot.getExprId(), inner);
+            }
+            return;
+        }
+        for (Plan child : plan.children()) {
+            walkPlan(child, exprIdToScan, deltas);
+        }
+        if (plan instanceof PhysicalFilter) {
+            PhysicalFilter<?> filter = (PhysicalFilter<?>) plan;
+            for (Expression conjunct : filter.getConjuncts()) {
+                conjunct.getInputSlots().forEach(slot -> {
+                    if (!(slot instanceof SlotReference)) {
+                        return;
+                    }
+                    SlotReference sr = (SlotReference) slot;
+                    PhysicalOlapScan sourceScan = exprIdToScan.get(sr.getExprId());
+                    if (sourceScan == null) {
+                        return;
+                    }
+                    StatsDelta delta = getOrCreateDelta(deltas, sourceScan);
+                    if (delta == null) {
+                        return;
+                    }
+                    sr.getOriginalColumn().ifPresent(col -> {
+                        if (!col.getName().isEmpty()) {
+                            delta.addFilterStats(col.getName());
+                        }
+                    });
+                });
+            }
+        }
+    }
+
+    private static SlotReference unwrapSlotRef(Expression expr) {
+        if (expr instanceof SlotReference) {
+            return (SlotReference) expr;
+        }
+        if (expr instanceof Alias) {
+            return unwrapSlotRef(((Alias) expr).child());
+        }
+        return null;
+    }
+
+    private static StatsDelta getOrCreateDelta(Map<String, StatsDelta> deltas,
+            PhysicalOlapScan scan) {
+        OlapTable t = scan.getTable();
+        DatabaseIf<?> db = scan.getDatabase();
+        if (t == null || db == null) {
+            return null;
+        }
+        String key = t.getCatalogId() + "_" + db.getId() + "_" + t.getId()
+                + "_" + scan.getSelectedIndexId();
+        return deltas.computeIfAbsent(key, k ->
+                new StatsDelta(t.getCatalogId(), db.getId(), t.getId(), scan.getSelectedIndexId()));
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/statistics/query/QueryStatsUtil.java

@@ -171,8 +171,9 @@ public static Map<Long, Long> getMergedReplicasStats(List<Long> replicaIds) {
 
     private static List<TQueryStatsResult> getStats(TGetQueryStatsRequest request) {
         List<TQueryStatsResult> results = new ArrayList<>();
+        String selfNodeName = Env.getCurrentEnv().getNodeName();
         for (Frontend fe : Env.getCurrentEnv().getFrontends(null /* all */)) {
-            if (!fe.isAlive() || fe.getHost().equals(Env.getCurrentEnv().getSelfNode().getHost())) {
+            if (!fe.isAlive() || fe.getNodeName().equals(selfNodeName)) {
                 continue;
             }
             FrontendService.Client client = null;

fe/fe-core/src/test/java/org/apache/doris/statistics/query/QueryStatsRecorderTest.java

@@ -0,0 +1,119 @@
+// 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.statistics.query;
+
+import org.apache.doris.analysis.ExplainOptions;
+import org.apache.doris.common.Config;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.glue.LogicalPlanAdapter;
+import org.apache.doris.nereids.trees.plans.commands.Command;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.QueryState;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class QueryStatsRecorderTest {
+
+    private boolean originalConfigValue;
+
+    @BeforeEach
+    public void setUp() {
+        originalConfigValue = Config.enable_query_hit_stats;
+    }
+
+    @AfterEach
+    public void tearDown() {
+        // Restore config to avoid affecting other tests.
+        Config.enable_query_hit_stats = originalConfigValue;
+    }
+
+    // ── shouldRecord guard tests ─────────────────────────────────────────────
+
+    @Test
+    public void testShouldNotRecordWhenConfigOff() {
+        Config.enable_query_hit_stats = false;
+        StatementContext ctx = new StatementContext();
+        LogicalPlanAdapter stmt = new LogicalPlanAdapter(
+                Mockito.mock(org.apache.doris.nereids.trees.plans.logical.LogicalPlan.class), ctx);
+        ctx.setParsedStatement(stmt);
+        Assertions.assertFalse(QueryStatsRecorder.shouldRecord(ctx));
+    }
+
+    @Test
+    public void testShouldNotRecordWhenStatementIsNull() {
+        Config.enable_query_hit_stats = true;
+        StatementContext ctx = new StatementContext();
+        // parsedStatement not set — getParsedStatement() returns null
+        Assertions.assertFalse(QueryStatsRecorder.shouldRecord(ctx));
+    }
+
+    @Test
+    public void testShouldNotRecordExplain() {
+        Config.enable_query_hit_stats = true;
+        StatementContext ctx = new StatementContext();
+        LogicalPlanAdapter stmt = new LogicalPlanAdapter(
+                Mockito.mock(org.apache.doris.nereids.trees.plans.logical.LogicalPlan.class), ctx);
+        // isExplain() returns true when explainOptions is non-null.
+        stmt.setIsExplain(new ExplainOptions(false, false, false));
+        ctx.setParsedStatement(stmt);
+        Assertions.assertFalse(QueryStatsRecorder.shouldRecord(ctx));
+    }
+
+    @Test
+    public void testShouldNotRecordDml() {
+        Config.enable_query_hit_stats = true;
+        StatementContext ctx = new StatementContext();
+        Command dmlCommand = Mockito.mock(Command.class);
+        LogicalPlanAdapter stmt = new LogicalPlanAdapter(dmlCommand, ctx);
+        ctx.setParsedStatement(stmt);
+        Assertions.assertFalse(QueryStatsRecorder.shouldRecord(ctx));
+    }
+
+    @Test
+    public void testShouldNotRecordInternalQuery() {
+        Config.enable_query_hit_stats = true;
+        StatementContext ctx = new StatementContext();
+        org.apache.doris.nereids.trees.plans.logical.LogicalPlan selectPlan =
+                Mockito.mock(org.apache.doris.nereids.trees.plans.logical.LogicalPlan.class);
+        LogicalPlanAdapter stmt = new LogicalPlanAdapter(selectPlan, ctx);
+        ctx.setParsedStatement(stmt);
+        // Simulate an internal query (e.g. auto-analyze)
+        QueryState state = new QueryState();
+        state.setInternal(true);
+        ConnectContext connectContext = Mockito.mock(ConnectContext.class);
+        Mockito.when(connectContext.getState()).thenReturn(state);
+        ctx.setConnectContext(connectContext);
+        Assertions.assertFalse(QueryStatsRecorder.shouldRecord(ctx));
+    }
+
+    @Test
+    public void testShouldRecordNormalSelect() {
+        Config.enable_query_hit_stats = true;
+        StatementContext ctx = new StatementContext();
+        // A LogicalPlanAdapter wrapping a non-Command plan represents a SELECT.
+        org.apache.doris.nereids.trees.plans.logical.LogicalPlan selectPlan =
+                Mockito.mock(org.apache.doris.nereids.trees.plans.logical.LogicalPlan.class);
+        LogicalPlanAdapter stmt = new LogicalPlanAdapter(selectPlan, ctx);
+        ctx.setParsedStatement(stmt);
+        Assertions.assertTrue(QueryStatsRecorder.shouldRecord(ctx));
+    }
+}