Merge branch 'main' into warehouse_rbac

.github/actions/pack_binaries/action.yml

@@ -23,6 +23,14 @@ runs:
         category: ${{ inputs.category }}
         path: distro/bin
         artifacts: metactl,meta,query,query.debug
+    - name: Download BendSQL
+      shell: bash
+      env:
+        GH_TOKEN: ${{ github.token }}
+      run: |
+        verison=$(gh release list --repo databendlabs/bendsql | head -n 1 | awk '{print $1}')
+        curl -sSLfo /tmp/bendsql.tar.gz https://github.com/databendlabs/bendsql/releases/download/${verison}/bendsql-${verison}-${{ inputs.target }}.tar.gz
+        tar -xzvf /tmp/bendsql.tar.gz -C distro/bin
     - name: Pack Binaries
       id: pack_binaries
       shell: bash

Cargo.toml

@@ -369,10 +369,10 @@ num-derive = "0.3.3"
 num-traits = "0.2.19"
 num_cpus = "1.13.1"
 object = "0.36.5"
-object_store_opendal = { version = "0.49.0" }
+object_store_opendal = { version = "0.49.0", package = "object_store_opendal", git = "https://github.com/apache/opendal", rev = "78b6a9f" }
 once_cell = "1.15.0"
 openai_api_rust = "0.1"
-opendal = { version = "0.51.1", features = [
+opendal = { version = "0.51.1", package = "opendal", git = "https://github.com/apache/opendal", rev = "78b6a9f", features = [
     "layers-fastrace",
     "layers-prometheus-client",
     "layers-async-backtrace",
@@ -639,7 +639,7 @@ deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "3038c145" }
 ethnum = { git = "https://github.com/datafuse-extras/ethnum-rs", rev = "4cb05f1" }
 openai_api_rust = { git = "https://github.com/datafuse-extras/openai-api", rev = "819a0ed" }
 openraft = { git = "https://github.com/databendlabs/openraft", tag = "v0.10.0-alpha.7" }
-orc-rust = { git = "https://github.com/datafusion-contrib/orc-rust", rev = "dfb1ede" }
+orc-rust = { git = "https://github.com/youngsofun/orc-rust", rev = "6c5ac57" }
 recursive = { git = "https://github.com/datafuse-extras/recursive.git", rev = "6af35a1" }
 sled = { git = "https://github.com/datafuse-extras/sled", tag = "v0.34.7-datafuse.1" }
 tantivy = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370" }

src/query/service/src/pipelines/pipeline_builder.rs

@@ -94,11 +94,13 @@ impl PipelineBuilder {
         }
 
         // unload spill metas
-        self.main_pipeline
-            .set_on_finished(always_callback(move |_info: &ExecutionInfo| {
-                self.ctx.unload_spill_meta();
-                Ok(())
-            }));
+        if !self.ctx.mark_unload_callbacked() {
+            self.main_pipeline
+                .set_on_finished(always_callback(move |_info: &ExecutionInfo| {
+                    self.ctx.unload_spill_meta();
+                    Ok(())
+                }));
+        }
 
         Ok(PipelineBuildResult {
             main_pipeline: self.main_pipeline,

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

@@ -322,6 +322,9 @@ impl QueryContext {
 
     pub fn update_init_query_id(&self, id: String) {
         self.shared.spilled_files.write().clear();
+        self.shared
+            .unload_callbacked
+            .store(false, Ordering::Release);
         self.shared.cluster_spill_progress.write().clear();
         *self.shared.init_query_id.write() = id;
     }
@@ -471,6 +474,12 @@ impl QueryContext {
         Ok(table)
     }
 
+    pub fn mark_unload_callbacked(&self) -> bool {
+        self.shared
+            .unload_callbacked
+            .fetch_or(true, Ordering::SeqCst)
+    }
+
     pub fn unload_spill_meta(&self) {
         const SPILL_META_SUFFIX: &str = ".list";
         let r = self.shared.spilled_files.read();

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

@@ -150,6 +150,7 @@ pub struct QueryContextShared {
     pub(in crate::sessions) cluster_spill_progress: Arc<RwLock<HashMap<String, SpillProgress>>>,
     pub(in crate::sessions) spilled_files:
         Arc<RwLock<HashMap<crate::spillers::Location, crate::spillers::Layout>>>,
+    pub(in crate::sessions) unload_callbacked: AtomicBool,
 }
 
 impl QueryContextShared {
@@ -209,6 +210,7 @@ impl QueryContextShared {
 
             cluster_spill_progress: Default::default(),
             spilled_files: Default::default(),
+            unload_callbacked: AtomicBool::new(false),
             warehouse_cache: Arc::new(RwLock::new(None)),
         }))
     }

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

@@ -182,15 +182,12 @@ impl Binder {
             files: stmt.files.clone(),
             pattern,
         };
-        let required_values_schema: DataSchemaRef = Arc::new(
-            match &stmt.dst_columns {
-                Some(cols) => self.schema_project(&table.schema(), cols)?,
-                None => self.schema_project(&table.schema(), &[])?,
-            }
-            .into(),
-        );
+        let stage_schema = match &stmt.dst_columns {
+            Some(cols) => self.schema_project(&table.schema(), cols)?,
+            None => self.schema_project(&table.schema(), &[])?,
+        };
 
-        let stage_schema = infer_table_schema(&required_values_schema)?;
+        let required_values_schema: DataSchemaRef = Arc::new(stage_schema.clone().into());
 
         let default_values = if stage_info.file_format_params.need_field_default() {
             Some(

src/query/sql/src/planner/expression_parser.rs

@@ -395,7 +395,30 @@ pub fn parse_cluster_keys(
     table_meta: Arc<dyn Table>,
     ast_exprs: Vec<AExpr>,
 ) -> Result<Vec<Expr>> {
-    let exprs = parse_ast_exprs(ctx, table_meta, ast_exprs)?;
+    let schema = table_meta.schema();
+    let (mut bind_context, metadata) = bind_table(table_meta)?;
+    let settings = ctx.get_settings();
+    let name_resolution_ctx = NameResolutionContext::try_from(settings.as_ref())?;
+    let mut type_checker = TypeChecker::try_create(
+        &mut bind_context,
+        ctx,
+        &name_resolution_ctx,
+        metadata,
+        &[],
+        false,
+    )?;
+
+    let exprs: Vec<Expr> = ast_exprs
+        .iter()
+        .map(|ast| {
+            let (scalar, _) = *type_checker.resolve(ast)?;
+            let expr = scalar
+                .as_expr()?
+                .project_column_ref(|col| schema.index_of(&col.column_name).unwrap());
+            Ok(expr)
+        })
+        .collect::<Result<_>>()?;
+
     let mut res = Vec::with_capacity(exprs.len());
     for expr in exprs {
         let inner_type = expr.data_type().remove_nullable();