From ff9c1dfd15bc1dc3db8cd1dae2a7a8f2f41a8d6a Mon Sep 17 00:00:00 2001
From: Gabriel <gabrielleebuaa@gmail.com>
Date: Fri, 19 Jan 2024 16:12:35 +0800
Subject: [PATCH] [pipelineX](filescan) Support parallel executing for external
 table scanning (#30121)

---
 .../java/org/apache/doris/planner/OlapScanNode.java  |  5 +++++
 .../main/java/org/apache/doris/planner/ScanNode.java |  7 +++++--
 .../doris/planner/external/FileQueryScanNode.java    | 12 ++++++++++++
 3 files changed, 22 insertions(+), 2 deletions(-)

diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
index e5381a2018fd84..be2c9a24c91d9f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
@@ -1736,6 +1736,11 @@ public boolean pushDownAggNoGroupingCheckCol(FunctionCallExpr aggExpr, Column co
 
         return true;
     }
+
+    @Override
+    public int getScanRangeNum() {
+        return getScanTabletIds().size();
+    }
 }
 
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
index 7ddf2ae1d4c043..9881fe6f6e3449 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
@@ -725,8 +725,11 @@ public boolean ignoreStorageDataDistribution(ConnectContext context) {
                 && context.getSessionVariable().isIgnoreStorageDataDistribution()
                 && context.getSessionVariable().getEnablePipelineXEngine()
                 && !fragment.isHasNullAwareLeftAntiJoin()
-                && ((this instanceof OlapScanNode) && ((OlapScanNode) this).getScanTabletIds().size()
-                < ConnectContext.get().getSessionVariable().getParallelExecInstanceNum());
+                && getScanRangeNum() < ConnectContext.get().getSessionVariable().getParallelExecInstanceNum();
+    }
+
+    public int getScanRangeNum() {
+        return Integer.MAX_VALUE;
     }
 
     public boolean haveLimitAndConjunts() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java
index 71c731498a643e..4a3c3d2ff9ab85 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java
@@ -485,6 +485,18 @@ protected void genSlotToSchemaIdMapForOrc() {
         params.setSlotNameToSchemaPos(columnNameToPosition);
     }
 
+    @Override
+    public int getScanRangeNum() {
+        Preconditions.checkNotNull(scanRangeLocations);
+        int i = 0;
+        for (TScanRangeLocations tScanRangeLocations : scanRangeLocations) {
+            TScanRange tScanRange = tScanRangeLocations.getScanRange();
+            TFileScanRange tFileScanRange = tScanRange.getExtScanRange().getFileScanRange();
+            i += tFileScanRange.getRangesSize();
+        }
+        return i;
+    }
+
     protected abstract TFileType getLocationType() throws UserException;
 
     protected abstract TFileType getLocationType(String location) throws UserException;