[improvement](external) Refine Parquet splits by row group in BE

be/src/exec/operator/file_scan_operator.cpp

@@ -152,6 +152,34 @@ bool FileScanLocalState::_should_use_file_scanner_v2(const TQueryOptions& query_
            !is_transactional_hive;
 }
 
+bool FileScanLocalState::_can_generate_physical_splits(const TQueryOptions& query_options,
+                                                       bool is_load,
+                                                       const TFileScanRangeParams& scan_params,
+                                                       const TFileRangeDesc& range) {
+    if (!_should_use_file_scanner_v2(query_options, is_load, scan_params)) {
+        return false;
+    }
+    const auto format = range.__isset.format_type ? range.format_type : scan_params.format_type;
+    if (format == TFileFormatType::FORMAT_PARQUET) {
+        // Keep scanner creation aligned with the downstream refinement guard. Otherwise an
+        // Iceberg delete split creates idle scanners even though it can never publish children.
+        return FileScannerV2::can_refine_source_split(range);
+    }
+    if (format != TFileFormatType::FORMAT_JNI || !range.__isset.table_format_params ||
+        range.table_format_params.table_format_type != "paimon" ||
+        !range.table_format_params.__isset.paimon_params) {
+        return false;
+    }
+    const auto& paimon = range.table_format_params.paimon_params;
+    return paimon.__isset.file_format && paimon.file_format == "parquet" &&
+           !paimon.__isset.paimon_split;
+}
+
+int FileScanLocalState::_adjust_scanner_count(int requested, int initial_ranges,
+                                              bool can_generate_physical_splits) {
+    return can_generate_physical_splits ? requested : std::min(requested, initial_ranges);
+}
+
 Status FileScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
     if (_split_source->num_scan_ranges() == 0) {
         _eos = true;
@@ -258,9 +286,38 @@ void FileScanLocalState::set_scan_ranges(RuntimeState* state,
         if (_split_source == nullptr) {
             _split_source = std::make_shared<LocalSplitSourceConnector>(scan_ranges, _max_scanners);
         }
-        // currently the total number of splits in the bach split mode cannot be accurately obtained,
-        // so we don't do it in the batch split mode.
-        _max_scanners = std::min(_max_scanners, _split_source->num_scan_ranges());
+        // A single FE Parquet split can publish many row-group children after its footer is read.
+        // Keep the requested scanner concurrency in that case; capping it to the initial range
+        // count would leave the generated children serial even though they share one footer.
+        bool can_generate_parquet_splits = false;
+        const TFileScanRangeParams* common_params = nullptr;
+        if (state->get_query_ctx() != nullptr &&
+            state->get_query_ctx()->file_scan_range_params_map.contains(parent_id())) {
+            common_params = &state->get_query_ctx()->file_scan_range_params_map[parent_id()];
+        }
+        for (const auto& scan_range_params : scan_ranges) {
+            const auto& file_scan_range =
+                    scan_range_params.scan_range.ext_scan_range.file_scan_range;
+            const auto* params =
+                    file_scan_range.__isset.params ? &file_scan_range.params : common_params;
+            if (params == nullptr) {
+                continue;
+            }
+            const bool is_load =
+                    state->desc_tbl().get_tuple_descriptor(params->src_tuple_id) != nullptr;
+            can_generate_parquet_splits =
+                    std::ranges::any_of(file_scan_range.ranges, [&](const auto& range) {
+                        return _can_generate_physical_splits(state->query_options(), is_load,
+                                                             *params, range);
+                    });
+            if (can_generate_parquet_splits) {
+                break;
+            }
+        }
+        // Currently the total number of remote splits cannot be accurately obtained, so batch
+        // mode already skips this cap.
+        _max_scanners = _adjust_scanner_count(_max_scanners, _split_source->num_scan_ranges(),
+                                              can_generate_parquet_splits);
     }
 
     if (!scan_ranges.empty() &&

be/src/exec/operator/file_scan_operator.h

@@ -59,6 +59,15 @@ class FileScanLocalState final : public ScanLocalState<FileScanLocalState> {
 #ifdef BE_TEST
     static bool TEST_should_use_file_scanner_v2(const TQueryOptions& query_options, bool is_load,
                                                 const TFileScanRangeParams& scan_params);
+    static bool TEST_can_generate_physical_splits(const TQueryOptions& query_options, bool is_load,
+                                                  const TFileScanRangeParams& scan_params,
+                                                  const TFileRangeDesc& range) {
+        return _can_generate_physical_splits(query_options, is_load, scan_params, range);
+    }
+    static int TEST_adjust_scanner_count(int requested, int initial_ranges,
+                                         bool can_generate_physical_splits) {
+        return _adjust_scanner_count(requested, initial_ranges, can_generate_physical_splits);
+    }
 #endif
 
 private:
@@ -73,6 +82,11 @@ class FileScanLocalState final : public ScanLocalState<FileScanLocalState> {
     bool _push_down_topn(const RuntimePredicate& predicate) override;
     static bool _should_use_file_scanner_v2(const TQueryOptions& query_options, bool is_load,
                                             const TFileScanRangeParams& scan_params);
+    static bool _can_generate_physical_splits(const TQueryOptions& query_options, bool is_load,
+                                              const TFileScanRangeParams& scan_params,
+                                              const TFileRangeDesc& range);
+    static int _adjust_scanner_count(int requested, int initial_ranges,
+                                     bool can_generate_physical_splits);
 
     PushDownType _should_push_down_is_null_predicate(VectorizedFnCall* fn_call) const override {
         return fn_call->fn().name.function_name == "is_null_pred" ||
@@ -94,6 +108,7 @@ class FileScanLocalState final : public ScanLocalState<FileScanLocalState> {
     // 2. parquet file meta
     // KVCache<std::string> _kv_cache;
     std::unique_ptr<ShardedKVCache> _kv_cache;
+    FileContextRegistry _file_context_registry;
     TupleId _output_tuple_id = -1;
     RuntimeProfile::Counter* _condition_cache_hit_counter = nullptr;
     RuntimeProfile::Counter* _condition_cache_filtered_rows_counter = nullptr;