From c38789538fc0336fe8e90e1422d302e19ffbfb92 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Tue, 28 Jul 2026 20:16:18 +0800 Subject: [PATCH] feat(c): build a catalog-free Table from a resolved schema JSON Add a catalog-free way to build a Table from an already-resolved Paimon TableSchema, so a non-Java engine (e.g. Doris via the C FFI) can rebuild a table for scan/read without a catalog lookup. Today the only way to obtain a Table is Catalog::get_table, which re-resolves the table (risking schema/ snapshot drift versus what the planner already resolved) and ties the reader to a filesystem-catalog layout. The caller passes everything a reader needs directly: table path, the full TableSchema JSON (preserved as-is, not re-loaded or normalized), database/ table name, branch, and filesystem options. The Rust side constructs the Table with a plain FileIO and rest_env = None; branch only selects the branch-scoped schema/snapshot/tag managers. Reads can still use FE-generated splits via paimon_plan_from_split_bytes. Mirroring Java FileStoreTableFactory.create(fileIO, path, schema), the table stays a normal (writable) table; existing non-main-branch / time-travel write guards apply. Validate the externally supplied schema's structural invariants without normalizing it (TableSchema::validate_resolved_structure): reject duplicate field names, duplicate field ids (including nested), primary-key/partition columns that do not exist, primary keys that are exactly the partition columns (the read path would panic on a zero-column key), and reserved system field names/ids (e.g. a user _ROW_ID column would be silently replaced by the system row number on read). Create-time policy checks (merge-engine/changelog/ aggregation/rowkind/blob strategy, bucket-key existence) are intentionally not run, since they normalize the schema or reject shapes that are valid to read. - crates/paimon/src/table/mod.rs: Table::from_resolved_schema. - crates/paimon/src/spec/schema.rs: TableSchema::validate_resolved_structure. - bindings/c/src/table.rs: paimon_table_from_schema_json (nullable branch defaults to main), mirroring the paimon_catalog_get_table handle/free contract with an ABI signature guard; freed via paimon_table_free. Known pre-existing limitations, unreachable via the split-bytes read path Doris uses, are left for follow-ups: incremental scan uses a root SnapshotManager for non-main branches; time-travel selectors embedded in schema JSON options are not honored by the plain read builder; a stale path option is preferred by FormatTableScan over the table location; a non-existent bucket-key is silently dropped on write (write path only); full Java-parity SchemaValidation is not mirrored. --- Cargo.lock | 1 + bindings/c/Cargo.toml | 1 + bindings/c/src/table.rs | 181 +++++++++++++++++- bindings/c/src/tests.rs | 308 +++++++++++++++++++++++++++++++ bindings/c/src/vector_search.rs | 3 +- bindings/c/src/write.rs | 3 +- crates/paimon/src/spec/schema.rs | 280 ++++++++++++++++++++++++++++ crates/paimon/src/table/mod.rs | 40 ++++ 8 files changed, 809 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7ee0257e2..47eb4598a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4595,6 +4595,7 @@ dependencies = [ "futures", "paimon", "paimon-vindex-core", + "serde_json", "tokio", ] diff --git a/bindings/c/Cargo.toml b/bindings/c/Cargo.toml index f02c3e83f..2340f9705 100644 --- a/bindings/c/Cargo.toml +++ b/bindings/c/Cargo.toml @@ -37,6 +37,7 @@ futures = "0.3" arrow = { workspace = true } arrow-array = { workspace = true } arrow-schema = { workspace = true } +serde_json = "1.0.120" [dev-dependencies] # Test-only: the vector-search integration tests build a real primary-key vindex diff --git a/bindings/c/src/table.rs b/bindings/c/src/table.rs index 05f82e319..0ccf6d349 100644 --- a/bindings/c/src/table.rs +++ b/bindings/c/src/table.rs @@ -16,19 +16,22 @@ // under the License. use std::collections::HashMap; -use std::ffi::c_void; +use std::ffi::{c_char, c_void}; use arrow_array::ffi::{FFI_ArrowArray, FFI_ArrowSchema}; use arrow_array::{Array, StructArray}; use futures::StreamExt; -use paimon::spec::{DataField, DataType, Datum, Predicate, PredicateBuilder}; +use paimon::catalog::{Identifier, DEFAULT_MAIN_BRANCH}; +use paimon::io::FileIO; +use paimon::spec::{DataField, DataType, Datum, Predicate, PredicateBuilder, TableSchema}; use paimon::table::{ArrowRecordBatchStream, DataSplit, Table}; use paimon::Plan; use crate::error::{check_non_null, paimon_error, validate_cstr, PaimonErrorCode}; use crate::result::{ - paimon_result_new_read, paimon_result_next_batch, paimon_result_plan, paimon_result_predicate, - paimon_result_read_builder, paimon_result_record_batch_reader, paimon_result_table_scan, + paimon_result_get_table, paimon_result_new_read, paimon_result_next_batch, paimon_result_plan, + paimon_result_predicate, paimon_result_read_builder, paimon_result_record_batch_reader, + paimon_result_table_scan, }; use crate::runtime; use crate::types::*; @@ -58,10 +61,166 @@ unsafe fn box_table_read_state(state: TableReadState) -> *mut paimon_table_read // ======================= Table =============================== +/// Create a table directly from a resolved Paimon table schema JSON. +/// +/// This constructor does not create a catalog or derive a warehouse. Storage +/// options are used only to build FileIO; they are not merged into the supplied +/// table schema. `branch` selects the branch-scoped managers while preserving +/// the supplied schema; pass null to default to the `main` branch. +/// +/// # Safety +/// All string pointers except `branch` must be valid null-terminated C strings. +/// `branch` may be null to select the default `main` branch, or a valid +/// null-terminated C string. `storage_options` must point to +/// `storage_options_len` valid `paimon_option` values, or be null when +/// `storage_options_len` is 0. +#[no_mangle] +pub unsafe extern "C" fn paimon_table_from_schema_json( + table_path: *const c_char, + table_schema_json: *const c_char, + database: *const c_char, + table_name: *const c_char, + branch: *const c_char, + storage_options: *const paimon_option, + storage_options_len: usize, +) -> paimon_result_get_table { + let table_path = match validate_cstr(table_path, "table_path") { + Ok(value) => value, + Err(error) => { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error, + } + } + }; + let table_schema_json = match validate_cstr(table_schema_json, "table_schema_json") { + Ok(value) => value, + Err(error) => { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error, + } + } + }; + let database = match validate_cstr(database, "database") { + Ok(value) => value, + Err(error) => { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error, + } + } + }; + let table_name = match validate_cstr(table_name, "table_name") { + Ok(value) => value, + Err(error) => { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error, + } + } + }; + let branch = if branch.is_null() { + DEFAULT_MAIN_BRANCH.to_string() + } else { + match validate_cstr(branch, "branch") { + Ok(value) => value, + Err(error) => { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error, + } + } + } + }; + if storage_options.is_null() && storage_options_len > 0 { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error: paimon_error::new( + PaimonErrorCode::InvalidInput, + "null storage_options pointer with non-zero length".to_string(), + ), + }; + } + + let schema = match serde_json::from_str::(&table_schema_json) { + Ok(schema) => schema, + Err(error) => { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error: paimon_error::new( + PaimonErrorCode::InvalidInput, + format!("Failed to parse table schema JSON: {error}"), + ), + } + } + }; + + let mut options = HashMap::with_capacity(storage_options_len); + if storage_options_len > 0 { + for option in std::slice::from_raw_parts(storage_options, storage_options_len) { + let key = match validate_cstr(option.key, "storage option key") { + Ok(value) => value, + Err(error) => { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error, + } + } + }; + let value = match validate_cstr(option.value, "storage option value") { + Ok(value) => value, + Err(error) => { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error, + } + } + }; + options.insert(key, value); + } + } + + let file_io = match FileIO::from_path(&table_path) + .and_then(|builder| builder.with_props(options.iter()).build()) + { + Ok(file_io) => file_io, + Err(error) => { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error: paimon_error::from_paimon(error), + } + } + }; + let table = match Table::from_resolved_schema( + file_io, + Identifier::new(database, table_name), + table_path, + schema, + branch, + ) { + Ok(table) => table, + Err(error) => { + return paimon_result_get_table { + table: std::ptr::null_mut(), + error: paimon_error::from_paimon(error), + } + } + }; + let wrapper = Box::new(paimon_table { + inner: Box::into_raw(Box::new(table)) as *mut c_void, + }); + paimon_result_get_table { + table: Box::into_raw(wrapper), + error: std::ptr::null_mut(), + } +} + /// Free a paimon_table. /// /// # Safety -/// Only call with a table returned from `paimon_catalog_get_table`. +/// Only call with a table returned from `paimon_catalog_get_table` or +/// `paimon_table_from_schema_json`. #[no_mangle] pub unsafe extern "C" fn paimon_table_free(table: *mut paimon_table) { free_table_wrapper(table, |t| t.inner); @@ -130,7 +289,8 @@ unsafe fn new_read_builder_state( /// Create a new ReadBuilder from a Table. /// /// # Safety -/// `table` must be a valid pointer from `paimon_catalog_get_table`, or null (returns error). +/// `table` must be a valid pointer from `paimon_catalog_get_table` or +/// `paimon_table_from_schema_json`, or null (returns error). #[no_mangle] pub unsafe extern "C" fn paimon_table_new_read_builder( table: *const paimon_table, @@ -1773,6 +1933,15 @@ const _: unsafe extern "C" fn( // constructors so an accidental signature change fails to compile rather than // silently breaking header consumers. To add behavior, introduce a new // `paimon_table_new_read_builder_*` symbol instead of changing one of these. +const _: unsafe extern "C" fn( + *const c_char, + *const c_char, + *const c_char, + *const c_char, + *const c_char, + *const paimon_option, + usize, +) -> paimon_result_get_table = paimon_table_from_schema_json; const _: unsafe extern "C" fn(*const paimon_table) -> paimon_result_read_builder = paimon_table_new_read_builder; const _: unsafe extern "C" fn( diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs index 9c35fb987..d29c7eb85 100644 --- a/bindings/c/src/tests.rs +++ b/bindings/c/src/tests.rs @@ -85,6 +85,10 @@ unsafe fn unwrap_table(table: *mut paimon_table) { } } +unsafe fn table_ref<'a>(table: *const paimon_table) -> &'a Table { + &*((*table).inner as *const Table) +} + fn make_batch(ids: Vec, names: Vec<&str>) -> RecordBatch { let schema = Arc::new(ArrowSchema::new(vec![ ArrowField::new("id", ArrowDataType::Int32, false), @@ -293,6 +297,310 @@ unsafe fn read_rows_ffi(table: *const paimon_table) -> Vec<(i32, String)> { rows } +// ========================================================================= +// Catalog-free table construction tests +// ========================================================================= + +#[test] +fn test_table_from_schema_json_preserves_resolved_schema_and_branch() { + let path = CString::new("memory:/test_resolved_branch").unwrap(); + let database = CString::new("default").unwrap(); + let table_name = CString::new("test").unwrap(); + let branch = CString::new("dev").unwrap(); + let schema = simple_table_schema(); + let schema_json = CString::new(serde_json::to_string(&schema).unwrap()).unwrap(); + let option_key = CString::new("storage-only-option").unwrap(); + let option_value = CString::new("secret").unwrap(); + let storage_options = [paimon_option { + key: option_key.as_ptr(), + value: option_value.as_ptr(), + }]; + + unsafe { + let result = paimon_table_from_schema_json( + path.as_ptr(), + schema_json.as_ptr(), + database.as_ptr(), + table_name.as_ptr(), + branch.as_ptr(), + storage_options.as_ptr(), + storage_options.len(), + ); + + assert!(result.error.is_null()); + assert!(!result.table.is_null()); + let table = table_ref(result.table); + assert_eq!(table.location(), "memory:/test_resolved_branch"); + assert_eq!(table.identifier(), &Identifier::new("default", "test")); + // The resolved schema is preserved as-is (no normalization or mutation). + assert_eq!(table.schema(), &schema); + assert_eq!(table.branch(), "dev"); + assert!(table.is_branch_reference()); + assert_eq!( + table.schema_manager().schema_path(schema.id()), + "memory:/test_resolved_branch/branch/branch-dev/schema/schema-0" + ); + assert_eq!( + table.snapshot_manager().snapshot_path(1), + "memory:/test_resolved_branch/branch/branch-dev/snapshot/snapshot-1" + ); + assert_eq!( + table.tag_manager().tag_path("release"), + "memory:/test_resolved_branch/branch/branch-dev/tag/tag-release" + ); + assert!(!table.schema().options().contains_key("storage-only-option")); + + let read_builder = paimon_table_new_read_builder(result.table); + assert!(read_builder.error.is_null()); + assert!(!read_builder.read_builder.is_null()); + paimon_read_builder_free(read_builder.read_builder); + + paimon_table_free(result.table); + } +} + +#[test] +fn test_table_from_schema_json_rejects_missing_primary_key_column() { + // A syntactically valid schema whose primaryKeys reference a non-existent + // column must be rejected at construction, not panic later (e.g. the write + // path unwraps a PartitionComputer built from missing columns). + let path = CString::new("memory:/test_resolved_bad_pk").unwrap(); + let database = CString::new("default").unwrap(); + let table_name = CString::new("test").unwrap(); + let branch = CString::new("main").unwrap(); + + // simple_table_schema has columns id, name; declare a PK on a missing column. + let base = simple_table_schema(); + let bad_json = serde_json::to_value(&base).unwrap(); + let mut bad_json = bad_json; + bad_json["primaryKeys"] = serde_json::json!(["missing"]); + let schema_json = CString::new(serde_json::to_string(&bad_json).unwrap()).unwrap(); + + unsafe { + let result = paimon_table_from_schema_json( + path.as_ptr(), + schema_json.as_ptr(), + database.as_ptr(), + table_name.as_ptr(), + branch.as_ptr(), + std::ptr::null(), + 0, + ); + assert!(result.table.is_null()); + assert!(!result.error.is_null()); + assert_eq!((*result.error).code, PaimonErrorCode::InvalidInput as i32); + paimon_error_free(result.error); + } +} + +#[test] +fn test_table_from_schema_json_rejects_reserved_field_name() { + // A user column colliding with a system field name (_ROW_ID) would be + // silently replaced by the system row number on read; reject at construction. + let path = CString::new("memory:/test_resolved_reserved_field").unwrap(); + let database = CString::new("default").unwrap(); + let table_name = CString::new("test").unwrap(); + let branch = CString::new("main").unwrap(); + + let base = simple_table_schema(); + let mut bad_json = serde_json::to_value(&base).unwrap(); + // Rename the second field ("name") to the reserved system name "_ROW_ID". + bad_json["fields"][1]["name"] = serde_json::json!("_ROW_ID"); + let schema_json = CString::new(serde_json::to_string(&bad_json).unwrap()).unwrap(); + + unsafe { + let result = paimon_table_from_schema_json( + path.as_ptr(), + schema_json.as_ptr(), + database.as_ptr(), + table_name.as_ptr(), + branch.as_ptr(), + std::ptr::null(), + 0, + ); + assert!(result.table.is_null()); + assert!(!result.error.is_null()); + assert_eq!((*result.error).code, PaimonErrorCode::InvalidInput as i32); + paimon_error_free(result.error); + } +} + +#[test] +fn test_table_from_schema_json_main_branch_uses_main_schema_directory() { + let path = CString::new("memory:/test_resolved_main").unwrap(); + let database = CString::new("default").unwrap(); + let table_name = CString::new("test").unwrap(); + let branch = CString::new("main").unwrap(); + let schema = simple_table_schema(); + let schema_json = CString::new(serde_json::to_string(&schema).unwrap()).unwrap(); + + unsafe { + let result = paimon_table_from_schema_json( + path.as_ptr(), + schema_json.as_ptr(), + database.as_ptr(), + table_name.as_ptr(), + branch.as_ptr(), + std::ptr::null(), + 0, + ); + + assert!(result.error.is_null()); + assert!(!result.table.is_null()); + let table = table_ref(result.table); + assert_eq!(table.branch(), "main"); + assert!(!table.is_branch_reference()); + assert_eq!( + table.schema_manager().schema_path(schema.id()), + "memory:/test_resolved_main/schema/schema-0" + ); + + paimon_table_free(result.table); + } +} + +#[test] +fn test_table_from_schema_json_null_branch_defaults_to_main() { + let path = CString::new("memory:/test_resolved_null_branch").unwrap(); + let database = CString::new("default").unwrap(); + let table_name = CString::new("test").unwrap(); + let schema = simple_table_schema(); + let schema_json = CString::new(serde_json::to_string(&schema).unwrap()).unwrap(); + + unsafe { + let result = paimon_table_from_schema_json( + path.as_ptr(), + schema_json.as_ptr(), + database.as_ptr(), + table_name.as_ptr(), + std::ptr::null(), + std::ptr::null(), + 0, + ); + + assert!(result.error.is_null()); + assert!(!result.table.is_null()); + let table = table_ref(result.table); + assert_eq!(table.branch(), "main"); + assert!(!table.is_branch_reference()); + assert_eq!( + table.schema_manager().schema_path(schema.id()), + "memory:/test_resolved_null_branch/schema/schema-0" + ); + + paimon_table_free(result.table); + } +} + +#[test] +fn test_table_from_schema_json_rejects_invalid_input() { + let path = CString::new("memory:/test_resolved_invalid").unwrap(); + let malformed_schema = CString::new("not-json").unwrap(); + let database = CString::new("default").unwrap(); + let table_name = CString::new("test").unwrap(); + let branch = CString::new("main").unwrap(); + + unsafe { + let malformed = paimon_table_from_schema_json( + path.as_ptr(), + malformed_schema.as_ptr(), + database.as_ptr(), + table_name.as_ptr(), + branch.as_ptr(), + std::ptr::null(), + 0, + ); + assert!(malformed.table.is_null()); + assert!(!malformed.error.is_null()); + assert_eq!( + (*malformed.error).code, + PaimonErrorCode::InvalidInput as i32 + ); + paimon_error_free(malformed.error); + + let schema = simple_table_schema(); + let schema_json = CString::new(serde_json::to_string(&schema).unwrap()).unwrap(); + let null_path = paimon_table_from_schema_json( + std::ptr::null(), + schema_json.as_ptr(), + database.as_ptr(), + table_name.as_ptr(), + branch.as_ptr(), + std::ptr::null(), + 0, + ); + assert!(null_path.table.is_null()); + assert!(!null_path.error.is_null()); + assert_eq!( + (*null_path.error).code, + PaimonErrorCode::InvalidInput as i32 + ); + paimon_error_free(null_path.error); + + let null_options = paimon_table_from_schema_json( + path.as_ptr(), + schema_json.as_ptr(), + database.as_ptr(), + table_name.as_ptr(), + branch.as_ptr(), + std::ptr::null(), + 1, + ); + assert!(null_options.table.is_null()); + assert!(!null_options.error.is_null()); + assert_eq!( + (*null_options.error).code, + PaimonErrorCode::InvalidInput as i32 + ); + paimon_error_free(null_options.error); + + let invalid_branch = CString::new("../dev").unwrap(); + let unsafe_branch = paimon_table_from_schema_json( + path.as_ptr(), + schema_json.as_ptr(), + database.as_ptr(), + table_name.as_ptr(), + invalid_branch.as_ptr(), + std::ptr::null(), + 0, + ); + assert!(unsafe_branch.table.is_null()); + assert!(!unsafe_branch.error.is_null()); + assert_eq!( + (*unsafe_branch.error).code, + PaimonErrorCode::InvalidInput as i32 + ); + paimon_error_free(unsafe_branch.error); + } +} + +#[test] +fn test_table_from_schema_json_rejects_invalid_identifier() { + let path = CString::new("memory:/test_resolved_bad_ident").unwrap(); + let schema = simple_table_schema(); + let schema_json = CString::new(serde_json::to_string(&schema).unwrap()).unwrap(); + let branch = CString::new("main").unwrap(); + // Path separators are rejected by Identifier::validate. + let database = CString::new("default").unwrap(); + let table_name = CString::new("nested/table").unwrap(); + + unsafe { + let result = paimon_table_from_schema_json( + path.as_ptr(), + schema_json.as_ptr(), + database.as_ptr(), + table_name.as_ptr(), + branch.as_ptr(), + std::ptr::null(), + 0, + ); + assert!(result.table.is_null()); + assert!(!result.error.is_null()); + assert_eq!((*result.error).code, PaimonErrorCode::InvalidInput as i32); + paimon_error_free(result.error); + } +} + // ========================================================================= // Read path tests // ========================================================================= diff --git a/bindings/c/src/vector_search.rs b/bindings/c/src/vector_search.rs index 4a3e0eccd..80d79efed 100644 --- a/bindings/c/src/vector_search.rs +++ b/bindings/c/src/vector_search.rs @@ -41,7 +41,8 @@ use crate::types::*; /// Create a new vector-search builder from a Table. /// /// # Safety -/// `table` must be a valid pointer from `paimon_catalog_get_table`, or null (returns error). +/// `table` must be a valid pointer from `paimon_catalog_get_table` or +/// `paimon_table_from_schema_json`, or null (returns error). #[no_mangle] pub unsafe extern "C" fn paimon_table_new_vector_search_builder( table: *const paimon_table, diff --git a/bindings/c/src/write.rs b/bindings/c/src/write.rs index 757923491..8e9637997 100644 --- a/bindings/c/src/write.rs +++ b/bindings/c/src/write.rs @@ -76,7 +76,8 @@ unsafe fn new_write_builder( /// used by both `new_write()` and `new_commit()` for duplicate-commit detection. /// /// # Safety -/// `table` must be a valid pointer from `paimon_catalog_get_table`, or null (returns error). +/// `table` must be a valid pointer from `paimon_catalog_get_table` or +/// `paimon_table_from_schema_json`, or null (returns error). #[no_mangle] pub unsafe extern "C" fn paimon_table_new_write_builder( table: *const paimon_table, diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 0245cf414..5c6a50e63 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -164,6 +164,97 @@ impl TableSchema { new_schema } + /// Validate the structural invariants of an externally supplied, already + /// resolved schema, without normalizing or mutating it. + /// + /// This is the safety check for [`crate::table::Table::from_resolved_schema`], + /// whose input is untrusted JSON. It rejects the malformed shapes that would + /// otherwise panic or silently read the wrong column downstream: + /// - duplicate top-level field names (a projected name could resolve to a + /// different field than a predicate on the same name), + /// - duplicate field ids (including nested), which break schema-evolution + /// column mapping, + /// - primary-key or partition columns that do not exist (e.g. the + /// `PartitionComputer` unwraps on a missing partition column), + /// - primary keys that are exactly the partition columns (the read path + /// selects the key-value merge path from the raw primary keys but feeds + /// the reader the *trimmed* keys, which are then empty, panicking on a + /// zero-column key), + /// - reserved system field names / ids (e.g. a user column named `_ROW_ID` + /// would be silently replaced by the system row number on read). + /// + /// It intentionally does NOT run create-time policy checks (merge-engine, + /// changelog, aggregation, rowkind, blob strategy, bucket-key existence, …): + /// those normalize the schema or reject shapes that are valid to *read*, + /// which is not this entry point's contract. + pub(crate) fn validate_resolved_structure(&self) -> crate::Result<()> { + let field_names: Vec = self.fields.iter().map(|f| f.name().to_string()).collect(); + Schema::validate_no_duplicate_fields(&field_names)?; + Schema::validate_primary_keys(&field_names, &self.primary_keys)?; + Schema::validate_partition_keys(&field_names, &self.partition_keys)?; + Schema::validate_primary_keys_not_partition_only(&self.partition_keys, &self.primary_keys)?; + self.validate_no_duplicate_field_ids()?; + self.validate_no_reserved_fields()?; + Ok(()) + } + + /// Reject reserved system field names, the `_KEY_` key-field prefix, and + /// reserved system field ids, mirroring Java `SpecialFields`. A user column + /// colliding with a system field (e.g. `_ROW_ID`) is otherwise excluded + /// from the physical read and silently filled with the system value. + fn validate_no_reserved_fields(&self) -> crate::Result<()> { + // Java SpecialFields.SYSTEM_FIELD_NAMES. + const SYSTEM_FIELD_NAMES: [&str; 5] = [ + SEQUENCE_NUMBER_FIELD_NAME, + VALUE_KIND_FIELD_NAME, + "_LEVEL", + ROW_KIND_FIELD_NAME, + ROW_ID_FIELD_NAME, + ]; + const KEY_FIELD_PREFIX: &str = "_KEY_"; + // Java SpecialFields.SYSTEM_FIELD_ID_START = Integer.MAX_VALUE / 2. + const SYSTEM_FIELD_ID_START: i32 = i32::MAX / 2; + + for field in &self.fields { + let name = field.name(); + if name.starts_with(KEY_FIELD_PREFIX) || SYSTEM_FIELD_NAMES.contains(&name) { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Field name '{name}' is reserved for system use and cannot be used in a table schema" + ), + }); + } + if field.id() >= SYSTEM_FIELD_ID_START { + return Err(crate::Error::DataInvalid { + message: format!( + "Field '{name}' uses reserved system field id {}", + field.id() + ), + source: None, + }); + } + } + Ok(()) + } + + /// Reject duplicate field ids, including ids nested inside complex types. + /// Field ids key schema-evolution column mapping, so a collision would map + /// two logical columns onto one id. + fn validate_no_duplicate_field_ids(&self) -> crate::Result<()> { + let mut seen = HashSet::new(); + let mut ids = Vec::new(); + collect_field_ids(&self.fields, &mut ids); + for id in ids { + if !seen.insert(id) { + return Err(crate::Error::DataInvalid { + message: format!("Table schema must not contain duplicate field id: {id}"), + source: None, + }); + } + } + Ok(()) + } + /// Apply a list of schema changes and return a new schema with incremented ID. /// /// Column-level changes operate on **top-level** columns only: a @@ -610,6 +701,28 @@ fn highest_nested_field_id(data_type: &DataType) -> i32 { } } +/// Collect the field ids of `fields` and of any fields nested inside their +/// complex types, in traversal order. Used to detect duplicate ids. +fn collect_field_ids(fields: &[DataField], out: &mut Vec) { + for field in fields { + out.push(field.id()); + collect_nested_field_ids(field.data_type(), out); + } +} + +fn collect_nested_field_ids(data_type: &DataType, out: &mut Vec) { + match data_type { + DataType::Array(t) => collect_nested_field_ids(t.element_type(), out), + DataType::Multiset(t) => collect_nested_field_ids(t.element_type(), out), + DataType::Map(t) => { + collect_nested_field_ids(t.key_type(), out); + collect_nested_field_ids(t.value_type(), out); + } + DataType::Row(t) => collect_field_ids(t.fields(), out), + _ => {} + } +} + /// Reassign the IDs of all row fields nested inside a data type from the /// table-wide highest field ID, so they cannot collide with existing fields. /// @@ -2572,6 +2685,173 @@ mod tests { DataType::Vector(VectorType::try_new(true, 4, DataType::Float(FloatType::new())).unwrap()) } + // Build a raw TableSchema without going through Schema::build validation, + // so we can exercise validate_resolved_structure against malformed input. + fn raw_table_schema( + fields: Vec, + partition_keys: Vec, + primary_keys: Vec, + ) -> TableSchema { + let highest_field_id = TableSchema::current_highest_field_id(&fields); + TableSchema { + version: TableSchema::CURRENT_VERSION, + id: 0, + fields, + highest_field_id, + partition_keys, + primary_keys, + options: HashMap::new(), + comment: None, + time_millis: 0, + } + } + + #[test] + fn test_validate_resolved_structure_accepts_valid_schema() { + let schema = raw_table_schema( + vec![ + DataField::new(0, "id".to_string(), DataType::Int(IntType::new())), + DataField::new(1, "name".to_string(), DataType::Int(IntType::new())), + ], + vec![], + vec!["id".to_string()], + ); + assert!(schema.validate_resolved_structure().is_ok()); + } + + #[test] + fn test_validate_resolved_structure_rejects_missing_primary_key() { + let schema = raw_table_schema( + vec![DataField::new( + 0, + "id".to_string(), + DataType::Int(IntType::new()), + )], + vec![], + vec!["missing".to_string()], + ); + let err = schema.validate_resolved_structure().unwrap_err(); + assert!( + matches!(err, crate::Error::ConfigInvalid { .. }), + "missing PK column should be rejected, got {err:?}" + ); + } + + #[test] + fn test_validate_resolved_structure_rejects_missing_partition_key() { + let schema = raw_table_schema( + vec![DataField::new( + 0, + "id".to_string(), + DataType::Int(IntType::new()), + )], + vec!["missing".to_string()], + vec![], + ); + let err = schema.validate_resolved_structure().unwrap_err(); + assert!( + matches!(err, crate::Error::ConfigInvalid { .. }), + "missing partition column should be rejected, got {err:?}" + ); + } + + #[test] + fn test_validate_resolved_structure_rejects_duplicate_field_names() { + let schema = raw_table_schema( + vec![ + DataField::new(0, "id".to_string(), DataType::Int(IntType::new())), + DataField::new(1, "id".to_string(), DataType::Int(IntType::new())), + ], + vec![], + vec![], + ); + let err = schema.validate_resolved_structure().unwrap_err(); + assert!( + matches!(err, crate::Error::ConfigInvalid { .. }), + "duplicate field names should be rejected, got {err:?}" + ); + } + + #[test] + fn test_validate_resolved_structure_rejects_duplicate_field_ids() { + let schema = raw_table_schema( + vec![ + DataField::new(0, "id".to_string(), DataType::Int(IntType::new())), + DataField::new(0, "name".to_string(), DataType::Int(IntType::new())), + ], + vec![], + vec![], + ); + let err = schema.validate_resolved_structure().unwrap_err(); + assert!( + matches!(err, crate::Error::DataInvalid { .. }), + "duplicate field ids should be rejected, got {err:?}" + ); + } + + #[test] + fn test_validate_resolved_structure_rejects_partition_only_primary_key() { + // PK == partition columns: the read path selects the KV/merge path from + // the raw primary keys but feeds the reader the trimmed keys (empty), + // which panics on a zero-column key. Must be rejected up front. + let schema = raw_table_schema( + vec![ + DataField::new(0, "p".to_string(), DataType::Int(IntType::new())), + DataField::new(1, "v".to_string(), DataType::Int(IntType::new())), + ], + vec!["p".to_string()], + vec!["p".to_string()], + ); + let err = schema.validate_resolved_structure().unwrap_err(); + assert!( + matches!(err, crate::Error::ConfigInvalid { .. }), + "partition-only primary key should be rejected, got {err:?}" + ); + } + + #[test] + fn test_validate_resolved_structure_rejects_reserved_field_name() { + for reserved in [ + "_ROW_ID", + "_SEQUENCE_NUMBER", + "_VALUE_KIND", + "_LEVEL", + "_KEY_x", + ] { + let schema = raw_table_schema( + vec![ + DataField::new(0, "id".to_string(), DataType::Int(IntType::new())), + DataField::new(1, reserved.to_string(), DataType::Int(IntType::new())), + ], + vec![], + vec![], + ); + let err = schema.validate_resolved_structure().unwrap_err(); + assert!( + matches!(err, crate::Error::ConfigInvalid { .. }), + "reserved field name {reserved:?} should be rejected, got {err:?}" + ); + } + } + + #[test] + fn test_validate_resolved_structure_rejects_reserved_field_id() { + let schema = raw_table_schema( + vec![DataField::new( + i32::MAX / 2, + "id".to_string(), + DataType::Int(IntType::new()), + )], + vec![], + vec![], + ); + let err = schema.validate_resolved_structure().unwrap_err(); + assert!( + matches!(err, crate::Error::DataInvalid { .. }), + "reserved system field id should be rejected, got {err:?}" + ); + } + #[test] fn test_vector_rejected_as_primary_key() { let err = Schema::builder() diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index be680cb0d..4078e3a23 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -189,6 +189,46 @@ impl Table { } } + /// Create a table from an already resolved schema without loading a catalog. + /// + /// The supplied schema is preserved as-is (no normalization). Its structural + /// invariants are validated — primary-key/partition columns must exist and + /// field names/ids must be unique — so a malformed external schema is + /// rejected here instead of panicking or reading the wrong column later. The + /// branch only selects the branch-scoped managers used by subsequent reads. + pub fn from_resolved_schema( + file_io: FileIO, + identifier: Identifier, + location: String, + schema: TableSchema, + branch: impl Into, + ) -> Result { + let branch = branch.into(); + identifier.validate()?; + validate_branch_name(&branch)?; + schema.validate_resolved_structure()?; + let schema_manager = SchemaManager::new(file_io.clone(), location.clone()); + let schema_manager = if branch == DEFAULT_MAIN_BRANCH { + schema_manager + } else { + schema_manager.with_branch(&branch) + }; + let branch_reference = branch != DEFAULT_MAIN_BRANCH; + + Ok(Self { + file_io, + identifier, + location, + schema, + schema_manager, + branch, + branch_reference, + rest_env: None, + time_traveled: false, + travel_snapshot: None, + }) + } + /// Get the table's identifier. pub fn identifier(&self) -> &Identifier { &self.identifier