feat(query): support sideload data for insert values

Author: everpcpc

scripts/ci/deploy/config/databend-query-node-1.toml

@@ -60,7 +60,7 @@ table_cache_bloom_index_data_bytes=1073741824
 [log]
 
 [log.file]
-level = "ERROR"
+level = "INFO"
 format = "text"
 dir = "./.databend/logs_1"
 

src/query/catalog/src/table_context.rs

@@ -54,6 +54,12 @@ pub struct ProcessInfo {
     pub created_time: SystemTime,
 }
 
+#[derive(Debug, Clone)]
+pub struct SideloadOptions {
+    pub uri: Option<String>,
+    pub stage: Option<String>,
+}
+
 #[async_trait::async_trait]
 pub trait TableContext: Send + Sync {
     /// Build a table instance the plan wants to operate on.
@@ -99,4 +105,5 @@ pub trait TableContext: Send + Sync {
     async fn get_table(&self, catalog: &str, database: &str, table: &str)
     -> Result<Arc<dyn Table>>;
     fn get_processes_info(&self) -> Vec<ProcessInfo>;
+    fn get_sideload(&self) -> Option<SideloadOptions>;
 }

src/query/service/src/interpreters/interpreter_insert_v2.rs

@@ -156,6 +156,26 @@ impl Interpreter for InsertInterpreterV2 {
                         .format
                         .exec_stream(input_context.clone(), &mut build_res.main_pipeline)?;
                 }
+                InsertInputSource::Sideload(opts) => {
+                    tracing::warn!("sideload insert: {:?}", opts);
+                    // TODO:(everpcpc)
+                    let settings = self.ctx.get_settings();
+
+                    build_res.main_pipeline.add_source(
+                        |output| {
+                            let name_resolution_ctx =
+                                NameResolutionContext::try_from(settings.as_ref())?;
+                            let inner = ValueSource::new(
+                                "".to_string(),
+                                self.ctx.clone(),
+                                name_resolution_ctx,
+                                plan.schema(),
+                            );
+                            AsyncSourcer::create(self.ctx.clone(), output, inner)
+                        },
+                        1,
+                    )?;
+                }
                 InsertInputSource::SelectPlan(plan) => {
                     let table1 = table.clone();
                     let (mut select_plan, select_column_bindings) = match plan.as_ref() {

src/query/service/src/servers/http/v1/query/http_query.rs

@@ -21,6 +21,7 @@ use common_base::base::tokio;
 use common_base::base::tokio::sync::Mutex as TokioMutex;
 use common_base::base::tokio::sync::RwLock;
 use common_base::runtime::TrySpawn;
+use common_catalog::table_context::SideloadOptions;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use serde::Deserialize;
@@ -236,6 +237,14 @@ impl HttpQuery {
         let sql = &request.sql;
         tracing::info!("run query_id={id} in session_id={session_id}, sql='{sql}'");
 
+        match &request.sideload {
+            Some(sideload) => ctx.attach_sideload(SideloadOptions {
+                uri: sideload.url.clone(),
+                stage: sideload.stage.clone(),
+            }),
+            None => {}
+        };
+
         let (block_sender, block_receiver) = sized_spsc(request.pagination.max_rows_in_buffer);
         let start_time = Instant::now();
         let state = Arc::new(RwLock::new(Executor {

src/query/service/src/sessions/query_ctx.rs

@@ -32,6 +32,7 @@ use common_catalog::plan::DataSourcePlan;
 use common_catalog::plan::PartInfoPtr;
 use common_catalog::plan::Partitions;
 use common_catalog::plan::StageTableInfo;
+use common_catalog::table_context::SideloadOptions;
 use common_config::DATABEND_COMMIT_VERSION;
 use common_datablocks::DataBlock;
 use common_datavalues::DataValue;
@@ -192,6 +193,10 @@ impl QueryContext {
         self.shared.set_executor(weak_ptr)
     }
 
+    pub fn attach_sideload(&self, sideload: SideloadOptions) {
+        self.shared.attach_sideload(sideload);
+    }
+
     pub fn get_created_time(&self) -> SystemTime {
         self.shared.created_time
     }
@@ -354,6 +359,11 @@ impl TableContext for QueryContext {
     fn get_processes_info(&self) -> Vec<ProcessInfo> {
         SessionManager::instance().processes_info()
     }
+
+    // Get Sideload Options.
+    fn get_sideload(&self) -> Option<SideloadOptions> {
+        self.shared.get_sideload()
+    }
 }
 
 impl TrySpawn for QueryContext {

src/query/service/src/sessions/query_ctx_shared.rs

@@ -22,6 +22,7 @@ use std::time::SystemTime;
 
 use common_base::base::Progress;
 use common_base::runtime::Runtime;
+use common_catalog::table_context::SideloadOptions;
 use common_config::Config;
 use common_datablocks::DataBlock;
 use common_exception::ErrorCode;
@@ -78,6 +79,7 @@ pub struct QueryContextShared {
     pub(in crate::sessions) data_operator: DataOperator,
     pub(in crate::sessions) executor: Arc<RwLock<Weak<PipelineExecutor>>>,
     pub(in crate::sessions) precommit_blocks: Arc<RwLock<Vec<DataBlock>>>,
+    pub(in crate::sessions) sideload_config: Arc<RwLock<Option<SideloadOptions>>>,
     pub(in crate::sessions) created_time: SystemTime,
 }
 
@@ -107,6 +109,7 @@ impl QueryContextShared {
             affect: Arc::new(Mutex::new(None)),
             executor: Arc::new(RwLock::new(Weak::new())),
             precommit_blocks: Arc::new(RwLock::new(vec![])),
+            sideload_config: Arc::new(RwLock::new(None)),
             created_time: SystemTime::now(),
         }))
     }
@@ -316,6 +319,15 @@ impl QueryContextShared {
         swaped_precommit_blocks
     }
 
+    pub fn get_sideload(&self) -> Option<SideloadOptions> {
+        self.sideload_config.read().clone()
+    }
+
+    pub fn attach_sideload(&self, sideload: SideloadOptions) {
+        let mut sideload_config = self.sideload_config.write();
+        *sideload_config = Some(sideload);
+    }
+
     pub fn get_created_time(&self) -> SystemTime {
         self.created_time
     }

src/query/service/tests/it/storages/fuse/operations/commit.rs

@@ -27,6 +27,7 @@ use common_catalog::plan::PartInfoPtr;
 use common_catalog::plan::Partitions;
 use common_catalog::table::Table;
 use common_catalog::table_context::ProcessInfo;
+use common_catalog::table_context::SideloadOptions;
 use common_catalog::table_context::TableContext;
 use common_datablocks::DataBlock;
 use common_exception::ErrorCode;
@@ -453,6 +454,10 @@ impl TableContext for CtxDelegation {
     fn get_processes_info(&self) -> Vec<ProcessInfo> {
         todo!()
     }
+
+    fn get_sideload(&self) -> Option<SideloadOptions> {
+        todo!()
+    }
 }
 
 #[derive(Clone)]

src/query/sql/src/planner/binder/insert.rs

@@ -96,10 +96,13 @@ impl<'a> Binder {
                     opts, start, None,
                 ))
             }
-            InsertSource::Values { rest_str } => {
-                let data = rest_str.trim_end_matches(';').trim_start().to_owned();
-                Ok(InsertInputSource::Values(data))
-            }
+            InsertSource::Values { rest_str } => match self.ctx.get_sideload() {
+                Some(sideload) => Ok(InsertInputSource::Sideload(Arc::new(sideload))),
+                None => {
+                    let data = rest_str.trim_end_matches(';').trim_start().to_owned();
+                    Ok(InsertInputSource::Values(data))
+                }
+            },
             InsertSource::Select { query } => {
                 let statement = Statement::Query(query);
                 let select_plan = self.bind_statement(bind_context, &statement).await?;

src/query/sql/src/planner/plans/insert.rs

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use common_catalog::table_context::SideloadOptions;
 use common_datablocks::DataBlock;
 use common_datavalues::DataSchemaRef;
 use common_meta_types::FileFormatOptions;
@@ -31,6 +32,8 @@ pub enum InsertInputSource {
     StreamingWithFileFormat(FileFormatOptions, usize, Option<Arc<InputContext>>),
     // From cloned String and format
     Values(String),
+    // From sideload stage or uri
+    Sideload(Arc<SideloadOptions>),
 }
 
 #[derive(Clone)]