apache · morningman · Jul 1, 2026
@@ -199,8 +199,12 @@ Status IcebergTableReader::get_next_block_inner(Block* block, size_t* read_rows,
 }
 
 Status IcebergTableReader::init_row_filters() {
-    // We get the count value by doris's be, so we don't need to read the delete file
-    if (_push_down_agg_type == TPushAggOp::type::COUNT && _table_level_row_count > 0) {
+    // We get the count value by doris's be, so we don't need to read the delete file.
+    // A table-level row count of 0 (e.g. an all-deleted table read with ignore_iceberg_dangling_delete,
+    // where total-records == total-position-deletes) is still a valid pushed-down count, so accept >= 0.
+    // FE sends -1 when there is no table-level count; using > 0 here would drop a genuine 0 into the
+    // delete-applying path below and never produce the intended CountReader(0).
+    if (_push_down_agg_type == TPushAggOp::type::COUNT && _table_level_row_count >= 0) {
         return Status::OK();
     }
 

@@ -1063,6 +1063,34 @@ public static List<Column> parseSchema(Schema schema, boolean enableMappingVarbi
         return resSchema;
     }
 
+    /**
+     * Decide whether a row count can be read from an Iceberg snapshot summary.
+     * Returns {@link TableIf#UNKNOWN_ROW_COUNT} when required counters are absent
+     * or when delete semantics make the summary count unsafe to use.
+     */
+    @VisibleForTesting
+    public static long getCountFromSummary(Map<String, String> summary, boolean ignoreDanglingDelete) {
+        String equalityDeletes = summary.get(TOTAL_EQUALITY_DELETES);
+        String positionDeletes = summary.get(TOTAL_POSITION_DELETES);
+        String totalRecords = summary.get(TOTAL_RECORDS);
+        if (equalityDeletes == null || positionDeletes == null || totalRecords == null) {
+            return TableIf.UNKNOWN_ROW_COUNT;
+        }
+        if (!equalityDeletes.equals("0")) {
+            return TableIf.UNKNOWN_ROW_COUNT;
+        }
+
+        long deleteCount = Long.parseLong(positionDeletes);
+        if (deleteCount == 0) {
+            return Long.parseLong(totalRecords);
+        }
+        if (ignoreDanglingDelete) {
+            return Long.parseLong(totalRecords) - deleteCount;
+        } else {
+            return TableIf.UNKNOWN_ROW_COUNT;
+        }
+    }
+
     /**
      * Estimate iceberg table row count.
      * Get the row count by adding all task file recordCount.
@@ -1082,7 +1110,12 @@ public static long getIcebergRowCount(ExternalTable tbl) {
             return TableIf.UNKNOWN_ROW_COUNT;
         }
         Map<String, String> summary = snapshot.summary();
-        long rows = Long.parseLong(summary.get(TOTAL_RECORDS)) - Long.parseLong(summary.get(TOTAL_POSITION_DELETES));
+        long rows = getCountFromSummary(summary, true);
+        if (rows == TableIf.UNKNOWN_ROW_COUNT) {
+            LOG.info("Iceberg table {}.{}.{} row count in summary is unknown, return -1.",
+                    tbl.getCatalog().getName(), tbl.getDbName(), tbl.getName());
+            return TableIf.UNKNOWN_ROW_COUNT;
+        }
         LOG.info("Iceberg table {}.{}.{} row count in summary is {}",
                 tbl.getCatalog().getName(), tbl.getDbName(), tbl.getName(), rows);
         return rows;

@@ -1067,24 +1067,7 @@ public long getCountFromSnapshot() throws UserException {
             return 0;
         }
 
-        Map<String, String> summary = snapshot.summary();
-        if (!summary.get(IcebergUtils.TOTAL_EQUALITY_DELETES).equals("0")) {
-            // has equality delete files, can not push down count
-            return -1;
-        }
-
-        long deleteCount = Long.parseLong(summary.get(IcebergUtils.TOTAL_POSITION_DELETES));
-        if (deleteCount == 0) {
-            // no delete files, can push down count directly
-            return Long.parseLong(summary.get(IcebergUtils.TOTAL_RECORDS));
-        }
-        if (sessionVariable.ignoreIcebergDanglingDelete) {
-            // has position delete files, if we ignore dangling delete, can push down count
-            return Long.parseLong(summary.get(IcebergUtils.TOTAL_RECORDS)) - deleteCount;
-        } else {
-            // otherwise, can not push down count
-            return -1;
-        }
+        return IcebergUtils.getCountFromSummary(snapshot.summary(), sessionVariable.ignoreIcebergDanglingDelete);
     }
 
     @Override

@@ -0,0 +1,86 @@
+// 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.datasource.iceberg.source;
+
+import org.apache.doris.datasource.iceberg.IcebergUtils;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+public class IcebergCountPushDownTest {
+
+    private static Map<String, String> summary(String equalityDeletes, String positionDeletes, String totalRecords) {
+        ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+        if (equalityDeletes != null) {
+            builder.put(IcebergUtils.TOTAL_EQUALITY_DELETES, equalityDeletes);
+        }
+        if (positionDeletes != null) {
+            builder.put(IcebergUtils.TOTAL_POSITION_DELETES, positionDeletes);
+        }
+        if (totalRecords != null) {
+            builder.put(IcebergUtils.TOTAL_RECORDS, totalRecords);
+        }
+        return builder.build();
+    }
+
+    @Test
+    public void testMissingCounterFallsBackToScan() {
+        // Snapshots written by compaction/replace (and some writers) may omit
+        // total-* counters. The pushdown previously NPE'd on the missing key;
+        // it must now fall back to a normal scan (return -1).
+        Assertions.assertEquals(-1L, IcebergUtils.getCountFromSummary(summary(null, "0", "100"), false));
+        Assertions.assertEquals(-1L, IcebergUtils.getCountFromSummary(summary("0", null, "100"), false));
+        Assertions.assertEquals(-1L, IcebergUtils.getCountFromSummary(summary("0", "0", null), false));
+        Assertions.assertEquals(-1L, IcebergUtils.getCountFromSummary(Collections.emptyMap(), false));
+    }
+
+    @Test
+    public void testUtilityMissingCounterReturnsUnknownCount() {
+        Assertions.assertEquals(-1L, IcebergUtils.getCountFromSummary(summary("0", null, "100"), true));
+    }
+
+    @Test
+    public void testNoDeletesPushesDownTotalRecords() {
+        Assertions.assertEquals(100L, IcebergUtils.getCountFromSummary(summary("0", "0", "100"), false));
+    }
+
+    @Test
+    public void testEqualityDeletesCannotPushDown() {
+        Assertions.assertEquals(-1L, IcebergUtils.getCountFromSummary(summary("3", "0", "100"), false));
+    }
+
+    @Test
+    public void testPositionDeletesRespectIgnoreDangling() {
+        // ignoreDanglingDelete = true -> total-records minus position-deletes
+        Assertions.assertEquals(90L, IcebergUtils.getCountFromSummary(summary("0", "10", "100"), true));
+        // ignoreDanglingDelete = false -> cannot push down (fall back to scan)
+        Assertions.assertEquals(-1L, IcebergUtils.getCountFromSummary(summary("0", "10", "100"), false));
+    }
+
+    @Test
+    public void testZeroCountWithPositionDeletesIsPushedDown() {
+        // total-records == position-deletes -> count is 0. With ignore_iceberg_dangling_delete this
+        // is a valid pushed-down count; FE returns 0 and BE honors it via CountReader(0) (the BE
+        // table-level guard accepts table_level_row_count >= 0). It must NOT fall back to -1.
+        Assertions.assertEquals(0L, IcebergUtils.getCountFromSummary(summary("0", "100", "100"), true));
+    }
+}