Merge branch 'main' into fix-incorrect-stats-of-compact-hook

scripts/build/build-debug.sh

@@ -8,5 +8,13 @@ SCRIPT_PATH="$(cd "$(dirname "$0")" >/dev/null 2>&1 && pwd)"
 cd "$SCRIPT_PATH/../.." || exit
 
 echo "Build(DEBUG) start..."
+
+if [[ $(uname -a) =~ Darwin ]]; then
+	export CMAKE_CC_COMPILER=/opt/homebrew/opt/llvm/bin/clang
+	export CMAKE_CXX_COMPILER=/opt/homebrew/opt/llvm/bin/clang++
+	export JEMALLOC_SYS_WITH_LG_PAGE=14
+	export JEMALLOC_SYS_WITH_MALLOC_CONF=oversize_threshold:0,dirty_decay_ms:5000,muzzy_decay_ms:5000
+fi
+
 cargo build --bin=databend-query --bin=databend-meta --bin=databend-metactl --bin=databend-sqllogictests --bin=databend-sqlsmith
 echo "All done..."

src/meta/app/src/principal/user_stage.rs

@@ -569,6 +569,7 @@ pub struct StageInfo {
     /// TODO(xuanwo): stage doesn't have this info anymore, remove it.
     pub number_of_files: u64,
     pub creator: Option<UserIdentity>,
+    pub created_on: DateTime<Utc>,
 }
 
 impl StageInfo {

src/meta/proto-conv/src/stage_from_to_protobuf_impl.rs

@@ -220,6 +220,10 @@ impl FromToProto for mt::principal::StageInfo {
                 Some(c) => Some(mt::principal::UserIdentity::from_pb(c)?),
                 None => None,
             },
+            created_on: match p.created_on {
+                Some(c) => DateTime::<Utc>::from_pb(c)?,
+                None => DateTime::<Utc>::default(),
+            },
         })
     }
 
@@ -241,6 +245,7 @@ impl FromToProto for mt::principal::StageInfo {
                 Some(c) => Some(mt::principal::UserIdentity::to_pb(c)?),
                 None => None,
             },
+            created_on: Some(self.created_on.to_pb()?),
         })
     }
 }

src/meta/proto-conv/src/util.rs

@@ -95,6 +95,7 @@ const META_CHANGE_LOG: &[(u64, &str)] = &[
     (63, "2023-10-30: Add: connection.proto"),
     (64, "2023-11-16: Add: user.proto/NDJsonFileFormatParams add field `missing_field_as` and `null_field_as`", ),
     (65, "2023-11-16: Retype: use Datetime<Utc> instead of u64 to in lvt.time", ),
+    (66, "2023-12-15: Add: stage.proto/StageInfo::created_on", ),
     // Dear developer:
     //      If you're gonna add a new metadata version, you'll have to add a test for it.
     //      You could just copy an existing test file(e.g., `../tests/it/v024_table_meta.rs`)

src/meta/proto-conv/tests/it/main.rs

@@ -69,3 +69,4 @@ mod v062_table_lock_meta;
 mod v063_connection;
 mod v064_ndjson_format_params;
 mod v065_least_visible_time;
+mod v066_stage_create_on;

src/meta/proto-conv/tests/it/user_proto_conv.rs

@@ -103,6 +103,7 @@ pub(crate) fn test_fs_stage_info() -> mt::principal::StageInfo {
             username: "databend".to_string(),
             hostname: "databend.rs".to_string(),
         }),
+        created_on: Utc::now(),
     }
 }
 

src/meta/proto-conv/tests/it/v025_user_stage.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use chrono::DateTime;
+use chrono::Utc;
 use common_meta_app as mt;
 use common_meta_app::principal::UserIdentity;
 use common_meta_app::storage::StorageFsConfig;
@@ -73,6 +75,7 @@ fn test_decode_v25_user_stage() -> anyhow::Result<()> {
             username: "databend".to_string(),
             hostname: "databend.rs".to_string(),
         }),
+        created_on: DateTime::<Utc>::default(),
     };
     common::test_load_old(func_name!(), stage_info_v25.as_slice(), 25, want())?;
     common::test_pb_from_to(func_name!(), want())?;

src/meta/proto-conv/tests/it/v035_user_stage.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use chrono::DateTime;
+use chrono::Utc;
 use common_meta_app as mt;
 use common_meta_app::principal::UserIdentity;
 use common_meta_app::storage::StorageFsConfig;
@@ -70,6 +72,7 @@ fn test_decode_v35_user_stage() -> anyhow::Result<()> {
             username: "databend".to_string(),
             hostname: "databend.rs".to_string(),
         }),
+        created_on: DateTime::<Utc>::default(),
     };
     common::test_load_old(func_name!(), stage_info_v35.as_slice(), 35, want())?;
     common::test_pb_from_to(func_name!(), want())?;

src/meta/proto-conv/tests/it/v042_s3_stage_new_field.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use chrono::DateTime;
+use chrono::Utc;
 use common_meta_app as mt;
 use common_meta_app::principal::UserIdentity;
 use common_meta_app::storage::StorageParams;
@@ -74,6 +76,7 @@ fn test_decode_v42_s3_stage_new_field() -> anyhow::Result<()> {
             username: "databend".to_string(),
             hostname: "databend.rs".to_string(),
         }),
+        created_on: DateTime::<Utc>::default(),
     };
 
     common::test_load_old(func_name!(), stage_info_v42.as_slice(), 42, want())?;

src/meta/proto-conv/tests/it/v066_stage_create_on.rs

@@ -0,0 +1,83 @@
+// Copyright 2023 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use chrono::DateTime;
+use chrono::Utc;
+use common_meta_app as mt;
+use common_meta_app::principal::UserIdentity;
+use common_meta_app::storage::StorageParams;
+use common_meta_app::storage::StorageS3Config;
+use minitrace::func_name;
+
+use crate::common;
+
+// These bytes are built when a new version in introduced,
+// and are kept for backward compatibility test.
+//
+// *************************************************************
+// * These messages should never be updated,                   *
+// * only be added when a new version is added,                *
+// * or be removed when an old version is no longer supported. *
+// *************************************************************
+#[test]
+fn test_decode_v66_stage() -> anyhow::Result<()> {
+    let stage_info_v66 = vec![
+        10, 10, 115, 116, 97, 103, 101, 95, 110, 97, 109, 101, 16, 2, 26, 50, 10, 48, 10, 46, 10,
+        4, 116, 101, 115, 116, 18, 24, 104, 116, 116, 112, 115, 58, 47, 47, 115, 51, 46, 97, 109,
+        97, 122, 111, 110, 97, 119, 115, 46, 99, 111, 109, 42, 4, 116, 101, 115, 116, 104, 1, 160,
+        6, 66, 168, 6, 24, 42, 11, 10, 2, 48, 2, 16, 231, 7, 24, 1, 56, 1, 50, 3, 99, 99, 99, 56,
+        100, 66, 19, 10, 8, 100, 97, 116, 97, 98, 101, 110, 100, 18, 1, 37, 160, 6, 66, 168, 6, 24,
+        74, 8, 10, 6, 160, 6, 66, 168, 6, 24, 82, 23, 50, 48, 50, 51, 45, 49, 50, 45, 49, 53, 32,
+        48, 49, 58, 50, 54, 58, 48, 57, 32, 85, 84, 67, 160, 6, 66, 168, 6, 24,
+    ];
+
+    let want = || mt::principal::StageInfo {
+        stage_name: "stage_name".to_string(),
+        stage_type: mt::principal::StageType::Internal,
+        stage_params: mt::principal::StageParams {
+            storage: StorageParams::S3(StorageS3Config {
+                bucket: "test".to_string(),
+                region: "test".to_string(),
+                allow_anonymous: true,
+                ..Default::default()
+            }),
+        },
+        is_temporary: false,
+        file_format_params: mt::principal::FileFormatParams::Parquet(
+            mt::principal::ParquetFileFormatParams {},
+        ),
+        copy_options: mt::principal::CopyOptions {
+            on_error: mt::principal::OnErrorMode::AbortNum(2),
+            size_limit: 999,
+            max_files: 0,
+            split_size: 0,
+            purge: true,
+            single: false,
+            max_file_size: 0,
+            disable_variant_check: true,
+            return_failed_only: false,
+        },
+        comment: "ccc".to_string(),
+        number_of_files: 100,
+        creator: Some(UserIdentity {
+            username: "databend".to_string(),
+            hostname: "%".to_string(),
+        }),
+        created_on: DateTime::<Utc>::from_timestamp(1702603569, 0).unwrap(),
+    };
+    common::test_pb_from_to(func_name!(), want())?;
+    common::test_load_old(func_name!(), stage_info_v66.as_slice(), 66, want())?;
+
+    Ok(())
+}

src/meta/protos/proto/stage.proto

@@ -74,6 +74,8 @@ message StageInfo {
   optional UserIdentity creator = 8;
 
   FileFormatParams file_format_params = 9;
+
+  optional string created_on = 10;
 }
 
 message StageFile {
@@ -87,4 +89,3 @@ message StageFile {
   optional UserIdentity creator = 5;
   optional string etag = 6;
 }
-

src/query/service/src/interpreters/interpreter_user_stage_create.rs

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use chrono::Utc;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_meta_app::principal::StageType;
@@ -78,6 +79,7 @@ impl Interpreter for CreateUserStageInterpreter {
 
         let mut user_stage = user_stage;
         user_stage.creator = Some(self.ctx.get_current_user()?.identity());
+        user_stage.created_on = Utc::now();
         let _create_stage = user_mgr
             .add_stage(&plan.tenant, user_stage, plan.if_not_exists)
             .await?;

src/query/service/tests/it/storages/testdata/columns_table.txt

@@ -66,6 +66,7 @@ DB.Table: 'system'.'columns', Table: columns-table_id:1, ver:0, Engine: SystemCo
 | 'created_on'                      | 'system'             | 'background_tasks'    | 'Timestamp'           | 'TIMESTAMP'         | ''       | ''       | 'NO'     | ''       |
 | 'created_on'                      | 'system'             | 'indexes'             | 'Timestamp'           | 'TIMESTAMP'         | ''       | ''       | 'NO'     | ''       |
 | 'created_on'                      | 'system'             | 'locks'               | 'Timestamp'           | 'TIMESTAMP'         | ''       | ''       | 'NO'     | ''       |
+| 'created_on'                      | 'system'             | 'stages'              | 'Timestamp'           | 'TIMESTAMP'         | ''       | ''       | 'NO'     | ''       |
 | 'created_on'                      | 'system'             | 'streams'             | 'Timestamp'           | 'TIMESTAMP'         | ''       | ''       | 'NO'     | ''       |
 | 'created_on'                      | 'system'             | 'tables'              | 'Timestamp'           | 'TIMESTAMP'         | ''       | ''       | 'NO'     | ''       |
 | 'created_on'                      | 'system'             | 'tables_with_history' | 'Timestamp'           | 'TIMESTAMP'         | ''       | ''       | 'NO'     | ''       |

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

@@ -349,7 +349,7 @@ impl<'a> Binder {
             })),
 
             // Stages
-            Statement::ShowStages => self.bind_rewrite_to_query(bind_context, "SELECT name, stage_type, number_of_files, creator, comment FROM system.stages ORDER BY name", RewriteKind::ShowStages).await?,
+            Statement::ShowStages => self.bind_rewrite_to_query(bind_context, "SELECT name, stage_type, number_of_files, creator, created_on, comment FROM system.stages ORDER BY name", RewriteKind::ShowStages).await?,
             Statement::ListStage { location, pattern } => {
                 let pattern = if let Some(pattern) = pattern {
                     format!(", pattern => '{pattern}'")

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

@@ -111,12 +111,11 @@ impl UpdatePlan {
                         }
                     }
 
-                    let mut right = right.ok_or_else(|| ErrorCode::Internal("It's a bug"))?;
-                    let right_data_type = right.data_type()?;
+                    let right = right.ok_or_else(|| ErrorCode::Internal("It's a bug"))?;
 
                     // corner case: for merge into, if target_table's fields are not null, when after bind_join, it will
-                    // change into nullable, so we need to cast this.
-                    right = wrap_cast_scalar(&right, &right_data_type, target_type)?;
+                    // change into nullable, so we need to cast this. but we will do cast after all macthed clauses,please
+                    // see `cast_data_type_for_merge()`.
 
                     ScalarExpr::FunctionCall(FunctionCall {
                         span: None,

src/query/storages/system/src/stages_table.rs

@@ -22,6 +22,7 @@ use common_exception::Result;
 use common_expression::types::number::UInt64Type;
 use common_expression::types::NumberDataType;
 use common_expression::types::StringType;
+use common_expression::types::TimestampType;
 use common_expression::utils::FromData;
 use common_expression::DataBlock;
 use common_expression::TableDataType;
@@ -79,6 +80,7 @@ impl AsyncSystemTable for StagesTable {
         let mut comment: Vec<Vec<u8>> = Vec::with_capacity(stages.len());
         let mut number_of_files: Vec<Option<u64>> = Vec::with_capacity(stages.len());
         let mut creator: Vec<Option<Vec<u8>>> = Vec::with_capacity(stages.len());
+        let mut created_on = Vec::with_capacity(stages.len());
         for stage in stages.into_iter() {
             name.push(stage.stage_name.clone().into_bytes());
             stage_type.push(stage.stage_type.clone().to_string().into_bytes());
@@ -95,6 +97,7 @@ impl AsyncSystemTable for StagesTable {
                 }
             };
             creator.push(stage.creator.map(|c| c.to_string().into_bytes().to_vec()));
+            created_on.push(stage.created_on.timestamp_micros());
             comment.push(stage.comment.clone().into_bytes());
         }
 
@@ -106,6 +109,7 @@ impl AsyncSystemTable for StagesTable {
             StringType::from_data(file_format_options),
             UInt64Type::from_opt_data(number_of_files),
             StringType::from_opt_data(creator),
+            TimestampType::from_data(created_on),
             StringType::from_data(comment),
         ]))
     }
@@ -128,6 +132,7 @@ impl StagesTable {
                 "creator",
                 TableDataType::Nullable(Box::new(TableDataType::String)),
             ),
+            TableField::new("created_on", TableDataType::Timestamp),
             TableField::new("comment", TableDataType::String),
         ]);
         let table_info = TableInfo {

tests/sqllogictests/suites/base/05_ddl/05_0016_ddl_stage.test

@@ -19,17 +19,11 @@ CREATE STAGE test_stage_internal file_format=(type=csv compression=AUTO record_d
 statement ok
 LIST @test_stage_internal
 
-onlyif mysql
-query TTTTTITT
+statement ok
 desc stage test_stage_internal
-----
-test_stage_internal Internal StageParams { storage: Fs(StorageFsConfig { root: "_data" }) } CopyOptions { on_error: AbortNum(1), size_limit: 0, max_files: 0, split_size: 0, purge: false, single: false, max_file_size: 0, disable_variant_check: false, return_failed_only: false } Csv(CsvFileFormatParams { compression: Auto, headers: 0, field_delimiter: ",", record_delimiter: "\n", null_display: "\\N", nan_display: "NaN", escape: "\\", quote: "\"", error_on_column_count_mismatch: true }) 0 'root'@'%' (empty)
 
-query TTTTT
+statement ok
 SHOW STAGES
-----
-test_stage External NULL 'root'@'%' (empty)
-test_stage_internal Internal 0 'root'@'%' (empty)
 
 statement ok
 DROP STAGE test_stage

tests/sqllogictests/suites/base/06_show/06_0011_show_stages.test

@@ -4,16 +4,11 @@ DROP STAGE if exists test_stage
 statement ok
 CREATE STAGE test_stage
 
-query TTITT
+statement ok
 SHOW STAGES
-----
-test_stage Internal 0 'root'@'%' (empty)
-
 
-query TTTTTITT
+statement ok
 DESC STAGE test_stage
-----
-test_stage Internal StageParams { storage: Fs(StorageFsConfig { root: "_data" }) } CopyOptions { on_error: AbortNum(1), size_limit: 0, max_files: 0, split_size: 0, purge: false, single: false, max_file_size: 0, disable_variant_check: false, return_failed_only: false } Parquet(ParquetFileFormatParams) 0 'root'@'%' (empty)
 
 statement ok
 DROP STAGE test_stage