Merge branch 'main' into bump-opendal-54.1

Author: dqhl76

.github/workflows/reuse.sqllogic.yml

@@ -90,11 +90,14 @@ jobs:
       - "${{ inputs.runner_provider }}"
     steps:
       - uses: actions/checkout@v4
+      - name: Setup uv
+        uses: astral-sh/setup-uv@v7
+      - uses: ./.github/actions/setup_minio
       - name: Start UDF Server
+        working-directory: tests/udf
         run: |
-          docker run -d --name minio -p 9000:9000 -p 9001:9001 -e MINIO_ROOT_USER=minioadmin -e MINIO_ROOT_PASSWORD=minioadmin quay.io/minio/minio server /data --console-address ":9001"
-          pip install databend-udf>=0.2.7
-          python3 tests/udf/udf_server.py &
+          uv sync
+          uv run python udf_server.py &
           sleep 2
       - uses: ./.github/actions/test_sqllogic_standalone_linux
         timeout-minutes: 15
@@ -118,10 +121,13 @@ jobs:
       - "${{ inputs.runner_provider }}"
     steps:
       - uses: actions/checkout@v4
+      - name: Setup uv
+        uses: astral-sh/setup-uv@v7
       - name: Start Cloud Control Server
+        working-directory: tests/cloud_control_server
         run: |
-          pip install grpcio grpcio-reflection protobuf grpcio-tools
-          python3 tests/cloud_control_server/simple_server.py &
+          uv sync
+          uv run python simple_server.py &
           sleep 2
       - uses: ./.github/actions/test_cloud_sqllogic_standalone_linux
         timeout-minutes: 15

Cargo.lock

@@ -6,7 +6,7 @@ set -ex
 
 BUILD_PROFILE=${BUILD_PROFILE:-debug}
 
-curl -s https://raw.githubusercontent.com/k3d-io/k3d/main/install.sh | TAG=v5.6.0 bash
+curl -s https://raw.githubusercontent.com/k3d-io/k3d/main/install.sh | TAG=v5.8.3 bash
 k3d registry create registry.localhost --port 0.0.0.0:5111 -i registry:latest
 k3d cluster create --config ./scripts/ci/meta-chaos/k3d.yaml meta-chaos
 
@@ -16,7 +16,7 @@ if kubectl version --client; then
 	echo "kubectl client already installed"
 else
 	echo "install kubectl client"
-	curl -LO "https://dl.k8s.io/release/v1.29.5/bin/linux/amd64/kubectl"
+	curl -LO "https://dl.k8s.io/release/v1.34.1/bin/linux/amd64/kubectl"
 	chmod +x kubectl
 	sudo mv kubectl /usr/local/bin/
 fi

scripts/ci/ci-setup-chaos-meta.sh

@@ -56,7 +56,7 @@ definition = "CREATE FUNCTION ping(STRING) RETURNS STRING LANGUAGE python HANDLE
 [log]
 
 [log.file]
-level = "INFO"
+level = "DEBUG"
 format = "text"
 dir = "./.databend/logs_2"
 limit = 12 # 12 files, 1 file per hour

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

@@ -57,7 +57,7 @@ definition = "CREATE FUNCTION ping(STRING) RETURNS STRING LANGUAGE python HANDLE
 [log]
 
 [log.file]
-level = "INFO"
+level = "DEBUG"
 format = "text"
 limit = 12 # 12 files, 1 file per hour
 dir = "./.databend/logs_3"

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

@@ -3,7 +3,7 @@ kind: Simple
 metadata:
   name: single
 servers: 1
-image: rancher/k3s:v1.29.5-k3s1
+image: rancher/k3s:v1.34.1-k3s1
 ports:
   - port: 8888:80
     nodeFilters:

scripts/ci/meta-chaos/k3d.yaml

@@ -440,7 +440,6 @@ impl Neg for months_days_micros {
     }
 }
 
-/// The in-memory representation of the MonthDayNano variant of the "Interval" logical type.
 #[derive(
     Debug,
     Copy,
@@ -460,12 +459,12 @@ pub struct timestamp_tz(pub i128);
 
 impl Hash for timestamp_tz {
     fn hash<H: Hasher>(&self, state: &mut H) {
-        self.total_micros().hash(state)
+        self.timestamp().hash(state)
     }
 }
 impl PartialEq for timestamp_tz {
     fn eq(&self, other: &Self) -> bool {
-        self.total_micros() == other.total_micros()
+        self.timestamp() == other.timestamp()
     }
 }
 impl PartialOrd for timestamp_tz {
@@ -476,17 +475,18 @@ impl PartialOrd for timestamp_tz {
 
 impl Ord for timestamp_tz {
     fn cmp(&self, other: &Self) -> Ordering {
-        let total_micros = self.total_micros();
-        let other_micros = other.total_micros();
-        total_micros.cmp(&other_micros)
+        let timestamp = self.timestamp();
+        let other_micros = other.timestamp();
+        timestamp.cmp(&other_micros)
     }
 }
 
 impl timestamp_tz {
     pub const MICROS_PER_SECOND: i64 = 1_000_000;
 
+    #[inline]
     pub fn new(timestamp: i64, offset: i32) -> Self {
-        let ts = timestamp as u64 as i128; // <- 中间加一次 u64 屏蔽符号位
+        let ts = timestamp as u64 as i128;
         let off = (offset as i128) << 64;
         Self(off | ts)
     }
@@ -507,32 +507,19 @@ impl timestamp_tz {
     }
 
     #[inline]
-    pub fn hours_offset(&self) -> i8 {
-        (self.seconds_offset() / 3600) as i8
+    pub fn micros_offset_inner(seconds: i64) -> Option<i64> {
+        seconds.checked_mul(Self::MICROS_PER_SECOND)
     }
 
     #[inline]
-    pub fn total_micros(&self) -> i64 {
-        self.try_total_micros().unwrap_or_else(|| {
-            error!(
-                "interval is out of range: timestamp={}, offset={}",
-                self.timestamp(),
-                self.seconds_offset()
-            );
-            0
-        })
-    }
-
-    #[inline]
-    pub fn try_total_micros(&self) -> Option<i64> {
-        let offset_micros = self.micros_offset()?;
-        self.timestamp().checked_sub(offset_micros)
+    pub fn hours_offset(&self) -> i8 {
+        (self.seconds_offset() / 3600) as i8
     }
 }
 
 impl Display for timestamp_tz {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        let timestamp = Timestamp::from_microsecond(self.total_micros()).unwrap();
+        let timestamp = Timestamp::from_microsecond(self.timestamp()).unwrap();
 
         let offset = tz::Offset::from_seconds(self.seconds_offset()).unwrap();
         let string = strtime::format(

src/common/column/src/types/native.rs

@@ -25,7 +25,6 @@ use databend_common_meta_app::data_mask::MaskPolicyTableIdListIdent;
 use databend_common_meta_app::data_mask::MaskpolicyTableIdList;
 use databend_common_meta_app::id_generator::IdGenerator;
 use databend_common_meta_app::row_access_policy::RowAccessPolicyNameIdent;
-use databend_common_meta_app::schema::CreateOption;
 use databend_common_meta_app::tenant::Tenant;
 use databend_common_meta_app::tenant_key::errors::ExistError;
 use databend_common_meta_app::KeyWithTenant;
@@ -71,31 +70,6 @@ impl<KV: kvapi::KVApi<Error = MetaError>> DatamaskApi for KV {
 
         let mut txn = TxnRequest::default();
 
-        let res = self.get_id_and_value(name_ident).await?;
-        debug!(res :? = res, name_key :? =(name_ident); "create_data_mask");
-
-        let mut curr_seq = 0;
-
-        if let Some((seq_id, seq_meta)) = res {
-            match req.create_option {
-                CreateOption::Create => {
-                    return Ok(Err(
-                        name_ident.exist_error(format!("{} already exists", req.name))
-                    ));
-                }
-                CreateOption::CreateIfNotExists => {
-                    return Ok(Ok(CreateDatamaskReply { id: *seq_id.data }));
-                }
-                CreateOption::CreateOrReplace => {
-                    let id_ident = seq_id.data.into_t_ident(name_ident.tenant());
-
-                    txn_delete_exact(&mut txn, &id_ident, seq_meta.seq);
-
-                    curr_seq = seq_id.seq;
-                }
-            };
-        }
-
         // Create data mask by inserting these record:
         // name -> id
         // id -> policy
@@ -114,7 +88,7 @@ impl<KV: kvapi::KVApi<Error = MetaError>> DatamaskApi for KV {
         {
             let meta: DatamaskMeta = req.data_mask_meta.clone();
             let id_list = MaskpolicyTableIdList::default();
-            txn.condition.push(txn_cond_eq_seq(name_ident, curr_seq));
+            txn.condition.push(txn_cond_eq_seq(name_ident, 0));
             txn.condition
                 .push(txn_cond_eq_seq(&row_access_name_ident, 0));
             txn.if_then.extend(vec![

src/meta/api/src/data_mask_api_impl.rs

@@ -70,25 +70,6 @@ impl<KV: kvapi::KVApi<Error = MetaError>> RowAccessPolicyApi for KV {
         let policy_id = RowAccessPolicyId::new(row_access_id);
         let mut txn = TxnRequest::default();
 
-        let res = self.get_id_and_value(name_ident).await?;
-        debug!(res :? = res, name_key :? =(name_ident); "create_row_access");
-
-        let mut curr_seq = 0;
-
-        if let Some((seq_id, seq_meta)) = res {
-            if req.can_replace {
-                let id_ident = seq_id.data.into_t_ident(name_ident.tenant());
-
-                txn_delete_exact(&mut txn, &id_ident, seq_meta.seq);
-
-                // TODO(eason): need to remove row policy from table meta
-
-                curr_seq = seq_id.seq;
-            } else {
-                return Ok(Err(name_ident.exist_error(func_name!())));
-            }
-        }
-
         // Create row policy by inserting these record:
         // name -> id
         // id -> policy
@@ -103,7 +84,7 @@ impl<KV: kvapi::KVApi<Error = MetaError>> RowAccessPolicyApi for KV {
 
         {
             let meta: RowAccessPolicyMeta = req.row_access_policy_meta.clone();
-            txn.condition.push(txn_cond_eq_seq(name_ident, curr_seq));
+            txn.condition.push(txn_cond_eq_seq(name_ident, 0));
             txn.condition.push(txn_cond_eq_seq(&mask_name_ident, 0));
             txn.if_then.extend(vec![
                 txn_op_put_pb(name_ident, &policy_id, None)?, // name -> policy_id

src/meta/api/src/row_access_policy_api_impl.rs

@@ -37,7 +37,6 @@ use databend_common_expression::VIRTUAL_COLUMNS_LIMIT;
 use databend_common_expression::VIRTUAL_COLUMN_ID_START;
 use databend_common_meta_app::app_error::AppError;
 use databend_common_meta_app::data_mask::CreateDatamaskReq;
-use databend_common_meta_app::data_mask::DataMaskIdIdent;
 use databend_common_meta_app::data_mask::DataMaskNameIdent;
 use databend_common_meta_app::data_mask::DatamaskMeta;
 use databend_common_meta_app::data_mask::MaskPolicyIdTableId;
@@ -3184,7 +3183,6 @@ impl SchemaApiTestSuite {
         let mask2_id;
         {
             let req = CreateDatamaskReq {
-                create_option: CreateOption::CreateIfNotExists,
                 name: DataMaskNameIdent::new(tenant.clone(), mask_name_1.to_string()),
                 data_mask_meta: DatamaskMeta {
                     args: vec![],
@@ -3201,7 +3199,6 @@ impl SchemaApiTestSuite {
             mask1_id = get_kv_u64_data(mt, &mask1_name_ident).await?;
 
             let req = CreateDatamaskReq {
-                create_option: CreateOption::CreateIfNotExists,
                 name: DataMaskNameIdent::new(tenant.clone(), mask_name_2.to_string()),
                 data_mask_meta: DatamaskMeta {
                     args: vec![],
@@ -3457,57 +3454,6 @@ impl SchemaApiTestSuite {
             assert!(res.is_none());
         }
 
-        info!("--- create or replace mask policy");
-        {
-            let mask_name = "replace_mask";
-            let name = DataMaskNameIdent::new(tenant.clone(), mask_name);
-            let req = CreateDatamaskReq {
-                create_option: CreateOption::CreateIfNotExists,
-                name: name.clone(),
-                data_mask_meta: DatamaskMeta {
-                    args: vec![],
-                    return_type: "".to_string(),
-                    body: "".to_string(),
-                    comment: Some("before".to_string()),
-                    create_on: created_on,
-                    update_on: None,
-                },
-            };
-            mt.create_data_mask(req).await??;
-            let old_id: u64 = get_kv_u64_data(mt, &name).await?;
-
-            let id_key = DataMaskIdIdent::new(&tenant, old_id);
-
-            let meta: DatamaskMeta = get_kv_data(mt, &id_key).await?;
-            assert_eq!(meta.comment, Some("before".to_string()));
-
-            let req = CreateDatamaskReq {
-                create_option: CreateOption::CreateOrReplace,
-                name: name.clone(),
-                data_mask_meta: DatamaskMeta {
-                    args: vec![],
-                    return_type: "".to_string(),
-                    body: "".to_string(),
-                    comment: Some("after".to_string()),
-                    create_on: created_on,
-                    update_on: None,
-                },
-            };
-            mt.create_data_mask(req).await??;
-
-            // assert old id key has been deleted
-            let meta: Result<DatamaskMeta, KVAppError> = get_kv_data(mt, &id_key).await;
-            assert!(meta.is_err());
-
-            let id: u64 = get_kv_u64_data(mt, &name).await?;
-            assert_ne!(old_id, id);
-
-            let id_key = DataMaskIdIdent::new(&tenant, id);
-
-            let meta: DatamaskMeta = get_kv_data(mt, &id_key).await?;
-            assert_eq!(meta.comment, Some("after".to_string()));
-        }
-
         Ok(())
     }
 
@@ -3549,7 +3495,6 @@ impl SchemaApiTestSuite {
         info!("--- create row access policy");
 
         let req = CreateRowAccessPolicyReq {
-            can_replace: false,
             name: RowAccessPolicyNameIdent::new(tenant.clone(), policy1.to_string()),
             row_access_policy_meta: RowAccessPolicyMeta {
                 args: vec![("number".to_string(), "UInt64".to_string())],
@@ -3566,7 +3511,6 @@ impl SchemaApiTestSuite {
         let policy1_id = res.0.data;
 
         let req = CreateRowAccessPolicyReq {
-            can_replace: false,
             name: RowAccessPolicyNameIdent::new(tenant.clone(), policy2.to_string()),
             row_access_policy_meta: RowAccessPolicyMeta {
                 args: vec![("number".to_string(), "UInt64".to_string())],
@@ -3887,7 +3831,6 @@ impl SchemaApiTestSuite {
         let mask_cleanup_ident =
             DataMaskNameIdent::new(tenant.clone(), mask_cleanup_name.to_string());
         mt.create_data_mask(CreateDatamaskReq {
-            create_option: CreateOption::Create,
             name: mask_cleanup_ident.clone(),
             data_mask_meta: DatamaskMeta {
                 args: vec![],
@@ -3950,7 +3893,6 @@ impl SchemaApiTestSuite {
         // Create another masking policy and bind it.
         let mask_guard_ident = DataMaskNameIdent::new(tenant.clone(), mask_guard_name.to_string());
         mt.create_data_mask(CreateDatamaskReq {
-            create_option: CreateOption::Create,
             name: mask_guard_ident.clone(),
             data_mask_meta: DatamaskMeta {
                 args: vec![],
@@ -4043,7 +3985,6 @@ impl SchemaApiTestSuite {
         let policy_cleanup_ident =
             RowAccessPolicyNameIdent::new(tenant.clone(), policy_cleanup_name.to_string());
         mt.create_row_access_policy(CreateRowAccessPolicyReq {
-            can_replace: false,
             name: policy_cleanup_ident.clone(),
             row_access_policy_meta: RowAccessPolicyMeta {
                 args: vec![("number".to_string(), "UInt64".to_string())],
@@ -4117,7 +4058,6 @@ impl SchemaApiTestSuite {
         let policy_guard_ident =
             RowAccessPolicyNameIdent::new(tenant.clone(), policy_guard_name.to_string());
         mt.create_row_access_policy(CreateRowAccessPolicyReq {
-            can_replace: false,
             name: policy_guard_ident.clone(),
             row_access_policy_meta: RowAccessPolicyMeta {
                 args: vec![("number".to_string(), "UInt64".to_string())],