[fix](scanner)(nereids) Harden FileScannerV2 and fix external COUNT pushdown semantics

be/src/exec/operator/scan_operator.cpp

@@ -1046,6 +1046,12 @@ TPushAggOp::type ScanLocalState<Derived>::get_push_down_agg_type() {
     return _parent->cast<typename Derived::Parent>()._push_down_agg_type;
 }
 
+template <typename Derived>
+const std::optional<std::vector<int32_t>>& ScanLocalState<Derived>::get_push_down_count_slot_ids()
+        const {
+    return _parent->cast<typename Derived::Parent>()._push_down_count_slot_ids;
+}
+
 template <typename Derived>
 int64_t ScanLocalState<Derived>::limit_per_scanner() {
     return _parent->cast<typename Derived::Parent>()._limit_per_scanner;
@@ -1231,6 +1237,9 @@ Status ScanOperatorX<LocalStateType>::init(const TPlanNode& tnode, RuntimeState*
     } else {
         _push_down_agg_type = TPushAggOp::type::NONE;
     }
+    if (tnode.__isset.push_down_count_slot_ids) {
+        _push_down_count_slot_ids = tnode.push_down_count_slot_ids;
+    }
 
     if (tnode.__isset.topn_filter_source_node_ids) {
         _topn_filter_source_node_ids = tnode.topn_filter_source_node_ids;

be/src/exec/operator/scan_operator.h

@@ -18,6 +18,7 @@
 #pragma once
 
 #include <cstdint>
+#include <optional>
 #include <set>
 #include <string>
 
@@ -74,6 +75,7 @@ class ScanLocalStateBase : public PipelineXLocalState<> {
     virtual void set_scan_ranges(RuntimeState* state,
                                  const std::vector<TScanRangeParams>& scan_ranges) = 0;
     virtual TPushAggOp::type get_push_down_agg_type() = 0;
+    virtual const std::optional<std::vector<int32_t>>& get_push_down_count_slot_ids() const = 0;
 
     // If scan operator is serial operator(like topn), its real parallelism is 1.
     // Otherwise, its real parallelism is query_parallel_instance_num.
@@ -252,6 +254,7 @@ class ScanLocalState : public ScanLocalStateBase {
                          const std::vector<TScanRangeParams>& scan_ranges) override {}
 
     TPushAggOp::type get_push_down_agg_type() override;
+    const std::optional<std::vector<int32_t>>& get_push_down_count_slot_ids() const override;
 
     std::vector<Dependency*> execution_dependencies() override {
         if (_filter_dependencies.empty()) {
@@ -452,6 +455,16 @@ class ScanOperatorX : public OperatorX<LocalStateType> {
 
     TPushAggOp::type _push_down_agg_type;
 
+    // Semantic arguments of a pushed-down COUNT. This is deliberately optional because absence
+    // and an empty list have different meanings during a BE-first rolling upgrade:
+    //
+    //  - nullopt: an old FE did not send the field, so the new BE must use the normal scan;
+    //  - empty: the new FE explicitly planned COUNT(*)/COUNT(1);
+    //  - non-empty: the new FE explicitly planned COUNT(col).
+    //
+    // Treating nullopt as empty would silently reinterpret an old plan as COUNT(*).
+    std::optional<std::vector<int32_t>> _push_down_count_slot_ids;
+
     // Record the value of the aggregate function 'count' from doris's be
     int64_t _push_down_count = -1;
     const int _parallel_tasks = 0;

be/src/exec/scan/file_scanner.h

@@ -21,6 +21,7 @@
 #include <stdint.h>
 
 #include <memory>
+#include <optional>
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
@@ -80,6 +81,10 @@ class FileScanner : public Scanner {
     const VExprContextSPtrs& TEST_runtime_filter_partition_prune_ctxs() const {
         return _runtime_filter_partition_prune_ctxs;
     }
+    static TPushAggOp::type TEST_effective_push_down_agg_type(
+            TPushAggOp::type agg_type, const std::optional<std::vector<int32_t>>& count_slot_ids) {
+        return _effective_push_down_agg_type(agg_type, count_slot_ids);
+    }
 #endif
 
     FileScanner(RuntimeState* state, FileScanLocalState* parent, int64_t limit,
@@ -334,9 +339,28 @@ class FileScanner : public Scanner {
     void _update_adaptive_batch_size_before_truncate(const Block& block);
     void _update_adaptive_batch_size_after_truncate(const Block& block);
 
+    static TPushAggOp::type _effective_push_down_agg_type(
+            TPushAggOp::type agg_type, const std::optional<std::vector<int32_t>>& count_slot_ids) {
+        if (agg_type != TPushAggOp::type::COUNT) {
+            return agg_type;
+        }
+        // V1's CountReader receives only the file's total row count and emits that many synthetic
+        // rows. This is exact for COUNT(*)/COUNT(1), but it has no column metadata for NULL or CAST
+        // semantics. For example, a 10,000-row file with 9,015 non-null values must return 10,000
+        // for COUNT(*) and 9,015 for COUNT(nullable_col); CountReader can produce only the former.
+        // Therefore a non-empty argument list must use the normal reader. nullopt is an old FE plan
+        // that predates the argument field; treating it as empty would silently reinterpret unknown
+        // semantics as COUNT(*).
+        return count_slot_ids.has_value() && count_slot_ids->empty() ? TPushAggOp::type::COUNT
+                                                                     : TPushAggOp::type::NONE;
+    }
+
     TPushAggOp::type _get_push_down_agg_type() const {
-        return _local_state == nullptr ? TPushAggOp::type::NONE
-                                       : _local_state->get_push_down_agg_type();
+        if (_local_state == nullptr) {
+            return TPushAggOp::type::NONE;
+        }
+        return _effective_push_down_agg_type(_local_state->get_push_down_agg_type(),
+                                             _local_state->get_push_down_count_slot_ids());
     }
 
     // enable the file meta cache only when

be/src/exec/scan/file_scanner_v2.cpp

@@ -272,6 +272,10 @@ void FileScannerV2::TEST_report_file_cache_profile(
 bool FileScannerV2::TEST_should_skip_not_found(const Status& status, bool ignore_not_found) {
     return _should_skip_not_found(status, ignore_not_found);
 }
+
+bool FileScannerV2::TEST_should_skip_empty(const Status& status, bool stopped) {
+    return _should_skip_empty(status, stopped);
+}
 #endif
 
 bool FileScannerV2::is_supported(const TFileScanRangeParams& params, const TFileRangeDesc& range) {
@@ -322,6 +326,8 @@ Status FileScannerV2::init(RuntimeState* state, const VExprContextSPtrs& conjunc
     RETURN_IF_ERROR(Scanner::init(state, conjuncts));
     _get_block_timer =
             ADD_TIMER_WITH_LEVEL(_local_state->scanner_profile(), "FileScannerV2GetBlockTime", 1);
+    _empty_file_counter =
+            ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(), "EmptyFileNum", TUnit::UNIT, 1);
     _not_found_file_counter = ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(),
                                                      "NotFoundFileNum", TUnit::UNIT, 1);
     _file_counter =
@@ -394,6 +400,20 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state, Block* block, bool* e
                 *eof = false;
                 continue;
             }
+            if (_should_skip_empty(status, _should_stop || _io_ctx->should_stop)) {
+                // END_OF_FILE here means the reader discovered a valid split with no data while
+                // opening or probing it, not that the Scanner has exhausted all splits. Examples
+                // are a zero-byte CSV with an explicit schema and a Doris Native file containing
+                // only its 12-byte header. Treat it like V1's empty-file path: finish this range,
+                // discard partial reader state, and let the loop fetch the next split.
+                RETURN_IF_ERROR(_table_reader->abort_split());
+                COUNTER_UPDATE(_empty_file_counter, 1);
+                _state->update_num_finished_scan_range(1);
+                _has_prepared_split = false;
+                block->clear_column_data(cast_set<int64_t>(_projected_columns.size()));
+                *eof = false;
+                continue;
+            }
             RETURN_IF_ERROR(status);
         }
         if (*eof) {
@@ -423,9 +443,12 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
 
         const auto format_type = get_range_format_type(*_params, _current_range);
         _init_adaptive_batch_size_state(format_type);
-        if (_should_run_adaptive_batch_size()) {
-            // JNI readers open eagerly in prepare_split(). Seed the probe size first so readers
-            // such as Paimon also use it for their first physical read batch.
+        if (_block_size_predictor != nullptr) {
+            // JNI readers open eagerly in prepare_split(). Always seed the probe before preparing
+            // the next split: its metadata-COUNT decision is not available yet, and the state
+            // exposed by TableReader can still describe the preceding split. Metadata shortcuts
+            // ignore this batch size, while row-scan fallbacks need it for their first physical
+            // read batch.
             _table_reader->set_batch_size(_predict_reader_batch_rows());
         }
         std::map<std::string, Field> partition_values;
@@ -438,6 +461,16 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
             _state->update_num_finished_scan_range(1);
             continue;
         }
+        if (_should_skip_empty(status, _should_stop || _io_ctx->should_stop)) {
+            // Schema discovery can reach EOF before a split becomes prepared. A header-only Native
+            // file follows this path, while a reader that discovers emptiness on its first
+            // get_block() follows the symmetric branch in _get_block_impl(). Both paths must
+            // advance exactly one scan range and preserve later files in the same scan.
+            RETURN_IF_ERROR(_table_reader->abort_split());
+            COUNTER_UPDATE(_empty_file_counter, 1);
+            _state->update_num_finished_scan_range(1);
+            continue;
+        }
         RETURN_IF_ERROR(status);
         if (_table_reader->current_split_pruned()) {
             _state->update_num_finished_scan_range(1);
@@ -458,6 +491,21 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) {
 
     VExprContextSPtrs table_conjuncts;
     RETURN_IF_ERROR(_build_table_conjuncts(&table_conjuncts));
+    std::optional<std::vector<format::GlobalIndex>> push_down_count_columns;
+    const auto& push_down_count_slot_ids = _local_state->get_push_down_count_slot_ids();
+    if (push_down_count_slot_ids.has_value()) {
+        push_down_count_columns.emplace();
+        push_down_count_columns->reserve(push_down_count_slot_ids->size());
+        for (const auto slot_id : *push_down_count_slot_ids) {
+            const auto global_index_it = _slot_id_to_global_index.find(slot_id);
+            if (global_index_it == _slot_id_to_global_index.end()) {
+                return Status::InternalError(
+                        "Pushed-down COUNT argument is not a projected file scan slot, slot_id={}",
+                        slot_id);
+            }
+            push_down_count_columns->push_back(global_index_it->second);
+        }
+    }
     RETURN_IF_ERROR(_table_reader->init({
             .projected_columns = _projected_columns,
             .conjuncts = std::move(table_conjuncts),
@@ -468,6 +516,7 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) {
             .scanner_profile = _local_state->scanner_profile(),
             .file_slot_descs = &_file_slot_descs,
             .push_down_agg_type = _local_state->get_push_down_agg_type(),
+            .push_down_count_columns = std::move(push_down_count_columns),
             .condition_cache_digest = _local_state->get_condition_cache_digest(),
     }));
     return Status::OK();
@@ -540,6 +589,14 @@ bool FileScannerV2::_should_skip_not_found(const Status& status, bool ignore_not
     return ignore_not_found && status.is<ErrorCode::NOT_FOUND>();
 }
 
+bool FileScannerV2::_should_skip_empty(const Status& status, bool stopped) {
+    // Several readers use END_OF_FILE both for a valid zero-row split and for an interrupted IO.
+    // For example, DeletionVectorReader returns END_OF_FILE("stop read.") after try_stop() marks
+    // the shared IOContext. That status must unwind the stopped scanner; counting it as an empty
+    // file would incorrectly finish the scan range and increment EmptyFileNum.
+    return !stopped && status.is<ErrorCode::END_OF_FILE>();
+}
+
 bool FileScannerV2::_should_enable_file_meta_cache() const {
     return ExecEnv::GetInstance()->file_meta_cache()->enabled() &&
            _split_source->num_scan_ranges() < config::max_external_file_meta_cache_num / 3;
@@ -809,10 +866,17 @@ bool FileScannerV2::_should_enable_adaptive_batch_size(TFileFormatType::type for
 }
 
 bool FileScannerV2::_should_run_adaptive_batch_size() const {
-    // COUNT pushdown emits synthetic rows from file metadata and does not materialize file columns,
-    // so there is no useful row-width sample to learn from.
-    return _block_size_predictor != nullptr &&
-           _local_state->get_push_down_agg_type() != TPushAggOp::type::COUNT;
+    DORIS_CHECK(_table_reader != nullptr);
+    return _should_run_adaptive_batch_size(_block_size_predictor != nullptr,
+                                           _table_reader->current_split_uses_metadata_count());
+}
+
+bool FileScannerV2::_should_run_adaptive_batch_size(bool predictor_initialized,
+                                                    bool current_split_uses_metadata_count) {
+    // Metadata COUNT emits synthetic rows and has no physical row width to learn from. A raw COUNT
+    // opcode is not sufficient here: unsupported argument counts, mappings, filters, or deletes
+    // make TableReader fall back to materializing normal rows, which still need adaptive batching.
+    return predictor_initialized && !current_split_uses_metadata_count;
 }
 
 size_t FileScannerV2::_predict_reader_batch_rows() {

be/src/exec/scan/file_scanner_v2.h

@@ -87,6 +87,12 @@ class FileScannerV2 final : public Scanner {
     static void TEST_report_file_cache_profile(
             RuntimeProfile* profile, const io::FileCacheStatistics& file_cache_statistics);
     static bool TEST_should_skip_not_found(const Status& status, bool ignore_not_found);
+    static bool TEST_should_skip_empty(const Status& status, bool stopped);
+    static bool TEST_should_run_adaptive_batch_size(bool predictor_initialized,
+                                                    bool current_split_uses_metadata_count) {
+        return _should_run_adaptive_batch_size(predictor_initialized,
+                                               current_split_uses_metadata_count);
+    }
 #endif
 
     FileScannerV2(RuntimeState* state, FileScanLocalState* parent, int64_t limit,
@@ -121,6 +127,7 @@ class FileScannerV2 final : public Scanner {
     Status _prepare_table_reader_split(const TFileRangeDesc& range,
                                        std::map<std::string, Field> partition_values);
     static bool _should_skip_not_found(const Status& status, bool ignore_not_found);
+    static bool _should_skip_empty(const Status& status, bool stopped);
     bool _should_enable_file_meta_cache() const;
     std::optional<format::GlobalRowIdContext> _create_global_rowid_context(
             const TFileRangeDesc& range) const;
@@ -138,6 +145,8 @@ class FileScannerV2 final : public Scanner {
     void _init_adaptive_batch_size_state(TFileFormatType::type format_type);
     bool _should_enable_adaptive_batch_size(TFileFormatType::type format_type) const;
     bool _should_run_adaptive_batch_size() const;
+    static bool _should_run_adaptive_batch_size(bool predictor_initialized,
+                                                bool current_split_uses_metadata_count);
     size_t _predict_reader_batch_rows();
     void _update_adaptive_batch_size(const Block& block);
     static RealtimeCounterDeltas _collect_realtime_counter_deltas(
@@ -181,6 +190,7 @@ class FileScannerV2 final : public Scanner {
     ShardedKVCache* _kv_cache = nullptr;
 
     RuntimeProfile::Counter* _get_block_timer = nullptr;
+    RuntimeProfile::Counter* _empty_file_counter = nullptr;
     RuntimeProfile::Counter* _not_found_file_counter = nullptr;
     RuntimeProfile::Counter* _file_counter = nullptr;
     RuntimeProfile::Counter* _file_read_bytes_counter = nullptr;