[scan](improvement) Adjust parallel scanners num (#33223)

be/src/pipeline/exec/file_scan_operator.h

@@ -79,6 +79,8 @@ public:
 
     Status prepare(RuntimeState* state) override;
 
+    bool is_file_scan_operator() const override { return true; }
+
 private:
     friend class FileScanLocalState;
 

be/src/pipeline/exec/scan_operator.cpp

@@ -1218,7 +1218,12 @@ Status ScanLocalState<Derived>::_start_scanners(
     _scanner_ctx = PipXScannerContext::create_shared(
             state(), this, p._output_tuple_desc, p.output_row_descriptor(), scanners, p.limit(),
             state()->scan_queue_mem_limit(), _scan_dependency,
-            p.ignore_data_distribution() ? 1 : state()->query_parallel_instance_num());
+            // 1. If data distribution is ignored , we use 1 instance to scan.
+            // 2. Else if this operator is not file scan operator, we use config::doris_scanner_thread_pool_thread_num scanners to scan.
+            // 3. Else, file scanner will consume much memory so we use config::doris_scanner_thread_pool_thread_num / query_parallel_instance_num scanners to scan.
+            p.ignore_data_distribution() || !p.is_file_scan_operator()
+                    ? 1
+                    : state()->query_parallel_instance_num());
     return Status::OK();
 }
 

be/src/pipeline/exec/scan_operator.h

@@ -394,6 +394,8 @@ public:
     }
     [[nodiscard]] bool is_source() const override { return true; }
 
+    [[nodiscard]] virtual bool is_file_scan_operator() const { return false; }
+
     const std::vector<TRuntimeFilterDesc>& runtime_filter_descs() override {
         return _runtime_filter_descs;
     }