diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index fce04169..ec23ad69 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -783,10 +783,20 @@ impl<'a> CoreOptions<'a> { } /// Explicit bucket key columns. If not set, defaults to primary keys for PK tables. + /// + /// Blank entries are dropped and an all-blank option resolves to `None`, + /// mirroring Java `TableSchema#originalBucketKeys`, so callers fall back to + /// the primary keys instead of treating `""` as a column name. pub fn bucket_key(&self) -> Option> { - self.options - .get(BUCKET_KEY_OPTION) - .map(|v| v.split(',').map(|s| s.trim().to_string()).collect()) + let keys: Vec = self + .options + .get(BUCKET_KEY_OPTION)? + .split(',') + .map(str::trim) + .filter(|key| !key.is_empty()) + .map(str::to_string) + .collect(); + (!keys.is_empty()).then_some(keys) } pub fn commit_max_retries(&self) -> u32 { diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 0245cf41..f96bc0d7 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -489,6 +489,12 @@ impl TableSchema { &new_schema.primary_keys, &new_schema.fields, )?; + Schema::validate_bucket_keys( + &new_schema.options, + &new_schema.fields, + &new_schema.partition_keys, + &new_schema.primary_keys, + )?; Schema::validate_read_batch_size(&new_schema.options)?; Ok(new_schema) } @@ -936,6 +942,7 @@ impl Schema { AggregationConfig::new(&options).validate_create_mode(&primary_keys, &fields)?; Self::validate_first_row_changelog_producer(&options)?; Self::validate_rowkind_field(&options, &primary_keys, &fields)?; + Self::validate_bucket_keys(&options, &fields, &partition_keys, &primary_keys)?; Self::validate_read_batch_size(&options)?; Ok(Self { @@ -1351,6 +1358,57 @@ impl Schema { Ok(()) } + /// Validate the explicit `bucket-key` option against the schema, mirroring + /// Java `TableSchema#originalBucketKeys`. A bucket key that is missing, + /// partitioned, or outside the primary key otherwise resolves to no field + /// index in `TableWrite`, which silently degrades to a constant bucket 0 + /// assigner instead of hashing. + fn validate_bucket_keys( + options: &HashMap, + fields: &[DataField], + partition_keys: &[String], + primary_keys: &[String], + ) -> crate::Result<()> { + let Some(bucket_keys) = CoreOptions::new(options).bucket_key() else { + return Ok(()); + }; + + let mut seen: HashSet<&str> = HashSet::new(); + for key in &bucket_keys { + if fields.iter().all(|f| f.name() != key) { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Field names should contain all bucket keys, but bucket key '{key}' \ + can not be found in table schema." + ), + }); + } + if !seen.insert(key.as_str()) { + return Err(crate::Error::ConfigInvalid { + message: format!("Bucket key '{key}' is defined repeatedly."), + }); + } + if partition_keys.contains(key) { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Bucket keys should not be in partition keys, but bucket key '{key}' \ + is a partition field." + ), + }); + } + if !primary_keys.is_empty() && !primary_keys.contains(key) { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Primary keys {primary_keys:?} should contain all bucket keys, but \ + bucket key '{key}' is not a primary key field." + ), + }); + } + } + + Ok(()) + } + fn options_error_to_config_invalid(error: crate::Error) -> crate::Error { match error { crate::Error::Unsupported { message } => crate::Error::ConfigInvalid { message }, @@ -2923,6 +2981,140 @@ mod tests { ); } + #[test] + fn test_create_schema_rejects_unknown_bucket_key() { + let err = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("name", DataType::VarChar(VarCharType::string_type())) + .option("bucket", "4") + // typo: `nmae` instead of `name` + .option("bucket-key", "nmae") + .build() + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { ref message } + if message.contains("nmae") && message.contains("can not be found")), + "bucket key missing from the schema should be rejected, got {err:?}" + ); + } + + #[test] + fn test_create_schema_rejects_repeated_bucket_key() { + let err = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("name", DataType::VarChar(VarCharType::string_type())) + .option("bucket", "4") + .option("bucket-key", "name,name") + .build() + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { ref message } + if message.contains("name") && message.contains("repeatedly")), + "repeated bucket key should be rejected, got {err:?}" + ); + } + + #[test] + fn test_create_schema_rejects_partitioned_bucket_key() { + let err = Schema::builder() + .column("pt", DataType::Int(IntType::new())) + .column("id", DataType::Int(IntType::new())) + .partition_keys(["pt"]) + .option("bucket", "4") + .option("bucket-key", "pt") + .build() + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { ref message } + if message.contains("pt") && message.contains("partition")), + "partition field used as bucket key should be rejected, got {err:?}" + ); + } + + #[test] + fn test_create_schema_rejects_bucket_key_outside_primary_key() { + let err = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("name", DataType::VarChar(VarCharType::string_type())) + .primary_key(["id"]) + .option("bucket", "4") + .option("bucket-key", "name") + .build() + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { ref message } + if message.contains("name") && message.contains("primary key")), + "non-primary-key bucket key on a PK table should be rejected, got {err:?}" + ); + } + + #[test] + fn test_create_schema_accepts_bucket_key_subset_of_primary_key() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("name", DataType::VarChar(VarCharType::string_type())) + .column("v", DataType::Int(IntType::new())) + .primary_key(["id", "name"]) + .option("bucket", "4") + .option("bucket-key", "id") + .build(); + + assert!( + schema.is_ok(), + "a bucket key that is a primary key field should be accepted, got {schema:?}" + ); + } + + #[test] + fn test_blank_bucket_key_falls_back_to_primary_keys() { + // A blank option must not resolve to a `""` column: `TableWrite` would + // find no field index for it and silently write every row to bucket 0. + let table_schema = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("name", DataType::VarChar(VarCharType::string_type())) + .primary_key(["id"]) + .option("bucket", "4") + .option("bucket-key", " ") + .build() + .unwrap(), + ); + + assert_eq!(table_schema.bucket_keys(), vec!["id".to_string()]); + } + + #[test] + fn test_alter_set_unknown_bucket_key_rejected() { + let table_schema = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("name", DataType::VarChar(VarCharType::string_type())) + .option("bucket", "4") + .option("bucket-key", "name") + .build() + .unwrap(), + ); + + let err = table_schema + .apply_changes(vec![crate::spec::SchemaChange::set_option( + BUCKET_KEY_OPTION.to_string(), + "nmae".to_string(), + )]) + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { ref message } + if message.contains("nmae") && message.contains("can not be found")), + "alter setting an unknown bucket key should be rejected, got {err:?}" + ); + } + #[test] fn test_drop_column_referenced_by_bucket_key_rejected() { let table_schema = TableSchema::new(