feat: add table option copy_dedup_full_path (#17473)

youngsofun · web-flow · commit 149c773b7f56 · 2025-02-18T23:03:20.000+08:00
* add table copy opt `copy_dedup_full_path`.

* rm println!
diff --git a/.github/actions/test_sqllogic_stage/action.yml b/.github/actions/test_sqllogic_stage/action.yml
@@ -6,13 +6,17 @@ inputs:
     required: true
     default: ""
   handlers:
-    description: "logic test handlers, choices: mysql,http,clickhouse"
+    description: "logic test handlers, choices: hybrid,http"
     required: true
     default: ""
   storage:
     description: "storage backend for stage, choices: s3,fs"
     required: true
     default: ""
+  deducp:
+    description: "path type for dedup when copy, choices: full_path,sub_path"
+    required: true
+    default: ""
 runs:
   using: "composite"
   steps:
@@ -43,4 +47,5 @@ runs:
       env:
         TEST_HANDLERS: ${{ inputs.handlers }}
         TEST_STAGE_STORAGE: ${{ inputs.storage }}
+        TEST_STAGE_DEDUP: ${{ inputs.dedup}}
       run: bash ./scripts/ci/ci-run-sqllogic-tests-without-sandbox.sh ${{ inputs.dirs }}
diff --git a/.github/workflows/reuse.sqllogic.yml b/.github/workflows/reuse.sqllogic.yml
@@ -206,19 +206,26 @@ jobs:
         storage:
           - "s3"
           - "fs"
+        handler:
+          - "hybrid"
+          - "http"
+        dedup:
+          - "full_path"
+          - "sub_path"
     steps:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/test_sqllogic_stage
         timeout-minutes: 15
         with:
           storage: ${{ matrix.storage }}
           dirs: stage
-          handlers: http,hybrid
+          handlers: ${{ matrix.handler }}
+          dedup: ${{ matrix.dedup }}
       - name: Upload failure
         if: failure()
         uses: ./.github/actions/artifact_failure
         with:
-          name: test-sqllogic-stage-${{ matrix.storage }}
+          name: test-sqllogic-stage-${{ matrix.storage }}-${{ matrix.handler }}-${{ matrix.dedup }}
 
   standalone_no_table_meta_cache:
     runs-on: [self-hosted, X64, Linux, 2c8g, "${{ inputs.runner_provider }}"]
diff --git a/src/query/catalog/src/table_context.rs b/src/query/catalog/src/table_context.rs
@@ -280,6 +280,7 @@ pub trait TableContext: Send + Sync {
         database_name: &str,
         table_name: &str,
         files: &[StageFileInfo],
+        path_prefix: Option<String>,
         max_files: Option<usize>,
     ) -> Result<FilteredCopyFiles>;
 
diff --git a/src/query/service/src/interpreters/common/table_option_validation.rs b/src/query/service/src/interpreters/common/table_option_validation.rs
@@ -37,6 +37,7 @@ use databend_storages_common_table_meta::table::OPT_KEY_CLUSTER_TYPE;
 use databend_storages_common_table_meta::table::OPT_KEY_COMMENT;
 use databend_storages_common_table_meta::table::OPT_KEY_CONNECTION_NAME;
 use databend_storages_common_table_meta::table::OPT_KEY_DATABASE_ID;
+use databend_storages_common_table_meta::table::OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH;
 use databend_storages_common_table_meta::table::OPT_KEY_ENGINE;
 use databend_storages_common_table_meta::table::OPT_KEY_LOCATION;
 use databend_storages_common_table_meta::table::OPT_KEY_RANDOM_MAX_ARRAY_LEN;
@@ -72,6 +73,7 @@ pub static CREATE_FUSE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(
 
     r.insert("transient");
     r.insert(OPT_KEY_TEMP_PREFIX);
+    r.insert(OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH);
     r
 });
 
@@ -109,6 +111,7 @@ pub static UNSET_TABLE_OPTIONS_WHITE_LIST: LazyLock<HashSet<&'static str>> = Laz
     r.insert(FUSE_OPT_KEY_BLOCK_IN_MEM_SIZE_THRESHOLD);
     r.insert(FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD);
     r.insert(FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS);
+    r.insert(OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH);
     r
 });
 
diff --git a/src/query/service/src/interpreters/interpreter_copy_into_table.rs b/src/query/service/src/interpreters/interpreter_copy_into_table.rs
@@ -232,6 +232,7 @@ impl CopyIntoTableInterpreter {
         duplicated_files_detected: Vec<String>,
         update_stream_meta: Vec<UpdateStreamMetaReq>,
         deduplicated_label: Option<String>,
+        path_prefix: Option<String>,
     ) -> Result<()> {
         let ctx = self.ctx.clone();
         let to_table = ctx
@@ -249,6 +250,7 @@ impl CopyIntoTableInterpreter {
                 to_table.as_ref(),
                 &files_to_copy,
                 &plan.stage_table_info.copy_into_table_options,
+                path_prefix,
             )?;
 
             to_table.commit_insertion(
@@ -371,6 +373,7 @@ impl Interpreter for CopyIntoTableInterpreter {
                 duplicated_files_detected,
                 update_stream_meta,
                 unsafe { self.ctx.get_settings().get_deduplicate_label()? },
+                self.plan.path_prefix.clone(),
             )
             .await?;
         }
diff --git a/src/query/service/src/interpreters/interpreter_table_create.rs b/src/query/service/src/interpreters/interpreter_table_create.rs
@@ -16,6 +16,7 @@ use std::str::FromStr;
 use std::sync::Arc;
 
 use chrono::Utc;
+use databend_common_ast::ast::Engine;
 use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_config::GlobalConfig;
 use databend_common_exception::ErrorCode;
@@ -50,6 +51,7 @@ use databend_storages_common_cache::LoadParams;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::meta::Versioned;
 use databend_storages_common_table_meta::table::OPT_KEY_COMMENT;
+use databend_storages_common_table_meta::table::OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH;
 use databend_storages_common_table_meta::table::OPT_KEY_SNAPSHOT_LOCATION;
 use databend_storages_common_table_meta::table::OPT_KEY_STORAGE_FORMAT;
 use databend_storages_common_table_meta::table::OPT_KEY_STORAGE_PREFIX;
@@ -383,6 +385,17 @@ impl CreateTableInterpreter {
         };
         let schema = TableSchemaRefExt::create(fields);
         let mut options = self.plan.options.clone();
+
+        if self.plan.engine == Engine::Fuse {
+            let settings = self.ctx.get_settings();
+            // change default to 1 when all query server is ready to processing it.
+            if settings.get_copy_dedup_full_path_by_default()? {
+                options.insert(
+                    OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH.to_string(),
+                    "1".to_string(),
+                );
+            };
+        }
         if let Some(storage_format) = options.get(OPT_KEY_STORAGE_FORMAT) {
             FuseStorageFormat::from_str(storage_format)?;
         }
@@ -416,10 +429,12 @@ impl CreateTableInterpreter {
         for table_option in table_meta.options.iter() {
             let key = table_option.0.to_lowercase();
             if !is_valid_create_opt(&key, &self.plan.engine) {
-                error!("invalid opt for fuse table in create table statement");
-                return Err(ErrorCode::TableOptionInvalid(format!(
-                    "table option {key} is invalid for create table statement",
-                )));
+                let msg = format!(
+                    "table option {key} is invalid for create table statement with engine {}",
+                    self.plan.engine
+                );
+                error!("{msg}");
+                return Err(ErrorCode::TableOptionInvalid(msg));
             }
         }
 
diff --git a/src/query/service/src/pipelines/builders/builder_copy_into_table.rs b/src/query/service/src/pipelines/builders/builder_copy_into_table.rs
@@ -197,6 +197,7 @@ impl PipelineBuilder {
         to_table: &dyn Table,
         copied_files: &[StageFileInfo],
         options: &CopyIntoTableOptions,
+        path_prefix: Option<String>,
     ) -> Result<Option<UpsertTableCopiedFileReq>> {
         let mut copied_file_tree = BTreeMap::new();
         for file in copied_files {
@@ -205,7 +206,12 @@ impl PipelineBuilder {
                 v.truncate(7);
                 v
             });
-            copied_file_tree.insert(file.path.clone(), TableCopiedFileInfo {
+            let path = if let Some(p) = &path_prefix {
+                format!("{}{}", p, file.path)
+            } else {
+                file.path.clone()
+            };
+            copied_file_tree.insert(path, TableCopiedFileInfo {
                 etag: short_etag,
                 content_length: file.size,
                 last_modified: Some(file.last_modified),
diff --git a/src/query/service/src/sessions/query_ctx.rs b/src/query/service/src/sessions/query_ctx.rs
@@ -1169,6 +1169,7 @@ impl TableContext for QueryContext {
         database_name: &str,
         table_name: &str,
         files: &[StageFileInfo],
+        path_prefix: Option<String>,
         max_files: Option<usize>,
     ) -> Result<FilteredCopyFiles> {
         if files.is_empty() {
@@ -1195,8 +1196,20 @@ impl TableContext for QueryContext {
         let mut duplicated_files = Vec::with_capacity(files.len());
 
         for chunk in files.chunks(batch_size) {
-            let files = chunk.iter().map(|v| v.path.clone()).collect::<Vec<_>>();
-            let req = GetTableCopiedFileReq { table_id, files };
+            let files = chunk
+                .iter()
+                .map(|v| {
+                    if let Some(p) = &path_prefix {
+                        format!("{}{}", p, v.path)
+                    } else {
+                        v.path.clone()
+                    }
+                })
+                .collect::<Vec<_>>();
+            let req = GetTableCopiedFileReq {
+                table_id,
+                files: files.clone(),
+            };
             let start_request = Instant::now();
             let copied_files = catalog
                 .get_table_copied_file_info(&tenant, database_name, req)
@@ -1207,8 +1220,8 @@ impl TableContext for QueryContext {
                 Instant::now().duration_since(start_request).as_millis() as u64,
             );
             // Colored
-            for file in chunk {
-                if !copied_files.contains_key(&file.path) {
+            for (file, key) in chunk.iter().zip(files.iter()) {
+                if !copied_files.contains_key(key) {
                     files_to_copy.push(file.clone());
                     result_size += 1;
                     if result_size == max_files {
diff --git a/src/query/service/tests/it/sql/exec/get_table_bind_test.rs b/src/query/service/tests/it/sql/exec/get_table_bind_test.rs
@@ -840,6 +840,7 @@ impl TableContext for CtxDelegation {
         _database_name: &str,
         _table_name: &str,
         _files: &[StageFileInfo],
+        _path_prefix: Option<String>,
         _max_files: Option<usize>,
     ) -> Result<FilteredCopyFiles> {
         todo!()
diff --git a/src/query/service/tests/it/storages/fuse/operations/commit.rs b/src/query/service/tests/it/storages/fuse/operations/commit.rs
@@ -715,6 +715,7 @@ impl TableContext for CtxDelegation {
         _database_name: &str,
         _table_name: &str,
         _files: &[StageFileInfo],
+        _path_prefix: Option<String>,
         _max_files: Option<usize>,
     ) -> Result<FilteredCopyFiles> {
         todo!()
diff --git a/src/query/settings/src/settings_default.rs b/src/query/settings/src/settings_default.rs
@@ -1193,6 +1193,13 @@ impl DefaultSettings {
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(0..=1)),
                 }),
+                ("copy_dedup_full_path_by_default", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(0),
+                    desc: "The default value if table option `copy_dedup_full_path` is not set when creating table.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
+                }),
 
             ]);
 
diff --git a/src/query/settings/src/settings_getter_setter.rs b/src/query/settings/src/settings_getter_setter.rs
@@ -878,4 +878,8 @@ impl Settings {
     pub fn get_hilbert_sample_size_per_block(&self) -> Result<u64> {
         self.try_get_u64("hilbert_sample_size_per_block")
     }
+
+    pub fn get_copy_dedup_full_path_by_default(&self) -> Result<bool> {
+        Ok(self.try_get_u64("copy_dedup_full_path_by_default")? == 1)
+    }
 }
diff --git a/src/query/sql/src/planner/binder/copy_into_table.rs b/src/query/sql/src/planner/binder/copy_into_table.rs
@@ -59,6 +59,7 @@ use databend_common_meta_app::principal::StageInfo;
 use databend_common_meta_app::principal::COPY_MAX_FILES_PER_COMMIT;
 use databend_common_storage::StageFilesInfo;
 use databend_common_users::UserApiProvider;
+use databend_storages_common_table_meta::table::OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH;
 use derive_visitor::Drive;
 use log::debug;
 use log::warn;
@@ -151,6 +152,12 @@ impl Binder {
             .ctx
             .get_table(&catalog_name, &database_name, &table_name)
             .await?;
+        let dedup_full_path = table
+            .get_table_info()
+            .meta
+            .options
+            .get(OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH)
+            == Some(&"1".to_string());
 
         let validation_mode = ValidationMode::from_str(stmt.options.validation_mode.as_str())
             .map_err(ErrorCode::SyntaxException)?;
@@ -204,6 +211,8 @@ impl Binder {
             table_name,
             validation_mode,
             is_transform,
+            dedup_full_path,
+            path_prefix: None,
             no_file_to_copy: false,
             from_attachment: false,
             stage_table_info: StageTableInfo {
@@ -383,6 +392,8 @@ impl Binder {
             from_attachment: true,
             required_source_schema: data_schema.clone(),
             required_values_schema,
+            dedup_full_path: false,
+            path_prefix: None,
             values_consts: const_columns,
             stage_table_info: StageTableInfo {
                 schema: stage_schema,
diff --git a/src/query/sql/src/planner/plans/copy_into_table.rs b/src/query/sql/src/planner/plans/copy_into_table.rs
@@ -37,6 +37,7 @@ use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_metrics::storage::*;
 use databend_common_storage::init_stage_operator;
 use log::info;
+use opendal::Operator;
 
 use crate::plans::Plan;
 
@@ -139,6 +140,19 @@ pub struct CopyIntoTablePlan {
     pub enable_distributed: bool,
 
     pub files_collected: bool,
+
+    pub dedup_full_path: bool,
+    pub path_prefix: Option<String>,
+}
+
+fn get_path_prefix(op: &Operator) -> String {
+    let info = op.info();
+    let p = format!("{}://{}{}", info.scheme(), info.name(), info.root());
+    if p.ends_with('/') {
+        p
+    } else {
+        format!("{}/", p)
+    }
 }
 
 impl CopyIntoTablePlan {
@@ -207,6 +221,10 @@ impl CopyIntoTablePlan {
             ctx.set_status_info("begin filtering out copied files");
 
             let filter_start = Instant::now();
+            if self.dedup_full_path {
+                let prefix = get_path_prefix(&operator);
+                self.path_prefix = Some(prefix.clone());
+            };
             let FilteredCopyFiles {
                 files_to_copy,
                 duplicated_files,
@@ -216,6 +234,7 @@ impl CopyIntoTablePlan {
                     &self.database_name,
                     &self.table_name,
                     &all_source_file_infos,
+                    self.path_prefix.clone(),
                     max_files,
                 )
                 .await?;
diff --git a/src/query/storages/common/table_meta/src/table/table_keys.rs b/src/query/storages/common/table_meta/src/table/table_keys.rs
@@ -55,6 +55,7 @@ pub const OPT_KEY_RANDOM_MAX_STRING_LEN: &str = "max_string_len";
 pub const OPT_KEY_RANDOM_MAX_ARRAY_LEN: &str = "max_array_len";
 
 pub const OPT_KEY_CLUSTER_TYPE: &str = "cluster_type";
+pub const OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH: &str = "copy_dedup_full_path";
 pub const LINEAR_CLUSTER_TYPE: &str = "linear";
 pub const HILBERT_CLUSTER_TYPE: &str = "hilbert";
 
diff --git a/tests/data/unload/parquet/null_If/data_ce70d99e-4e06-421d-ad84-5a9ac3d7763a_0000_00000000.parquet b/tests/data/unload/parquet/null_If/data_ce70d99e-4e06-421d-ad84-5a9ac3d7763a_0000_00000000.parquet
diff --git a/tests/sqllogictests/scripts/prepare_stage.sh b/tests/sqllogictests/scripts/prepare_stage.sh
@@ -20,3 +20,14 @@ fi
 
 echo "drop table if exists ontime" | $BENDSQL_CLIENT_CONNECT
 cat tests/data/ddl/ontime.sql | $BENDSQL_CLIENT_CONNECT
+
+if [ "$TEST_STAGE_DEDUP" = "full_path" ]; then
+    FULL_PATH=1
+elif [ "$TEST_STAGE_DEDUP" = "sub_path" ]; then
+    FULL_PATH=0
+else
+    echo "TEST_STAGE_DEDUP must be 'full_path' or 'sub_path'" >&2
+    exit 1
+fi
+
+echo "set global copy_dedup_full_path_by_default = ${FULL_PATH}" | $BENDSQL_CLIENT_CONNECT
diff --git a/tests/sqllogictests/suites/stage/dedup_full_path.test b/tests/sqllogictests/suites/stage/dedup_full_path.test
