This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 25a04aac feat(c): build a catalog-free Table from a resolved schema
JSON (#595)
25a04aac is described below
commit 25a04aac6ddc0ac523fbee4393cc50c8b462387c
Author: Junrui Lee <[email protected]>
AuthorDate: Thu Jul 30 09:09:31 2026 +0800
feat(c): build a catalog-free Table from a resolved schema JSON (#595)
---
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 7ee0257e..47eb4598 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 f02c3e83..2340f970 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 05f82e31..0ccf6d34 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::<TableSchema>(&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 9c35fb98..d29c7eb8 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<i32>, 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 4a3e0ecc..80d79efe 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 75792349..8e963799 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 90d74ded..b3c48d13 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<String> = 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
@@ -652,6 +743,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<i32>) {
+ 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<i32>) {
+ 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.
///
@@ -2666,6 +2779,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<DataField>,
+ partition_keys: Vec<String>,
+ primary_keys: Vec<String>,
+ ) -> 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 be680cb0..4078e3a2 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<String>,
+ ) -> Result<Self> {
+ 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
