[fix](fe) refuse broadcast join when build-side row count was clamped from unknown

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

@@ -527,6 +527,11 @@ private Optional<String> checkNdvValidation(OlapScan olapScan, double rowCount)
     public Statistics computeOlapScan(OlapScan olapScan) {
         OlapTable olapTable = olapScan.getTable();
         double tableRowCount = getOlapTableRowCount(olapScan);
+        // Track whether the row count was UNKNOWN_ROW_COUNT (-1) before the
+        // clamp below. Downstream gates (e.g. checkBroadcastJoinStats) treat a
+        // clamped row count as untrustworthy so they cannot pick a 1-row build
+        // side based on a value that was actually "we don't know".
+        boolean rowCountWasUnknown = tableRowCount < 0;
         tableRowCount = Math.max(1, tableRowCount);
 
         OlapTableStatistics olapTableStats = Env.getCurrentEnv().getStatisticsCache().getOlapTableStats(olapScan);
@@ -563,6 +568,7 @@ public Statistics computeOlapScan(OlapScan olapScan) {
         }
 
         StatisticsBuilder builder = new StatisticsBuilder();
+        builder.setRowCountWasUnknown(rowCountWasUnknown);
 
         // query for system table or FeUt or analysis, use ColumnStatistic.UNKNOWN
 
@@ -1300,6 +1306,12 @@ private ColumnStatistic getColumnStatistic(
     public Statistics computeCatalogRelation(CatalogRelation catalogRelation) {
         StatisticsBuilder builder = new StatisticsBuilder();
         double tableRowCount = catalogRelation.getTable().getRowCount();
+        // Track whether the row count was UNKNOWN_ROW_COUNT (-1) before any
+        // clamp below. Downstream gates (e.g. checkBroadcastJoinStats) use this
+        // flag to refuse acting on a clamped value, otherwise an external table
+        // with no stats would be wrongly accepted as a broadcast build side.
+        boolean rowCountWasUnknown = tableRowCount < 0;
+        builder.setRowCountWasUnknown(rowCountWasUnknown);
         // for FeUt, use ColumnStatistic.UNKNOWN
         if (!FeConstants.enableInternalSchemaDb
                 || ConnectContext.get() == null
@@ -1323,10 +1335,25 @@ public Statistics computeCatalogRelation(CatalogRelation catalogRelation) {
         if (tableRowCount <= 0) {
             tableRowCount = 1;
             // try to get row count from col stats
+            boolean recoveredFromColStats = false;
             for (SlotReference slot : slotSet) {
                 ColumnStatistic cache = getColumnStatsFromTableCache(catalogRelation, slot);
+                if (cache.count > 0) {
+                    recoveredFromColStats = true;
+                }
                 tableRowCount = Math.max(cache.count, tableRowCount);
             }
+            // Refresh the unknown flag with both conditions ANDed:
+            //  - the upstream value really was UNKNOWN_ROW_COUNT (-1), i.e.
+            //    rowCountWasUnknown captured at the top of this method, AND
+            //  - per-column stats did not recover a positive row count.
+            // ANDing prevents two failure modes:
+            //  1. Original tableRowCount == 0 (a legitimate empty-table signal,
+            //     not unknown) reaching this branch would otherwise be wrongly
+            //     marked unknown.
+            //  2. Future refactors that broaden the outer `if` predicate cannot
+            //     silently clear a flag that should still be true.
+            builder.setRowCountWasUnknown(rowCountWasUnknown && !recoveredFromColStats);
         }
         for (SlotReference slot : slotSet) {
             ColumnStatistic cache;

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

@@ -89,9 +89,25 @@ public static boolean checkBroadcastJoinStats(PhysicalHashJoin<? extends Plan, ?
         double memLimit = sessionVariable.getMaxExecMemByte();
         double rowsLimit = sessionVariable.getBroadcastRowCountLimit();
         double brMemlimit = sessionVariable.getBroadcastHashtableMemLimitPercentage();
-        double datasize = join.getGroupExpression().get().child(1)
-                .getStatistics().computeSize(join.right().getOutput());
-        double rowCount = join.getGroupExpression().get().child(1).getStatistics().getRowCount();
+        // Refuse broadcast when the right child's row count is unknown (UNKNOWN_ROW_COUNT).
+        // Otherwise, an unknown-but-likely-huge table could be picked as build side
+        // because UNKNOWN_ROW_COUNT (-1) gets clamped to 1 row downstream.
+        // Note: We check row count directly instead of isStatsReliable() because estimation
+        // values (even if not perfectly accurate) are considered reliable for decisions.
+        // Additionally, if the row count was clamped from UNKNOWN at scan derivation
+        // (rowCountWasUnknown == true) we must also refuse broadcast: a clamped value
+        // looks like "1 row" but actually means "we don't know", and trusting it would
+        // wrongly broadcast an un-ANALYZE'd large table.
+        if (!join.getGroupExpression().isPresent()) {
+            return false;
+        }
+        org.apache.doris.statistics.Statistics rightStats =
+                join.getGroupExpression().get().child(1).getStatistics();
+        double rowCount = rightStats.getRowCount();
+        if (rowCount < 0 || rightStats.isRowCountWasUnknown()) {
+            return false;
+        }
+        double datasize = rightStats.computeSize(join.right().getOutput());
         return rowCount <= rowsLimit && datasize <= memLimit * brMemlimit;
     }
 

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

@@ -49,6 +49,15 @@ public class Statistics {
 
     private long actualRowCount = -1L;
     private boolean isFromHbo = false;
+    // True iff the row count was clamped from UNKNOWN_ROW_COUNT (-1) to a sentinel
+    // (e.g. 1) during stats derivation. Downstream sizing decisions that take a
+    // hard yes/no on the build side (broadcast eligibility) MUST refuse to act on a
+    // clamped row count, otherwise an un-ANALYZE'd large table can be wrongly chosen
+    // as broadcast build side and OOM at runtime. Cost-model arithmetic still uses
+    // the clamped value because it cannot tolerate negatives. The flag is propagated
+    // through Statistics(Statistics) and StatisticsBuilder so derived stats
+    // (filter / project / single-table-sub-plan) remain marked.
+    private boolean rowCountWasUnknown = false;
 
     public Statistics(Statistics another) {
         this.rowCount = another.rowCount;
@@ -57,6 +66,7 @@ public Statistics(Statistics another) {
         this.tupleSize = another.tupleSize;
         this.deltaRowCount = another.getDeltaRowCount();
         this.isFromHbo = another.isFromHbo;
+        this.rowCountWasUnknown = another.rowCountWasUnknown;
     }
 
     public Statistics(double rowCount, int widthInJoinCluster,
@@ -68,6 +78,13 @@ public Statistics(double rowCount, int widthInJoinCluster,
         this.isFromHbo = isFromHbo;
     }
 
+    public Statistics(double rowCount, int widthInJoinCluster,
+            Map<Expression, ColumnStatistic> expressionToColumnStats, double deltaRowCount, boolean isFromHbo,
+            boolean rowCountWasUnknown) {
+        this(rowCount, widthInJoinCluster, expressionToColumnStats, deltaRowCount, isFromHbo);
+        this.rowCountWasUnknown = rowCountWasUnknown;
+    }
+
     public Statistics(double rowCount, Map<Expression, ColumnStatistic> expressionToColumnStats) {
         this(rowCount, 1, expressionToColumnStats, 0, false);
     }
@@ -89,21 +106,33 @@ public double getRowCount() {
         return rowCount;
     }
 
+    /**
+     * @return true iff this Statistics' row count was clamped from
+     *     UNKNOWN_ROW_COUNT (-1) to a non-negative sentinel during derivation.
+     *     Callers that take a hard yes/no decision on this row count
+     *     (e.g. {@code JoinUtils#checkBroadcastJoinStats}) should treat the
+     *     value as untrustworthy and bail out.
+     */
+    public boolean isRowCountWasUnknown() {
+        return rowCountWasUnknown;
+    }
+
     public Statistics withRowCount(double rowCount) {
         return new Statistics(rowCount, widthInJoinCluster, new HashMap<>(expressionToColumnStats),
-                0, isFromHbo);
+                0, isFromHbo, rowCountWasUnknown);
     }
 
     public Statistics withExpressionToColumnStats(Map<Expression, ColumnStatistic> expressionToColumnStats) {
-        return new Statistics(rowCount, widthInJoinCluster, expressionToColumnStats, 0, isFromHbo);
+        return new Statistics(rowCount, widthInJoinCluster, expressionToColumnStats, 0, isFromHbo,
+                rowCountWasUnknown);
     }
 
     /**
      * Update by count.
      */
     public Statistics withRowCountAndEnforceValid(double rowCount) {
         Statistics statistics = new Statistics(rowCount, widthInJoinCluster,
-                expressionToColumnStats, 0, isFromHbo);
+                expressionToColumnStats, 0, isFromHbo, rowCountWasUnknown);
         statistics.normalizeColumnStatistics(this.rowCount, false);
         return statistics;
     }

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

@@ -32,6 +32,8 @@ public class StatisticsBuilder {
     private double deltaRowCount = 0.0;
 
     private boolean isFromHbo = false;
+    // See Statistics#rowCountWasUnknown for semantics.
+    private boolean rowCountWasUnknown = false;
 
     public StatisticsBuilder() {
         this.expressionToColumnStats = new HashMap<>();
@@ -44,13 +46,25 @@ public StatisticsBuilder(Statistics statistics) {
         this.expressionToColumnStats = new HashMap<>();
         this.expressionToColumnStats.putAll(statistics.columnStatistics());
         this.isFromHbo = statistics.isFromHbo();
+        this.rowCountWasUnknown = statistics.isRowCountWasUnknown();
     }
 
     public StatisticsBuilder setRowCount(double rowCount) {
         this.rowCount = rowCount;
         return this;
     }
 
+    /**
+     * Mark that the row count value being set was clamped from
+     * UNKNOWN_ROW_COUNT during derivation. Use together with {@link #setRowCount}.
+     * The flag is honored by {@link org.apache.doris.nereids.util.JoinUtils#checkBroadcastJoinStats}
+     * which refuses to broadcast a build side whose row count is not trustworthy.
+     */
+    public StatisticsBuilder setRowCountWasUnknown(boolean rowCountWasUnknown) {
+        this.rowCountWasUnknown = rowCountWasUnknown;
+        return this;
+    }
+
     public StatisticsBuilder setWidthInJoinCluster(int widthInJoinCluster) {
         this.widthInJoinCluster = widthInJoinCluster;
         return this;
@@ -77,6 +91,7 @@ public Set<Map.Entry<Expression, ColumnStatistic>> getExpressionColumnStatsEntri
     }
 
     public Statistics build() {
-        return new Statistics(rowCount, widthInJoinCluster, expressionToColumnStats, deltaRowCount, isFromHbo);
+        return new Statistics(rowCount, widthInJoinCluster, expressionToColumnStats,
+                deltaRowCount, isFromHbo, rowCountWasUnknown);
     }
 }

fe/fe-core/src/test/java/org/apache/doris/nereids/util/JoinUtilsTest.java

@@ -20,6 +20,8 @@
 import org.apache.doris.catalog.ColocateTableIndex;
 import org.apache.doris.catalog.ColocateTableIndex.GroupId;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.nereids.memo.Group;
+import org.apache.doris.nereids.memo.GroupExpression;
 import org.apache.doris.nereids.properties.DistributionSpecHash;
 import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType;
 import org.apache.doris.nereids.trees.expressions.Add;
@@ -28,8 +30,12 @@
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.SlotReference;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.Random;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
 import org.apache.doris.nereids.types.TinyIntType;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.statistics.Statistics;
+import org.apache.doris.statistics.StatisticsBuilder;
 
 import com.google.common.collect.Lists;
 import org.junit.jupiter.api.Assertions;
@@ -39,6 +45,7 @@
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 
 public class JoinUtilsTest {
 
@@ -288,4 +295,81 @@ public void testCouldColocateJoinForDiffTableNotInSameGroup() {
             Assertions.assertFalse(JoinUtils.couldColocateJoin(left, right, conjuncts));
         }
     }
+
+    @Test
+    public void testCheckBroadcastJoinStatsRowCountBoundary() {
+        ConnectContext ctx = new ConnectContext();
+        ctx.setThreadLocalInfo();
+        double broadcastRowLimit = ctx.getSessionVariable().getBroadcastRowCountLimit();
+
+        // rowCount strictly above broadcast_row_count_limit: reject.
+        PhysicalHashJoin<?, ?> joinTooMany = Mockito.mock(PhysicalHashJoin.class);
+        GroupExpression geTooMany = Mockito.mock(GroupExpression.class);
+        Statistics statsTooMany = new StatisticsBuilder().setRowCount(broadcastRowLimit + 1).build();
+        Mockito.when(joinTooMany.getGroupExpression()).thenReturn(Optional.of(geTooMany));
+        Mockito.when(geTooMany.child(1)).thenReturn(Mockito.mock(Group.class));
+        Mockito.when(geTooMany.child(1).getStatistics()).thenReturn(statsTooMany);
+        Mockito.when(joinTooMany.right()).thenReturn(Mockito.mock(Plan.class));
+        Mockito.when(joinTooMany.right().getOutput()).thenReturn(Lists.newArrayList());
+
+        Assertions.assertFalse(JoinUtils.checkBroadcastJoinStats(joinTooMany),
+                "Should reject broadcast when rowCount exceeds broadcast_row_count_limit");
+
+        // rowCount exactly at limit: accept (the check is rowCount <= limit).
+        PhysicalHashJoin<?, ?> joinAtLimit = Mockito.mock(PhysicalHashJoin.class);
+        GroupExpression geAtLimit = Mockito.mock(GroupExpression.class);
+        Statistics statsAtLimit = new StatisticsBuilder().setRowCount(broadcastRowLimit).build();
+        Mockito.when(joinAtLimit.getGroupExpression()).thenReturn(Optional.of(geAtLimit));
+        Mockito.when(geAtLimit.child(1)).thenReturn(Mockito.mock(Group.class));
+        Mockito.when(geAtLimit.child(1).getStatistics()).thenReturn(statsAtLimit);
+        Mockito.when(joinAtLimit.right()).thenReturn(Mockito.mock(Plan.class));
+        Mockito.when(joinAtLimit.right().getOutput()).thenReturn(Lists.newArrayList());
+
+        Assertions.assertTrue(JoinUtils.checkBroadcastJoinStats(joinAtLimit),
+                "Should accept broadcast when rowCount equals broadcast_row_count_limit "
+                        + "(check is inclusive, datasize is 0 with no output cols)");
+    }
+
+    @Test
+    public void testCheckBroadcastJoinStatsHappyPath() {
+        ConnectContext ctx = new ConnectContext();
+        ctx.setThreadLocalInfo();
+
+        PhysicalHashJoin<?, ?> join = Mockito.mock(PhysicalHashJoin.class);
+        GroupExpression ge = Mockito.mock(GroupExpression.class);
+        Statistics stats = new StatisticsBuilder().setRowCount(100).build();
+        Mockito.when(join.getGroupExpression()).thenReturn(Optional.of(ge));
+        Mockito.when(ge.child(1)).thenReturn(Mockito.mock(Group.class));
+        Mockito.when(ge.child(1).getStatistics()).thenReturn(stats);
+        Mockito.when(join.right()).thenReturn(Mockito.mock(Plan.class));
+        Mockito.when(join.right().getOutput()).thenReturn(Lists.newArrayList());
+
+        Assertions.assertTrue(JoinUtils.checkBroadcastJoinStats(join),
+                "Small build side under both gates should be accepted as broadcast candidate");
+    }
+
+    @Test
+    public void testCheckBroadcastJoinStatsRejectsClampedRowCount() {
+        ConnectContext ctx = new ConnectContext();
+        ctx.setThreadLocalInfo();
+
+        PhysicalHashJoin<?, ?> join = Mockito.mock(PhysicalHashJoin.class);
+        GroupExpression ge = Mockito.mock(GroupExpression.class);
+        // Mirror what StatsCalculator.computeOlapScan does for an un-ANALYZE'd
+        // Olap table: setRowCountWasUnknown(true) + setRowCount(1).
+        Statistics stats = new StatisticsBuilder()
+                .setRowCount(1)
+                .setRowCountWasUnknown(true)
+                .build();
+        Mockito.when(join.getGroupExpression()).thenReturn(Optional.of(ge));
+        Mockito.when(ge.child(1)).thenReturn(Mockito.mock(Group.class));
+        Mockito.when(ge.child(1).getStatistics()).thenReturn(stats);
+        Mockito.when(join.right()).thenReturn(Mockito.mock(Plan.class));
+        Mockito.when(join.right().getOutput()).thenReturn(Lists.newArrayList());
+
+        Assertions.assertFalse(JoinUtils.checkBroadcastJoinStats(join),
+                "Clamped rowCount = 1 (rowCountWasUnknown=true) must NOT be accepted "
+                        + "as a broadcast candidate; otherwise un-ANALYZE'd large tables "
+                        + "would be wrongly broadcast.");
+    }
 }