diff --git a/src/query/functions/src/scalars/hash.rs b/src/query/functions/src/scalars/hash.rs index c93ed7c5a497a..c723c95ab0a37 100644 --- a/src/query/functions/src/scalars/hash.rs +++ b/src/query/functions/src/scalars/hash.rs @@ -12,7 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::collections::hash_map::DefaultHasher; use std::hash::Hash; use std::hash::Hasher; @@ -48,6 +47,7 @@ use md5::Digest; use md5::Md5 as Md5Hasher; use naive_cityhash::cityhash64_with_seed; use num_traits::AsPrimitive; +use siphasher::sip::SipHasher13; use twox_hash::XxHash32; use twox_hash::XxHash64; @@ -70,7 +70,7 @@ pub fn register(registry: &mut FunctionRegistry) { NumberClass::Decimal128 => { struct Siphash64; impl HashFunction for Siphash64 { - type Hasher = DefaultHasher; + type Hasher = SipHasher13; fn name() -> &'static str { "siphash64" } @@ -102,6 +102,8 @@ pub fn register(registry: &mut FunctionRegistry) { }); } + register_bucket(registry); + // Could be used in exchange registry .scalar_builder("siphash64") @@ -113,9 +115,7 @@ pub fn register(registry: &mut FunctionRegistry) { GenericType<0>, NumberType, >(|val, output, _| { - let mut hasher = DefaultHasher::default(); - DFHash::hash(&val.to_owned(), &mut hasher); - output.push(hasher.finish()); + output.push(siphash64(&val.to_owned())); })) .register(); @@ -222,6 +222,93 @@ pub fn register(registry: &mut FunctionRegistry) { ); } +fn register_bucket(registry: &mut FunctionRegistry) { + registry + .register_passthrough_nullable_2_arg::, StringType, NumberType, _, _>( + "bucket", + |_, _, _| FunctionDomain::MayThrow, + vectorize_with_builder_2_arg::, StringType, NumberType>( + |buckets, value, output, ctx| { + output.push(bucket(buckets, value, output.len(), ctx)); + }, + ), + ); + registry + .register_passthrough_nullable_2_arg::, DateType, NumberType, _, _>( + "bucket", + |_, _, _| FunctionDomain::MayThrow, + vectorize_with_builder_2_arg::, DateType, NumberType>( + |buckets, value, output, ctx| { + output.push(bucket(buckets, &value, output.len(), ctx)); + }, + ), + ); + registry.register_passthrough_nullable_2_arg::< + NumberType, + TimestampType, + NumberType, + _, + _, + >( + "bucket", + |_, _, _| FunctionDomain::MayThrow, + vectorize_with_builder_2_arg::, TimestampType, NumberType>( + |buckets, value, output, ctx| { + output.push(bucket(buckets, &value, output.len(), ctx)); + }, + ), + ); + for num_type in ALL_INTEGER_TYPES { + with_integer_mapped_type!(|NUM_TYPE| match num_type { + NumberDataType::NUM_TYPE => { + registry.register_passthrough_nullable_2_arg::< + NumberType, + NumberType, + NumberType, + _, + _, + >( + "bucket", + |_, _, _| FunctionDomain::MayThrow, + vectorize_with_builder_2_arg::< + NumberType, + NumberType, + NumberType, + >(|buckets, value, output, ctx| { + output.push(bucket(buckets, &value, output.len(), ctx)); + }), + ); + } + _ => unreachable!(), + }); + } +} + +fn bucket( + buckets: u64, + value: &T, + row: usize, + ctx: &mut databend_common_expression::EvalContext, +) -> u32 { + let Ok(buckets) = u32::try_from(buckets) else { + ctx.set_error(row, "bucket count must be between 1 and 4294967295"); + return 0; + }; + if buckets == 0 { + ctx.set_error(row, "bucket count must be greater than zero"); + return 0; + } + + (siphash64(value) % buckets as u64) as u32 +} + +fn siphash64(value: &T) -> u64 { + // Keep SQL hash results independent of Rust's unspecified DefaultHasher. + let mut hasher = SipHasher13::new_with_keys(0, 0); + DFHash::hash(value, &mut hasher); + hasher.finish() +} + fn register_simple_domain_type_hash(registry: &mut FunctionRegistry) where for<'a> T::ScalarRef<'a>: DFHash { registry @@ -232,9 +319,7 @@ where for<'a> T::ScalarRef<'a>: DFHash { .calc_domain(|_, _| FunctionDomain::Full) .vectorized(vectorize_with_builder_1_arg::>( |val, output, _| { - let mut hasher = DefaultHasher::default(); - DFHash::hash(&val, &mut hasher); - output.push(hasher.finish()); + output.push(siphash64(&val)); }, )) .register(); diff --git a/src/query/functions/tests/it/scalars/hash.rs b/src/query/functions/tests/it/scalars/hash.rs index 0a8439e7a39f4..a12a000b87bf3 100644 --- a/src/query/functions/tests/it/scalars/hash.rs +++ b/src/query/functions/tests/it/scalars/hash.rs @@ -33,6 +33,19 @@ fn test_hash() { test_siphash64(file); test_xxhash64(file); test_xxhash32(file); + test_bucket(file); +} + +fn test_bucket(file: &mut impl Write) { + run_ast(file, "bucket(16, 34)", &[]); + run_ast(file, "bucket(16, 'iceberg')", &[]); + run_ast(file, "bucket(16, to_date(10000))", &[]); + run_ast(file, "bucket(16, to_datetime(100000))", &[]); + run_ast(file, "bucket(0, 1)", &[]); + run_ast(file, "bucket(8, a)", &[( + "a", + Int64Type::from_data(vec![1, 2, 3, 4]), + )]); } fn test_md5(file: &mut impl Write) { diff --git a/src/query/functions/tests/it/scalars/testdata/function_list.txt b/src/query/functions/tests/it/scalars/testdata/function_list.txt index 5879be79072ec..3757a33643be1 100644 --- a/src/query/functions/tests/it/scalars/testdata/function_list.txt +++ b/src/query/functions/tests/it/scalars/testdata/function_list.txt @@ -934,6 +934,28 @@ Functions overloads: 1 bitmap_xor(Bitmap NULL, Bitmap NULL) :: Bitmap NULL 0 blake3(String) :: String 1 blake3(String NULL) :: String NULL +0 bucket(UInt64, String) :: UInt32 +1 bucket(UInt64 NULL, String NULL) :: UInt32 NULL +2 bucket(UInt64, Date) :: UInt32 +3 bucket(UInt64 NULL, Date NULL) :: UInt32 NULL +4 bucket(UInt64, Timestamp) :: UInt32 +5 bucket(UInt64 NULL, Timestamp NULL) :: UInt32 NULL +6 bucket(UInt64, UInt8) :: UInt32 +7 bucket(UInt64 NULL, UInt8 NULL) :: UInt32 NULL +8 bucket(UInt64, UInt16) :: UInt32 +9 bucket(UInt64 NULL, UInt16 NULL) :: UInt32 NULL +10 bucket(UInt64, UInt32) :: UInt32 +11 bucket(UInt64 NULL, UInt32 NULL) :: UInt32 NULL +12 bucket(UInt64, UInt64) :: UInt32 +13 bucket(UInt64 NULL, UInt64 NULL) :: UInt32 NULL +14 bucket(UInt64, Int8) :: UInt32 +15 bucket(UInt64 NULL, Int8 NULL) :: UInt32 NULL +16 bucket(UInt64, Int16) :: UInt32 +17 bucket(UInt64 NULL, Int16 NULL) :: UInt32 NULL +18 bucket(UInt64, Int32) :: UInt32 +19 bucket(UInt64 NULL, Int32 NULL) :: UInt32 NULL +20 bucket(UInt64, Int64) :: UInt32 +21 bucket(UInt64 NULL, Int64 NULL) :: UInt32 NULL 0 build_bitmap(Array(UInt8 NULL)) :: Bitmap 1 build_bitmap(Array(UInt8 NULL) NULL) :: Bitmap NULL 2 build_bitmap(Array(UInt16 NULL)) :: Bitmap diff --git a/src/query/functions/tests/it/scalars/testdata/hash.txt b/src/query/functions/tests/it/scalars/testdata/hash.txt index f6a46b45593a5..92aab3a3b6572 100644 --- a/src/query/functions/tests/it/scalars/testdata/hash.txt +++ b/src/query/functions/tests/it/scalars/testdata/hash.txt @@ -851,3 +851,71 @@ evaluation (internal): +--------+----------------------------------+ +ast : bucket(16, 34) +raw expr : bucket(16, 34) +checked expr : bucket(CAST(16_u8 AS UInt64), 34_u8) +optimized expr : 5_u32 +output type : UInt32 +output domain : {5..=5} +output : 5 + + +ast : bucket(16, 'iceberg') +raw expr : bucket(16, 'iceberg') +checked expr : bucket(CAST(16_u8 AS UInt64), "iceberg") +optimized expr : 6_u32 +output type : UInt32 +output domain : {6..=6} +output : 6 + + +ast : bucket(16, to_date(10000)) +raw expr : bucket(16, to_date(10000)) +checked expr : bucket(CAST(16_u8 AS UInt64), CAST(CAST(10000_u16 AS Int64) AS Date)) +optimized expr : 0_u32 +output type : UInt32 +output domain : {0..=0} +output : 0 + + +ast : bucket(16, to_datetime(100000)) +raw expr : bucket(16, to_datetime(100000)) +checked expr : bucket(CAST(16_u8 AS UInt64), CAST(CAST(100000_u32 AS Int64) AS Timestamp)) +optimized expr : 11_u32 +output type : UInt32 +output domain : {11..=11} +output : 11 + + +error: + --> SQL:1:1 + | +1 | bucket(0, 1) + | ^^^^^^^^^^^^ bucket count must be greater than zero while evaluating function `bucket(0, 1)` in expr `bucket(CAST(0 AS UInt64), 1)` + + + +ast : bucket(8, a) +raw expr : bucket(8, a::Int64) +checked expr : bucket(CAST(8_u8 AS UInt64), a) +optimized expr : bucket(8_u64, a) +evaluation: ++--------+---------+---------+ +| | a | Output | ++--------+---------+---------+ +| Type | Int64 | UInt32 | +| Domain | {1..=4} | Unknown | +| Row 0 | 1 | 1 | +| Row 1 | 2 | 6 | +| Row 2 | 3 | 2 | +| Row 3 | 4 | 1 | ++--------+---------+---------+ +evaluation (internal): ++--------+-----------------------------+ +| Column | Data | ++--------+-----------------------------+ +| a | Column(Int64([1, 2, 3, 4])) | +| Output | UInt32([1, 6, 2, 1]) | ++--------+-----------------------------+ + + diff --git a/src/query/service/src/interpreters/common/table_option_validation.rs b/src/query/service/src/interpreters/common/table_option_validation.rs index cf8e0b3386ff5..2790c1b784998 100644 --- a/src/query/service/src/interpreters/common/table_option_validation.rs +++ b/src/query/service/src/interpreters/common/table_option_validation.rs @@ -72,6 +72,7 @@ use databend_storages_common_table_meta::table::OPT_KEY_SEGMENT_FORMAT; use databend_storages_common_table_meta::table::OPT_KEY_STORAGE_FORMAT; use databend_storages_common_table_meta::table::OPT_KEY_TABLE_COMPRESSION; use databend_storages_common_table_meta::table::OPT_KEY_TEMP_PREFIX; +use databend_storages_common_table_meta::table::OPT_KEY_WRITE_DISTRIBUTION_MODE; pub use databend_storages_common_table_meta::table::analyze_count_min_sketch_error_rate_from_options; pub use databend_storages_common_table_meta::table::analyze_top_n_size_from_options; use log::error; @@ -101,6 +102,7 @@ pub static CREATE_FUSE_OPTIONS: LazyLock> = LazyLock::new( r.insert(OPT_KEY_DATABASE_ID); r.insert(OPT_KEY_COMMENT); r.insert(OPT_KEY_CHANGE_TRACKING); + r.insert(OPT_KEY_WRITE_DISTRIBUTION_MODE); r.insert(OPT_KEY_ENGINE); diff --git a/src/query/service/src/interpreters/interpreter_insert.rs b/src/query/service/src/interpreters/interpreter_insert.rs index df8cea1bcf509..709ea224bd9c7 100644 --- a/src/query/service/src/interpreters/interpreter_insert.rs +++ b/src/query/service/src/interpreters/interpreter_insert.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::BTreeSet; use std::sync::Arc; use chrono::Duration; @@ -32,6 +33,7 @@ use databend_common_expression::RemoteExpr; use databend_common_expression::SendableDataBlockStream; use databend_common_expression::types::DataType; use databend_common_expression::types::UInt64Type; +use databend_common_functions::BUILTIN_FUNCTIONS; use databend_common_meta_app::schema::Constraint; use databend_common_pipeline::sources::AsyncSourcer; use databend_common_pipeline_transforms::TransformPipelineHelper; @@ -39,6 +41,7 @@ use databend_common_pipeline_transforms::TransformPipelineHelper; use databend_common_pipeline_transforms::columns::TransformAddConstColumns; use databend_common_sql::ColumnBinding; use databend_common_sql::ColumnBindingBuilder; +use databend_common_sql::MetadataRef; use databend_common_sql::NameResolutionContext; use databend_common_sql::Symbol; use databend_common_sql::Visibility; @@ -49,6 +52,7 @@ use databend_common_sql::plans::Insert; use databend_common_sql::plans::InsertInputSource; use databend_common_sql::plans::InsertValue; use databend_common_sql::plans::Plan; +use databend_common_storages_fuse::FuseTable; use databend_common_storages_fuse::operations::TransformConstraintVerify; use databend_common_storages_paimon::PAIMON_ENGINE; use databend_common_storages_paimon::PaimonTable; @@ -65,15 +69,16 @@ use crate::interpreters::common::check_deduplicate_label; use crate::interpreters::common::dml_build_update_stream_req; use crate::physical_plans::ConstantTableScan; use crate::physical_plans::DistributedInsertSelect; +use crate::physical_plans::EvalScalar; use crate::physical_plans::Exchange; use crate::physical_plans::IPhysicalPlan; use crate::physical_plans::PAIMON_ROUTE_KEY_NAME; -use crate::physical_plans::PaimonWritePrepare; use crate::physical_plans::PaimonWriteRoute; use crate::physical_plans::PhysicalPlan; use crate::physical_plans::PhysicalPlanBuilder; use crate::physical_plans::PhysicalPlanCast; use crate::physical_plans::PhysicalPlanMeta; +use crate::physical_plans::TableWritePrepare; use crate::pipelines::PipelineBuildResult; use crate::pipelines::PipelineBuilder; use crate::pipelines::RawValueSource; @@ -123,6 +128,25 @@ fn wrap_paimon_write_distribution( })) } +fn prepare_table_write_input( + input: PhysicalPlan, + table: &Arc, + select_schema: DataSchemaRef, + select_column_bindings: Vec, + insert_schema: DataSchemaRef, + cast_needed: bool, +) -> PhysicalPlan { + PhysicalPlan::new(TableWritePrepare { + meta: PhysicalPlanMeta::new("TableWritePrepare"), + input, + table_info: table.get_table_info().clone(), + insert_schema, + select_schema, + select_column_bindings, + cast_needed, + }) +} + /// Build the INSERT SELECT physical plan (including Paimon route/exchange when needed). pub fn build_insert_select_physical_plan( select_plan: PhysicalPlan, @@ -131,6 +155,7 @@ pub fn build_insert_select_physical_plan( insert_schema: DataSchemaRef, table: Arc, cast_needed: bool, + input_prepared: bool, table_meta_timestamps: TableMetaTimestamps, distributed: bool, ) -> Result { @@ -138,20 +163,21 @@ pub fn build_insert_select_physical_plan( let needs_pk_route = is_fixed_bucket_primary_key(&table)?; let prepare_and_route = |input: PhysicalPlan| -> Result { - let prepared = PhysicalPlan::new(PaimonWritePrepare { - meta: PhysicalPlanMeta::new("PaimonWritePrepare"), + let prepared = prepare_table_write_input( input, - table_info: table_info.clone(), - insert_schema: insert_schema.clone(), - select_schema: select_schema.clone(), - select_column_bindings: select_column_bindings.clone(), + &table, + select_schema.clone(), + select_column_bindings.clone(), + insert_schema.clone(), cast_needed, - }); + ); wrap_paimon_write_distribution(prepared, table.clone()) }; + let input_prepared = input_prepared || needs_pk_route; if table.support_distributed_insert() && let Some(exchange) = Exchange::from_physical_plan(&select_plan) + && exchange.kind == FragmentKind::Merge { let input = if needs_pk_route { prepare_and_route(exchange.input.clone())? @@ -165,7 +191,7 @@ pub fn build_insert_select_physical_plan( select_column_bindings, insert_schema, cast_needed, - input_prepared: needs_pk_route, + input_prepared, table_meta_timestamps, meta: PhysicalPlanMeta::new("DistributedInsertSelect"), }); @@ -190,12 +216,12 @@ pub fn build_insert_select_physical_plan( select_column_bindings, insert_schema, cast_needed, - input_prepared: needs_pk_route, + input_prepared, table_meta_timestamps, meta: PhysicalPlanMeta::new("DistributedInsertSelect"), }); - // VALUES / local select has no top Merge; synthesize one for PK so commit metas gather. - if needs_pk_route && distributed { + // VALUES / local select has no top Merge; synthesize one so commit metas gather. + if input_prepared && distributed { Ok(PhysicalPlan::new(Exchange { input: insert, kind: FragmentKind::Merge, @@ -276,6 +302,73 @@ impl InsertInterpreter { let cast_needed = select_schema.as_ref() != &DataSchema::from(output_schema.as_ref()); Ok(cast_needed) } + + fn build_hash_distributed_input( + &self, + input: PhysicalPlan, + table: &Arc, + select_schema: &DataSchemaRef, + insert_schema: &DataSchemaRef, + select_column_bindings: &[ColumnBinding], + metadata: &MetadataRef, + cast_needed: bool, + ) -> Result<(PhysicalPlan, bool)> { + if table.engine() != "FUSE" { + return Ok((input, false)); + } + let fuse_table = FuseTable::try_from_table(table.as_ref())?; + if !fuse_table.use_hash_write_distribution() { + return Ok((input, false)); + } + + let Some(partition_info) = fuse_table.partition_pruning_info(self.ctx.clone()) else { + return Ok((input, false)); + }; + let input = prepare_table_write_input( + input, + table, + select_schema.clone(), + select_column_bindings.to_vec(), + insert_schema.clone(), + cast_needed, + ); + let input_schema = input.output_schema()?; + let mut partition_exprs = Vec::with_capacity(partition_info.partition_keys.len()); + for partition_key in partition_info.partition_keys { + let expr = partition_key.as_expr(&BUILTIN_FUNCTIONS); + partition_exprs.push(expr.project_column_ref(|name| input_schema.index_of(name))?); + } + + let input_columns = input_schema.num_fields(); + let mut eval_exprs = Vec::with_capacity(partition_exprs.len()); + let mut hash_keys = Vec::with_capacity(partition_exprs.len()); + for (index, expr) in partition_exprs.into_iter().enumerate() { + let data_type = expr.data_type().clone(); + let display_name = format!("partition_key_{index}"); + let symbol = metadata + .write() + .add_derived_column(display_name.clone(), data_type.clone()); + eval_exprs.push((expr.as_remote_expr(), symbol)); + hash_keys.push(RemoteExpr::ColumnRef { + span: None, + id: input_columns + index, + data_type, + display_name: display_name.clone(), + }); + } + + let projections: BTreeSet<_> = (0..input_columns + eval_exprs.len()).collect(); + let eval = PhysicalPlan::new(EvalScalar::create(input, eval_exprs, projections, None)); + let exchange = PhysicalPlan::new(Exchange { + meta: PhysicalPlanMeta::new("Exchange"), + input: eval, + kind: FragmentKind::GlobalShuffle, + keys: hash_keys, + ignore_exchange: false, + allow_adjust_parallelism: true, + }); + Ok((exchange, true)) + } } #[async_trait::async_trait] @@ -360,6 +453,7 @@ impl Interpreter for InsertInterpreter { insert_schema, table.clone(), false, + false, table_meta_timestamps, self.ctx.get_cluster().get_nodes().len() > 1, )?; @@ -452,15 +546,48 @@ impl Interpreter for InsertInterpreter { let update_stream_meta = dml_build_update_stream_req(self.ctx.clone()).await?; + let select_schema = plan.schema(); + let insert_schema = self.plan.dest_schema(); + let cast_needed = self.check_schema_cast(plan)?; + let distributed = self.ctx.get_cluster().get_nodes().len() > 1; + let (select_plan, input_prepared) = if distributed { + if table1.support_distributed_insert() + && let Some(exchange) = Exchange::from_physical_plan(&select_plan) + { + let (input, input_prepared) = self.build_hash_distributed_input( + exchange.input.clone(), + &table1, + &select_schema, + &insert_schema, + &select_column_bindings, + metadata, + cast_needed, + )?; + (exchange.derive(vec![input]), input_prepared) + } else { + self.build_hash_distributed_input( + select_plan, + &table1, + &select_schema, + &insert_schema, + &select_column_bindings, + metadata, + cast_needed, + )? + } + } else { + (select_plan, false) + }; let mut insert_select_plan = build_insert_select_physical_plan( select_plan, - plan.schema(), + select_schema, select_column_bindings, - self.plan.dest_schema(), + insert_schema, table1, - self.check_schema_cast(plan)?, + cast_needed, + input_prepared, table_meta_timestamps, - self.ctx.get_cluster().get_nodes().len() > 1, + distributed, )?; insert_select_plan.adjust_plan_id(&mut 0); diff --git a/src/query/service/src/interpreters/interpreter_table_set_options.rs b/src/query/service/src/interpreters/interpreter_table_set_options.rs index b69db9d66d3bf..2131237519cfc 100644 --- a/src/query/service/src/interpreters/interpreter_table_set_options.rs +++ b/src/query/service/src/interpreters/interpreter_table_set_options.rs @@ -53,6 +53,8 @@ use databend_storages_common_table_meta::table::OPT_KEY_SEGMENT_FORMAT; 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_TEMP_PREFIX; +use databend_storages_common_table_meta::table::OPT_KEY_WRITE_DISTRIBUTION_MODE; +use databend_storages_common_table_meta::table::WriteDistributionMode; use log::error; use crate::interpreters::Interpreter; @@ -179,6 +181,17 @@ impl Interpreter for SetOptionsInterpreter { .get_table(&self.ctx.get_tenant(), database, table_name) .await?; + if let Some(mode) = self.plan.set_options.get(OPT_KEY_WRITE_DISTRIBUTION_MODE) { + let mode = mode.parse::()?; + if mode == WriteDistributionMode::Hash + && !table.options().contains_key(OPT_KEY_PARTITION_BY) + { + return Err(ErrorCode::TableOptionInvalid(format!( + "{OPT_KEY_WRITE_DISTRIBUTION_MODE}='hash' requires PARTITION BY" + ))); + } + } + let engine = Engine::from(table.engine()); for table_option in self.plan.set_options.iter() { let key = table_option.0.to_lowercase(); diff --git a/src/query/service/src/physical_plans/mod.rs b/src/query/service/src/physical_plans/mod.rs index caafefcb67581..9d84d40b8b54f 100644 --- a/src/query/service/src/physical_plans/mod.rs +++ b/src/query/service/src/physical_plans/mod.rs @@ -43,7 +43,6 @@ mod physical_mutation_into_organize; mod physical_mutation_into_split; mod physical_mutation_manipulate; mod physical_mutation_source; -mod physical_paimon_write_prepare; mod physical_paimon_write_route; mod physical_project_set; mod physical_r_cte_scan; @@ -56,6 +55,7 @@ mod physical_row_fetch; mod physical_sort; mod physical_spatial_join; mod physical_table_scan; +mod physical_table_write_prepare; mod physical_udf; mod physical_union_all; mod physical_window; @@ -95,7 +95,6 @@ pub use physical_mutation_into_organize::MutationOrganize; pub use physical_mutation_into_split::MutationSplit; pub use physical_mutation_manipulate::MutationManipulate; pub use physical_mutation_source::*; -pub use physical_paimon_write_prepare::PaimonWritePrepare; pub use physical_paimon_write_route::PAIMON_ROUTE_KEY_NAME; pub use physical_paimon_write_route::PaimonWriteRoute; pub use physical_project_set::ProjectSet; @@ -108,8 +107,10 @@ pub use physical_replace_into::ReplaceInto; pub use physical_row_fetch::RowFetch; pub use physical_sequence::*; pub use physical_sort::Sort; +pub use physical_sort::SortStep; pub use physical_spatial_join::PhysicalSpatialJoin; pub use physical_table_scan::TableScan; +pub use physical_table_write_prepare::TableWritePrepare; pub use physical_udf::UdfFunctionDesc; pub use physical_union_all::UnionAll; pub use physical_window::*; diff --git a/src/query/service/src/physical_plans/physical_distributed_insert_select.rs b/src/query/service/src/physical_plans/physical_distributed_insert_select.rs index b9db273200ce3..2df3c4ca4019b 100644 --- a/src/query/service/src/physical_plans/physical_distributed_insert_select.rs +++ b/src/query/service/src/physical_plans/physical_distributed_insert_select.rs @@ -16,6 +16,7 @@ use std::any::Any; use databend_common_catalog::plan::DataSourcePlan; use databend_common_exception::Result; +use databend_common_expression::DataSchema; use databend_common_expression::DataSchemaRef; use databend_common_meta_app::schema::TableInfo; use databend_common_pipeline_transforms::TransformPipelineHelper; @@ -96,6 +97,9 @@ impl IPhysicalPlan for DistributedInsertSelect { let select_schema = &self.select_schema; let insert_schema = &self.insert_schema; + let table = builder + .ctx + .build_table_by_table_info(&self.table_info, None)?; // should render result for select if !self.input_prepared { PipelineBuilder::build_result_projection( @@ -106,7 +110,9 @@ impl IPhysicalPlan for DistributedInsertSelect { false, )?; } else { - let projection = (0..insert_schema.num_fields()).collect::>(); + let table_schema = table.schema().remove_virtual_computed_fields(); + let prepared_schema = DataSchema::from(&table_schema); + let projection = (0..prepared_schema.num_fields()).collect::>(); let num_input_columns = self.input.output_schema()?.num_fields(); builder.main_pipeline.add_transformer(|| { CompoundBlockOperator::new( @@ -129,10 +135,6 @@ impl IPhysicalPlan for DistributedInsertSelect { })?; } - let table = builder - .ctx - .build_table_by_table_info(&self.table_info, None)?; - let source_schema = insert_schema; if !self.input_prepared { PipelineBuilder::fill_and_reorder_columns( @@ -142,6 +144,11 @@ impl IPhysicalPlan for DistributedInsertSelect { source_schema.clone(), )?; } + PipelineBuilder::build_table_write_layout( + builder.ctx.clone(), + &mut builder.main_pipeline, + table.clone(), + )?; table.append_data( builder.ctx.clone(), diff --git a/src/query/service/src/physical_plans/physical_paimon_write_prepare.rs b/src/query/service/src/physical_plans/physical_table_write_prepare.rs similarity index 89% rename from src/query/service/src/physical_plans/physical_paimon_write_prepare.rs rename to src/query/service/src/physical_plans/physical_table_write_prepare.rs index 70b94b4543dcc..651b51655ab39 100644 --- a/src/query/service/src/physical_plans/physical_paimon_write_prepare.rs +++ b/src/query/service/src/physical_plans/physical_table_write_prepare.rs @@ -15,6 +15,7 @@ use std::any::Any; use databend_common_exception::Result; +use databend_common_expression::DataSchema; use databend_common_expression::DataSchemaRef; use databend_common_meta_app::schema::TableInfo; use databend_common_pipeline_transforms::TransformPipelineHelper; @@ -26,8 +27,10 @@ use crate::physical_plans::PhysicalPlan; use crate::physical_plans::PhysicalPlanMeta; use crate::pipelines::PipelineBuilder; +/// Normalize a table-write input to the destination table schema before +/// distribution-specific routing is applied. #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] -pub struct PaimonWritePrepare { +pub struct TableWritePrepare { pub meta: PhysicalPlanMeta, pub input: PhysicalPlan, pub table_info: TableInfo, @@ -38,7 +41,7 @@ pub struct PaimonWritePrepare { } #[typetag::serde] -impl IPhysicalPlan for PaimonWritePrepare { +impl IPhysicalPlan for TableWritePrepare { fn as_any(&self) -> &dyn Any { self } @@ -49,7 +52,8 @@ impl IPhysicalPlan for PaimonWritePrepare { &mut self.meta } fn output_schema(&self) -> Result { - Ok(self.insert_schema.clone()) + let table_schema = self.table_info.schema().remove_virtual_computed_fields(); + Ok(DataSchema::from(&table_schema).into()) } fn children<'a>(&'a self) -> Box + 'a> { Box::new(std::iter::once(&self.input)) diff --git a/src/query/service/src/pipelines/builders/builder_append_table.rs b/src/query/service/src/pipelines/builders/builder_append_table.rs index 60fd997f3f651..3c4df8ee0cfb1 100644 --- a/src/query/service/src/pipelines/builders/builder_append_table.rs +++ b/src/query/service/src/pipelines/builders/builder_append_table.rs @@ -16,17 +16,112 @@ use std::sync::Arc; use databend_common_catalog::table::Table; use databend_common_exception::Result; +use databend_common_expression::DataField; +use databend_common_expression::DataSchema; use databend_common_expression::DataSchemaRef; +use databend_common_expression::DataSchemaRefExt; +use databend_common_expression::SortColumnDescription; +use databend_common_functions::BUILTIN_FUNCTIONS; use databend_common_meta_app::schema::UpdateStreamMetaReq; use databend_common_meta_app::schema::UpsertTableCopiedFileReq; use databend_common_pipeline::core::Pipeline; +use databend_common_pipeline_transforms::TransformPipelineHelper; +use databend_common_pipeline_transforms::blocks::CompoundBlockOperator; +use databend_common_sql::evaluator::BlockOperator; +use databend_common_storages_fuse::FuseTable; use databend_storages_common_table_meta::meta::TableMetaTimestamps; use crate::pipelines::PipelineBuilder; +use crate::pipelines::builders::SortPipelineBuilder; use crate::sessions::QueryContext; +use crate::sessions::TableContextSettings; /// This file implements append to table pipeline builder. impl PipelineBuilder { + /// Coalesce hash-partitioned writes on the current node. Distributed inputs + /// must route the same partition keys to one node before reaching this stage. + pub fn build_table_write_layout( + ctx: Arc, + pipeline: &mut Pipeline, + table: Arc, + ) -> Result<()> { + if table.engine() != "FUSE" { + return Ok(()); + } + let fuse_table = FuseTable::try_from_table(table.as_ref())?; + if !fuse_table.use_hash_write_distribution() { + return Ok(()); + } + + let Some(partition_info) = fuse_table.partition_pruning_info(ctx.clone()) else { + return Ok(()); + }; + let table_schema = table.schema().remove_virtual_computed_fields(); + let input_schema = DataSchema::from(&table_schema); + let num_input_columns = input_schema.num_fields(); + let partition_exprs = partition_info + .partition_keys + .into_iter() + .map(|key| { + key.as_expr(&BUILTIN_FUNCTIONS) + .project_column_ref(|name| input_schema.index_of(name)) + }) + .collect::>>()?; + + let mut fields = input_schema.fields().clone(); + let mut sort_desc = Vec::with_capacity(partition_exprs.len()); + for (index, expr) in partition_exprs.iter().enumerate() { + fields.push(DataField::new( + &format!("partition_key_{index}"), + expr.data_type().clone(), + )); + sort_desc.push(SortColumnDescription { + offset: num_input_columns + index, + asc: true, + nulls_first: false, + }); + } + + let projections = (0..fields.len()).collect(); + let op = BlockOperator::Map { + exprs: partition_exprs, + projections: Some(projections), + }; + let func_ctx = ctx.get_function_context()?; + pipeline.add_transformer(|| { + CompoundBlockOperator::new(vec![op.clone()], func_ctx.clone(), num_input_columns) + }); + + let num_sort_columns = fields.len(); + let sort_schema = DataSchemaRefExt::create(fields); + let enable_fixed_rows = ctx.get_settings().get_enable_fixed_rows_sort()?; + let sort_builder = SortPipelineBuilder::create( + ctx.clone(), + sort_schema, + sort_desc.into(), + None, + enable_fixed_rows, + )? + .with_block_size_hit(table.get_block_thresholds().max_rows_per_block); + let max_threads = ctx.get_settings().get_max_threads()? as usize; + if pipeline.output_len() == 1 || max_threads == 1 { + pipeline.try_resize(max_threads)?; + } + sort_builder.build_full_sort_pipeline(pipeline, false)?; + + let projection = (0..num_input_columns).collect::>(); + pipeline.add_transformer(|| { + CompoundBlockOperator::new( + vec![BlockOperator::Project { + projection: projection.clone(), + }], + func_ctx.clone(), + num_sort_columns, + ) + }); + Ok(()) + } + #[allow(clippy::too_many_arguments)] pub fn build_append2table_with_commit_pipeline( ctx: Arc, @@ -40,6 +135,7 @@ impl PipelineBuilder { table_meta_timestamps: TableMetaTimestamps, ) -> Result<()> { Self::fill_and_reorder_columns(ctx.clone(), main_pipeline, table.clone(), source_schema)?; + Self::build_table_write_layout(ctx.clone(), main_pipeline, table.clone())?; table.append_data(ctx.clone(), main_pipeline, table_meta_timestamps)?; table.commit_insertion( @@ -64,6 +160,10 @@ impl PipelineBuilder { table_meta_timestamps: TableMetaTimestamps, ) -> Result<()> { Self::fill_and_reorder_columns(ctx.clone(), main_pipeline, table.clone(), source_schema)?; + // This layout is local to each writer. Distributed COPY does not yet route + // partition keys across nodes, so one partition may retain a tail block + // from each worker. + Self::build_table_write_layout(ctx.clone(), main_pipeline, table.clone())?; table.append_data(ctx, main_pipeline, table_meta_timestamps)?; diff --git a/src/query/service/tests/it/sql/exec/insert.rs b/src/query/service/tests/it/sql/exec/insert.rs index fc72b81b6c541..3542f8a27c99d 100644 --- a/src/query/service/tests/it/sql/exec/insert.rs +++ b/src/query/service/tests/it/sql/exec/insert.rs @@ -17,6 +17,7 @@ use std::collections::HashMap; use chrono::Duration; use databend_common_expression::DataField; use databend_common_expression::DataSchemaRefExt; +use databend_common_expression::RemoteExpr; use databend_common_expression::types::DataType; use databend_common_expression::types::NumberDataType; use databend_common_sql::ColumnBindingBuilder; @@ -27,11 +28,11 @@ use databend_query::interpreters::build_insert_select_physical_plan; use databend_query::physical_plans::ConstantTableScan; use databend_query::physical_plans::DistributedInsertSelect; use databend_query::physical_plans::Exchange; -use databend_query::physical_plans::PaimonWritePrepare; use databend_query::physical_plans::PaimonWriteRoute; use databend_query::physical_plans::PhysicalPlan; use databend_query::physical_plans::PhysicalPlanCast; use databend_query::physical_plans::PhysicalPlanMeta; +use databend_query::physical_plans::TableWritePrepare; use databend_storages_common_table_meta::meta::TableMetaTimestamps; use paimon::Catalog; use paimon::catalog::Identifier; @@ -123,6 +124,22 @@ fn wrap_with_merge_exchange(input: PhysicalPlan) -> PhysicalPlan { }) } +fn wrap_with_global_shuffle(input: PhysicalPlan) -> PhysicalPlan { + PhysicalPlan::new(Exchange { + input, + kind: FragmentKind::GlobalShuffle, + keys: vec![RemoteExpr::ColumnRef { + span: None, + id: 0, + data_type: DataType::Number(NumberDataType::Int32), + display_name: "partition_key".to_string(), + }], + allow_adjust_parallelism: true, + ignore_exchange: false, + meta: PhysicalPlanMeta::new("Exchange"), + }) +} + fn assert_pk_write_route_shape(plan: &PhysicalPlan) { let plan_text = format_plan(plan); assert!( @@ -157,7 +174,7 @@ fn assert_pk_write_route_shape(plan: &PhysicalPlan) { let route = PaimonWriteRoute::from_physical_plan(&shuffle.input) .expect("GlobalShuffle must wrap PaimonWriteRoute"); assert!( - PaimonWritePrepare::from_physical_plan(&route.input).is_some(), + TableWritePrepare::from_physical_plan(&route.input).is_some(), "route input must be cast/fill/reorder prepared" ); assert!( @@ -182,6 +199,7 @@ async fn test_paimon_write_route_plan() -> databend_common_exception::Result<()> pk_schema, pk_table, false, + false, TableMetaTimestamps::new(None, Duration::hours(1)), true, )?; @@ -199,6 +217,7 @@ async fn test_paimon_write_route_plan() -> databend_common_exception::Result<()> pk_schema, pk_table, false, + false, TableMetaTimestamps::new(None, Duration::hours(1)), false, )?; @@ -220,6 +239,7 @@ async fn test_paimon_write_route_plan() -> databend_common_exception::Result<()> append_schema, append_table, false, + false, TableMetaTimestamps::new(None, Duration::hours(1)), true, )?; @@ -260,6 +280,7 @@ async fn test_paimon_write_route_plan_preserves_select_merge() pk_schema, pk_table, false, + false, TableMetaTimestamps::new(None, Duration::hours(1)), true, )?; @@ -276,6 +297,7 @@ async fn test_paimon_write_route_plan_preserves_select_merge() append_schema, append_table, false, + false, TableMetaTimestamps::new(None, Duration::hours(1)), true, )?; @@ -298,3 +320,37 @@ async fn test_paimon_write_route_plan_preserves_select_merge() Ok(()) } + +#[tokio::test(flavor = "multi_thread")] +async fn test_prepared_global_shuffle_stays_below_insert() -> databend_common_exception::Result<()> +{ + let warehouse = TestWarehouse::new(); + let (_, append_id) = setup_tables(&warehouse.warehouse).await; + let table = databend_table(&warehouse.warehouse, &append_id).await; + let (select, bindings) = dummy_select_plan(2); + let schema = select.output_schema()?; + let shuffle = wrap_with_global_shuffle(select); + + let plan = build_insert_select_physical_plan( + shuffle, + schema.clone(), + bindings, + schema, + table, + false, + true, + TableMetaTimestamps::new(None, Duration::hours(1)), + true, + )?; + + let outer = Exchange::from_physical_plan(&plan) + .expect("distributed insert must synthesize a commit-gather Merge"); + assert_eq!(outer.kind, FragmentKind::Merge); + let insert = DistributedInsertSelect::from_physical_plan(&outer.input) + .expect("Merge must wrap DistributedInsertSelect"); + let shuffle = Exchange::from_physical_plan(&insert.input) + .expect("prepared GlobalShuffle must remain below the insert"); + assert_eq!(shuffle.kind, FragmentKind::GlobalShuffle); + + Ok(()) +} diff --git a/src/query/sql/src/planner/binder/ddl/table.rs b/src/query/sql/src/planner/binder/ddl/table.rs index 1c288e3b19d95..0fc5a296dcb31 100644 --- a/src/query/sql/src/planner/binder/ddl/table.rs +++ b/src/query/sql/src/planner/binder/ddl/table.rs @@ -35,7 +35,9 @@ use databend_common_ast::ast::DropTableStmt; use databend_common_ast::ast::Engine; use databend_common_ast::ast::ExistsTableStmt; use databend_common_ast::ast::Expr as AstExpr; +use databend_common_ast::ast::FunctionCall; use databend_common_ast::ast::Identifier; +use databend_common_ast::ast::Literal; use databend_common_ast::ast::ModifyColumnAction; use databend_common_ast::ast::OptimizeTableAction as AstOptimizeTableAction; use databend_common_ast::ast::OptimizeTableStmt; @@ -106,9 +108,13 @@ use databend_storages_common_table_meta::table::OPT_KEY_STORAGE_PREFIX; use databend_storages_common_table_meta::table::OPT_KEY_TABLE_ATTACHED_DATA_URI; use databend_storages_common_table_meta::table::OPT_KEY_TABLE_COMPRESSION; use databend_storages_common_table_meta::table::OPT_KEY_TEMP_PREFIX; +use databend_storages_common_table_meta::table::OPT_KEY_WRITE_DISTRIBUTION_MODE; use databend_storages_common_table_meta::table::TableCompression; +use databend_storages_common_table_meta::table::WriteDistributionMode; use databend_storages_common_table_meta::table::is_reserved_opt_key; +use derive_visitor::Drive; use derive_visitor::DriveMut; +use derive_visitor::Visitor; use log::debug; use opendal::Operator; use parking_lot::RwLock; @@ -178,6 +184,29 @@ use crate::plans::VacuumTemporaryFilesPlan; const FUSE_OPT_KEY_AGGRESSIVE_RECLUSTER: &str = "aggressive_recluster"; +#[derive(Visitor)] +#[visitor(FunctionCall(enter))] +struct PartitionBucketValidator { + invalid: bool, +} + +impl PartitionBucketValidator { + fn enter_function_call(&mut self, func: &FunctionCall) { + if func.name.name.eq_ignore_ascii_case("bucket") { + self.invalid |= !matches!( + func.args.as_slice(), + [ + AstExpr::Literal { + value: Literal::UInt64(buckets), + .. + }, + _ + ] if (1..=u32::MAX as u64).contains(buckets) + ); + } + } +} + pub(in crate::planner::binder) struct AnalyzeCreateTableResult { pub(in crate::planner::binder) schema: TableSchemaRef, pub(in crate::planner::binder) field_comments: Vec, @@ -966,6 +995,17 @@ impl Binder { ))); } + if matches!(engine, Engine::Fuse) { + if let Some(mode) = options.get(OPT_KEY_WRITE_DISTRIBUTION_MODE) { + let mode = mode.parse::()?; + if mode == WriteDistributionMode::Hash && partition_by.is_none() { + return Err(ErrorCode::TableOptionInvalid(format!( + "{OPT_KEY_WRITE_DISTRIBUTION_MODE}='hash' requires PARTITION BY" + ))); + } + } + } + if matches!(engine, Engine::Fuse) && let Some(partition_exprs) = partition_by { @@ -2422,6 +2462,17 @@ impl Binder { let mut cluster_keys = Vec::with_capacity(expr_len); let mut vector_cluster_key_num = 0; for cluster_expr in cluster_exprs.iter() { + if !allow_vector { + let mut validator = PartitionBucketValidator { invalid: false }; + cluster_expr.drive(&mut validator); + if validator.invalid { + return Err(ErrorCode::InvalidClusterKeys(format!( + "{key_name} expression `{cluster_expr:#}` must use bucket with a constant count between 1 and {}", + u32::MAX + ))); + } + } + let (cluster_key, _) = scalar_binder.bind(cluster_expr)?; if cluster_key.used_columns().len() != 1 || !cluster_key.evaluable() { return Err(ErrorCode::InvalidClusterKeys(format!( 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 index 284f1f33eafd3..cdd30d8eaa0b3 100644 --- a/src/query/storages/common/table_meta/src/table/table_keys.rs +++ b/src/query/storages/common/table_meta/src/table/table_keys.rs @@ -16,6 +16,7 @@ use std::collections::BTreeMap; use std::collections::HashSet; use std::fmt::Display; use std::fmt::Formatter; +use std::str::FromStr; use std::sync::LazyLock; use databend_common_exception::ErrorCode; @@ -77,10 +78,31 @@ pub const OPT_KEY_CLUSTER_TYPE: &str = "cluster_type"; /// The normalized Fuse partition expressions. This is set by CREATE TABLE /// PARTITION BY and is not a user-settable table option. pub const OPT_KEY_PARTITION_BY: &str = "partition_by"; +pub const OPT_KEY_WRITE_DISTRIBUTION_MODE: &str = "write_distribution_mode"; 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"; +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub enum WriteDistributionMode { + None, + Hash, +} + +impl FromStr for WriteDistributionMode { + type Err = ErrorCode; + + fn from_str(value: &str) -> Result { + match value.to_ascii_lowercase().as_str() { + "none" => Ok(Self::None), + "hash" => Ok(Self::Hash), + _ => Err(ErrorCode::TableOptionInvalid(format!( + "{OPT_KEY_WRITE_DISTRIBUTION_MODE} must be either 'none' or 'hash', got: {value}" + ))), + } + } +} + /// Table option keys that reserved for internal usage only /// - Users are not allowed to specify this option keys in DDL /// - Should not be shown in `show create table` statement diff --git a/src/query/storages/fuse/src/fuse_table.rs b/src/query/storages/fuse/src/fuse_table.rs index 3141f8c31af6b..464a8f2132ecd 100644 --- a/src/query/storages/fuse/src/fuse_table.rs +++ b/src/query/storages/fuse/src/fuse_table.rs @@ -114,6 +114,7 @@ use databend_storages_common_table_meta::table::OPT_KEY_SNAPSHOT_LOCATION_FIXED_ use databend_storages_common_table_meta::table::OPT_KEY_STORAGE_FORMAT; use databend_storages_common_table_meta::table::OPT_KEY_TABLE_ATTACHED_DATA_URI; use databend_storages_common_table_meta::table::OPT_KEY_TABLE_COMPRESSION; +use databend_storages_common_table_meta::table::OPT_KEY_WRITE_DISTRIBUTION_MODE; use databend_storages_common_table_meta::table::TableCompression; use databend_storages_common_table_meta::table::analyze_top_n_size_from_options; use futures_util::TryStreamExt; @@ -614,6 +615,15 @@ impl FuseTable { self.resolve_partition_keys().map_or(0, |keys| keys.len()) } + pub fn use_hash_write_distribution(&self) -> bool { + self.partition_key_count() != 0 + && self + .table_info + .options() + .get(OPT_KEY_WRITE_DISTRIBUTION_MODE) + .is_some_and(|mode| mode.eq_ignore_ascii_case("hash")) + } + /// The key used for physical Fuse layout. Partition expressions are always the /// leading dimensions, followed by the user-visible linear cluster key. pub fn resolve_physical_cluster_keys(&self) -> Option> { diff --git a/tests/sqllogictests/suites/base/09_fuse_engine/09_0054_partition_by.test b/tests/sqllogictests/suites/base/09_fuse_engine/09_0054_partition_by.test index 005c28091092e..619b28df57012 100644 --- a/tests/sqllogictests/suites/base/09_fuse_engine/09_0054_partition_by.test +++ b/tests/sqllogictests/suites/base/09_fuse_engine/09_0054_partition_by.test @@ -369,6 +369,94 @@ SELECT block_count FROM fuse_snapshot('db_09_0054', 'fragmented_input') LIMIT 1 ---- 6 +# The common write layout prepares the INT table row from a BIGINT input and +# full-sorts by the physical partition transform before Fuse serialization. +statement ok +SET max_threads=4 + +statement ok +SET max_block_size=1000 + +statement ok +CREATE TABLE hash_partitioned(id INT) PARTITION BY (bucket(16, id)) WRITE_DISTRIBUTION_MODE='hash' ROW_PER_BLOCK=1000 BLOCK_PER_SEGMENT=1000 + +statement ok +CREATE TABLE hash_partitioned_none(id INT) PARTITION BY (bucket(16, id)) ROW_PER_BLOCK=1000 BLOCK_PER_SEGMENT=1000 + +statement ok +INSERT INTO hash_partitioned SELECT number::BIGINT FROM numbers(9600) + +statement ok +INSERT INTO hash_partitioned_none SELECT number::BIGINT FROM numbers(9600) + +query I +SELECT count(*) FROM hash_partitioned +---- +9600 + +# The base suite also runs against a three-node cluster. Each bucket has about +# 600 rows, below ROW_PER_BLOCK. GlobalShuffle routes every bucket to one writer +# before the node-local full sort, keeping the total below 32 after allowing +# sort block-boundary tails. Without the shuffle, all three writers retain tails +# for all 16 buckets. The none table also verifies the block-local baseline. +query II +SELECT to_int64(h.block_count <= 32), to_int64(h.block_count < n.block_count) +FROM fuse_snapshot('db_09_0054', 'hash_partitioned') AS h +CROSS JOIN fuse_snapshot('db_09_0054', 'hash_partitioned_none') AS n +LIMIT 1 +---- +1 1 + +# VALUES reaches the same common write layout without a dedicated physical path. +statement ok +INSERT INTO hash_partitioned VALUES (40000), (40001) + +query I +SELECT count(*) FROM hash_partitioned WHERE id >= 40000 +---- +2 + +# In a cluster, a constant SELECT has no original top Merge. Its synthesized +# hash shuffle must remain below the distributed writer. +statement ok +INSERT INTO hash_partitioned SELECT 40002::INT + +query I +SELECT count(*) FROM hash_partitioned WHERE id >= 40000 +---- +3 + +statement error write_distribution_mode must be either 'none' or 'hash' +CREATE TABLE invalid_write_distribution(p INT) PARTITION BY (p) WRITE_DISTRIBUTION_MODE='random' + +statement error write_distribution_mode='hash' requires PARTITION BY +CREATE TABLE missing_write_partition(p INT) WRITE_DISTRIBUTION_MODE='hash' + +# bucket remains a valid cluster expression without physical partitioning. +statement ok +CREATE TABLE bucket_cluster_only(id INT) CLUSTER BY (bucket(16, id)) + +statement ok +INSERT INTO bucket_cluster_only SELECT number FROM numbers(100) + +query I +SELECT count(*) FROM bucket_cluster_only +---- +100 + +# A bucket cluster key does not substitute for a PARTITION BY declaration. +statement error write_distribution_mode='hash' requires PARTITION BY +CREATE TABLE bucket_cluster_hash(id INT) CLUSTER BY (bucket(16, id)) WRITE_DISTRIBUTION_MODE='hash' + +statement error must use bucket with a constant count between 1 and 4294967295 +CREATE TABLE invalid_bucket_count(id BIGINT) PARTITION BY (bucket(id, id)) + +statement error must use bucket with a constant count between 1 and 4294967295 +CREATE TABLE zero_bucket_count(id BIGINT) PARTITION BY (bucket(0, id)) + +statement error must use bucket with a constant count between 1 and 4294967295 +CREATE TABLE excessive_bucket_count(id BIGINT) PARTITION BY (bucket(4294967296, id)) + # Like CLUSTER BY, each physical key expression may reference only one source # column. Composite partitioning is expressed as PARTITION BY (a, b). statement error PARTITION BY expression `a \+ b` is invalid