[pipelineX](filescan) Support parallel executing for external table s…

…canning (apache#30121)
s7monk · Jan 19, 2024 · ff9c1df · ff9c1df
1 parent 1321c92
commit ff9c1df
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 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
@@ -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
@@ -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
@@ -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;