[opt](multi-catalog) Optimize file split size. (#58858)

### What problem does this PR solve?

### Release note

This PR introduces a **dynamic and progressive file split size
adjustment mechanism** to improve scan parallelism and resource
utilization for external table scans, while avoiding excessive small
splits or inefficiently large initial splits.

#### 1. Split Size Adjustment Strategy

##### 1.1 Non-Batch Split Mode

In non-batch split mode, a **two-phase split size selection strategy**
is applied based on the total size of all input files:

* The total size of all splits is calculated in advance.
* If the total size **exceeds `maxInitialSplitNum *
maxInitialSplitSize`**:

  * `split_size = maxSplitSize` (default **64MB**)
* Otherwise:

  * `split_size = maxInitialSplitSize` (default **32MB**)

This strategy reduces the number of splits for small datasets while
improving parallelism for large-scale scans.

---

##### 1.2 Batch Split Mode

In batch split mode, a **progressive split size adjustment strategy** is
introduced:

* As the total file size increases,
* When the number of files gradually **exceeds `maxInitialSplitNum`**,
* The `split_size` is **smoothly increased from `maxInitialSplitSize`
(32MB) toward `maxSplitSize` (64MB)**.

This approach avoids generating too many small splits at the early stage
while gradually increasing scan parallelism as the workload grows,
resulting in more stable scheduling and execution behavior.

---

##### 1.3 User-Specified Split Size (Backward Compatibility)

This PR **preserves the session variable `file_split_size`** for
user-defined split size configuration:

* If `file_split_size` is explicitly set by the user:

  * The user-defined value takes precedence.
  * The dynamic split size adjustment logic is bypassed.
* This ensures full backward compatibility with existing configurations
and tuning practices.

---

#### 2. Support Status by Data Source

| Data Source | Non-Batch Split Mode | Batch Split Mode | Notes |
| ----------- | -------------------- | ---------------- |
----------------------------------------------------- |
| Hive | ✅ Supported | ✅ Supported | Uses Doris internal HDFS
FileSplitter |
| Iceberg | ✅ Supported | ❌ Not supported | File splitting is currently
delegated to Iceberg APIs |
| Paimon | ✅ Supported | ❌ Not supported | Only non-batch split mode is
implemented |

---

#### 3. New Hive HDFS FileSplitter Logic

For Hive HDFS files, this PR introduces an enhanced file splitting
strategy:

1. **Splits never span multiple HDFS blocks**

   * Prevents cross-block reads and avoids unnecessary IO overhead.

2. **Tail split optimization**

   * If the remaining file size is smaller than `split_size * 2`,
   * The remaining part is **evenly divided** into splits,
* Preventing the creation of very small tail splits and improving
overall scan efficiency.

---

#### Summary

* Introduces dynamic and progressive split size adjustment
* Supports both batch and non-batch split modes
* Preserves user-defined split size configuration for backward
compatibility
* Optimizes Hive HDFS file splitting to reduce small tail splits and
cross-block IO

Author: kaka11chen

fe/fe-core/src/main/java/org/apache/doris/datasource/FederationBackendPolicy.java

@@ -63,12 +63,16 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
 public class FederationBackendPolicy {
     private static final Logger LOG = LogManager.getLogger(FederationBackendPolicy.class);
+
+    private static final long FIXED_SHUFFLE_SEED = 123456789L;
+
     protected final List<Backend> backends = Lists.newArrayList();
     private final Map<String, List<Backend>> backendMap = Maps.newHashMap();
 
@@ -220,6 +224,7 @@ public void setEnableSplitsRedistribution(boolean enableSplitsRedistribution) {
     public Multimap<Backend, Split> computeScanRangeAssignment(List<Split> splits) throws UserException {
         ListMultimap<Backend, Split> assignment = ArrayListMultimap.create();
 
+        Collections.shuffle(splits, new Random(FIXED_SHUFFLE_SEED));
         List<Split> remainingSplits;
 
         List<Backend> backends = new ArrayList<>();
@@ -228,8 +233,6 @@ public Multimap<Backend, Split> computeScanRangeAssignment(List<Split> splits) t
         }
         ResettableRandomizedIterator<Backend> randomCandidates = new ResettableRandomizedIterator<>(backends);
 
-        boolean splitsToBeRedistributed = false;
-
         // optimizedLocalScheduling enables prioritized assignment of splits to local nodes when splits contain
         // locality information
         if (Config.split_assigner_optimized_local_scheduling) {
@@ -246,7 +249,6 @@ public Multimap<Backend, Split> computeScanRangeAssignment(List<Split> splits) t
                         assignment.put(selectedBackend, split);
                         assignedWeightPerBackend.put(selectedBackend,
                                 assignedWeightPerBackend.get(selectedBackend) + split.getSplitWeight().getRawValue());
-                        splitsToBeRedistributed = true;
                         continue;
                     }
                 }
@@ -276,7 +278,6 @@ public Multimap<Backend, Split> computeScanRangeAssignment(List<Split> splits) t
                     case CONSISTENT_HASHING: {
                         candidateNodes = consistentHash.getNode(split,
                                 Config.split_assigner_min_consistent_hash_candidate_num);
-                        splitsToBeRedistributed = true;
                         break;
                     }
                     default: {
@@ -302,7 +303,7 @@ public Multimap<Backend, Split> computeScanRangeAssignment(List<Split> splits) t
                     assignedWeightPerBackend.get(selectedBackend) + split.getSplitWeight().getRawValue());
         }
 
-        if (enableSplitsRedistribution && splitsToBeRedistributed) {
+        if (enableSplitsRedistribution) {
             equateDistribution(assignment);
         }
         return assignment;
@@ -499,3 +500,4 @@ public void funnel(Split split, PrimitiveSink primitiveSink) {
         }
     }
 }
+

fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java

@@ -94,6 +94,8 @@ public abstract class FileQueryScanNode extends FileScanNode {
 
     protected TableScanParams scanParams;
 
+    protected FileSplitter fileSplitter;
+
     /**
      * External file scan node for Query hms table
      * needCheckColumnPriv: Some of ExternalFileScanNode do not need to check column priv
@@ -134,6 +136,8 @@ protected void doInitialize() throws UserException {
         }
         initBackendPolicy();
         initSchemaParams();
+        fileSplitter = new FileSplitter(sessionVariable.maxInitialSplitSize, sessionVariable.maxSplitSize,
+                sessionVariable.maxInitialSplitNum);
     }
 
     // Init schema (Tuple/Slot) related params.
@@ -618,19 +622,4 @@ public TableScanParams getScanParams() {
         }
         return this.scanParams;
     }
-
-    /**
-     * The real file split size is determined by:
-     * 1. If user specify the split size in session variable `file_split_size`, use user specified value.
-     * 2. Otherwise, use the max value of DEFAULT_SPLIT_SIZE and block size.
-     * @param blockSize, got from file system, eg, hdfs
-     * @return the real file split size
-     */
-    protected long getRealFileSplitSize(long blockSize) {
-        long realSplitSize = sessionVariable.getFileSplitSize();
-        if (realSplitSize <= 0) {
-            realSplitSize = Math.max(DEFAULT_SPLIT_SIZE, blockSize);
-        }
-        return realSplitSize;
-    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/FileScanNode.java

@@ -62,9 +62,6 @@
  * Base class for External File Scan, including external query and load.
  */
 public abstract class FileScanNode extends ExternalScanNode {
-
-    public static final long DEFAULT_SPLIT_SIZE = 64 * 1024 * 1024; // 64MB
-
     // For explain
     protected long totalFileSize = 0;
     protected long totalPartitionNum = 0;
@@ -115,12 +112,7 @@ public String getNodeExplainString(String prefix, TExplainLevel detailLevel) {
         }
 
         output.append(prefix);
-        boolean isBatch;
-        try {
-            isBatch = isBatchMode();
-        } catch (UserException e) {
-            throw new RuntimeException(e);
-        }
+        boolean isBatch = isBatchMode();
         if (isBatch) {
             output.append("(approximate)");
         }