feat: add async function read_file. (#19426)

Author: youngsofun

src/query/functions/src/lib.rs

@@ -54,7 +54,11 @@ pub fn is_cacheable_function(name: &str) -> bool {
 #[ctor]
 pub static BUILTIN_FUNCTIONS: FunctionRegistry = builtin_functions();
 
-pub const ASYNC_FUNCTIONS: [Ascii<&str>; 2] = [Ascii::new("nextval"), Ascii::new("dict_get")];
+pub const ASYNC_FUNCTIONS: [Ascii<&str>; 3] = [
+    Ascii::new("nextval"),
+    Ascii::new("dict_get"),
+    Ascii::new("read_file"),
+];
 
 pub const GENERAL_WITHIN_GROUP_FUNCTIONS: [Ascii<&str>; 5] = [
     Ascii::new("array_agg"),

src/query/service/src/physical_plans/physical_async_func.rs

@@ -114,14 +114,14 @@ impl IPhysicalPlan for AsyncFunction {
         let sequence_counters =
             TransformAsyncFunction::create_sequence_counters(self.async_func_descs.len());
 
-        builder.main_pipeline.add_async_transformer(|| {
+        builder.main_pipeline.try_add_async_transformer(|| {
             TransformAsyncFunction::new(
                 builder.ctx.clone(),
                 self.async_func_descs.clone(),
                 operators.clone(),
                 sequence_counters.clone(),
             )
-        });
+        })?;
 
         Ok(())
     }

src/query/service/src/physical_plans/physical_multi_table_insert.rs

@@ -552,15 +552,14 @@ impl IPhysicalPlan for ChunkFillAndReorder {
                             TransformAsyncFunction::create_sequence_counters(async_funcs.len());
                         let ctx = builder.ctx.clone();
                         plan.async_builder = Some(Box::new(move |input, output| {
+                            let transform = TransformAsyncFunction::new(
+                                ctx.clone(),
+                                async_funcs.clone(),
+                                BTreeMap::new(),
+                                counters.clone(),
+                            )?;
                             Ok(ProcessorPtr::create(AsyncTransformer::create(
-                                input,
-                                output,
-                                TransformAsyncFunction::new(
-                                    ctx.clone(),
-                                    async_funcs.clone(),
-                                    BTreeMap::new(),
-                                    counters.clone(),
-                                ),
+                                input, output, transform,
                             )))
                         }));
 

src/query/service/src/pipelines/builders/builder_fill_missing_columns.rs

@@ -57,12 +57,12 @@ impl PipelineBuilder {
                     TransformAsyncFunction::create_sequence_counters(async_funcs.len());
 
                 pipeline.try_add_async_transformer(|| {
-                    Ok(TransformAsyncFunction::new(
+                    TransformAsyncFunction::new(
                         ctx.clone(),
                         async_funcs.clone(),
                         BTreeMap::new(),
                         sequence_counters.clone(),
-                    ))
+                    )
                 })?;
                 if new_default_schema != new_default_schema_no_cast {
                     pipeline.try_add_transformer(|| {

src/query/service/src/pipelines/builders/builder_mutation.rs

@@ -41,6 +41,7 @@ use databend_common_storages_fuse::operations::UnMatchedExprs;
 
 use crate::pipelines::PipelineBuilder;
 use crate::pipelines::processors::transforms::AsyncFunctionBranch;
+use crate::pipelines::processors::transforms::ReadFileContext;
 use crate::pipelines::processors::transforms::TransformAsyncFunction;
 use crate::pipelines::processors::transforms::TransformBranchedAsyncFunction;
 use crate::pipelines::processors::transforms::TransformResortAddOnWithoutSourceSchema;
@@ -125,6 +126,7 @@ impl PipelineBuilder {
                         Ok(TransformBranchedAsyncFunction {
                             ctx: self.ctx.clone(),
                             branches: branches.clone(),
+                            read_file_ctx: ReadFileContext::try_new(&self.ctx)?,
                         })
                     },
                     transform_len,

src/query/service/src/pipelines/processors/transforms/mod.rs

@@ -48,7 +48,10 @@ pub use new_hash_join::Join;
 pub use new_hash_join::TransformHashJoin;
 pub use new_hash_join::*;
 pub use sort::*;
+pub(crate) use transform_async_function::AutoIncrementNextValFetcher;
+pub(crate) use transform_async_function::ReadFileContext;
 pub use transform_async_function::SequenceCounters;
+pub(crate) use transform_async_function::SequenceNextValFetcher;
 pub use transform_async_function::TransformAsyncFunction;
 pub use transform_branched_async_function::AsyncFunctionBranch;
 pub use transform_branched_async_function::TransformBranchedAsyncFunction;