You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "NoahKusaba (via GitHub)" <gi...@apache.org> on 2026/09/26 20:33:31 UTC
[PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
NoahKusaba opened a new pull request, #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19
## Which issue does this PR close?
- No issue tracks this.
- Part of the Ballista-Iceberg integration (follows #14), but the bug here affects any DataFusion user writing to a partitioned table.
## What changes are included in this PR?
An `INSERT` into a **partitioned** table fails when the source has a `NOT NULL` column where the table's column is optional:
```
Error: Plan("Input schema does not match Iceberg table schema.
Expected schema: Field { "id": nullable Int32 }, Field { "category": nullable Utf8 }
Input schema: Field { "id": Int32 }, Field { "category": Utf8 }")
```
Every value of a non-nullable column is valid in an optional one, so the write is safe. The same `INSERT` into an **unpartitioned** table already succeeds, because `project_with_partition` returns before this check when the spec is unpartitioned, so today the result depends on whether the table is partitioned.
`project_with_partition` compared the input and table schemas with `==`. It now uses Arrow's `Schema::contains`, which allows the input to be narrower in nullability and is otherwise as strict as before:
- Same columns, same order, same names, same types (checked recursively for nested types).
- Field metadata is still stripped from both sides before comparing, as before.
- A nullable input column into a required table column is still rejected, at any nesting depth.
The public doc of `project_with_partition` now states this contract. The error message changes from "does not match" to "is not compatible with", since the schemas no longer need to be equal.
The new unit tests need a partitioned table with a given schema. The three existing schema-validation tests each built one inline with the same ~50 lines, so they now share a `table_partitioned_by_id` helper, with their imports moved to the top of the `tests` module. Their assertions are unchanged apart from the new error message.
## Are these changes tested?
- `test_insert_not_null_source_into_partitioned_table` (integration): inserts from a `NOT NULL` `MemTable` into a table partitioned on an optional column, and reads the rows back. It fails on `main` with the error above.
- `test_schema_validation_nullability`: non-nullable and nullable input into an optional column are both accepted; into a required column, non-nullable is accepted and nullable is rejected.
- `test_schema_validation_nested_nullability`: the same four cases for a field inside a struct.
- The existing matching, mismatched-name and metadata tests still pass.
`cargo fmt --all -- --check`, `cargo clippy --workspace --locked --all-targets -- -D warnings` and `cargo test --workspace --locked` all pass locally.
## AI Disclosure
- Used Claude Code to find the bug, write the tests and draft this description. I reviewed the change and reproduced the failure on `main`.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4128890243
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,142 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
+ }
- let input = Arc::new(EmptyExec::new(arrow_schema));
+ #[test]
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ #[test]
+ fn test_schema_validation_nested_nullability() {
+ let child = |nullable| Field::new("x", DataType::Int32, nullable);
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![child(nullable)])),
+ false,
+ ),
+ ])
+ };
+ let table = |x: NestedField| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![Arc::new(x)])),
+ ),
+ ])
+ };
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // The same rule applies inside a struct.
+ let optional = table(NestedField::optional(3, "x", int.clone()));
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
+
+ let required = table(NestedField::required(3, "x", int));
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
}
Review Comment:
Thanks, good call. Adding the list and map cases turned up a bug that's already on `main`: iceberg-rust's `strip_metadata_from_schema` fails on any list or map column with "Field stack underflow in list", so a partitioned INSERT with such a column errors out before the schemas are even compared. `MetadataStripVisitor` only pushes onto its field stack in `before_field`, which the visitor doesn't call for list elements or map keys and values.
I have a fix for iceberg-rust ready and will open a PR there. In abee042 the four cases now live in a shared `assert_nested_nullability` helper, used by the struct test, with the list and map tests added but commented out until that fix lands. I ran them against the patched iceberg-rust and they pass, including the map case with `keys_sorted = false`, which is what Iceberg maps convert to.
I hope in the time being, we can merge now and look to re-enable the commented out tests I've added until the fix is merged in?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4128966655
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,142 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
+ }
- let input = Arc::new(EmptyExec::new(arrow_schema));
+ #[test]
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ #[test]
+ fn test_schema_validation_nested_nullability() {
+ let child = |nullable| Field::new("x", DataType::Int32, nullable);
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![child(nullable)])),
+ false,
+ ),
+ ])
+ };
+ let table = |x: NestedField| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![Arc::new(x)])),
+ ),
+ ])
+ };
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // The same rule applies inside a struct.
+ let optional = table(NestedField::optional(3, "x", int.clone()));
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
+
+ let required = table(NestedField::required(3, "x", int));
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
}
Review Comment:
Filed apache/iceberg-rust#3297 for the `strip_metadata_from_schema` bug. The fix plus a regression test is ready locally, and I'll open the PR after apache/iceberg-rust#3286 and apache/iceberg-rust#2904 are in.
The staged list and map tests are [here](https://github.com/apache/datafusion-iceberg/blob/abee0425e1c7af8ff0f5b323209c6d36e27fa4d1/crates/datafusion/src/physical_plan/project.rs#L831-L878). Once datafusion-iceberg is on an iceberg-rust rev with the fix, enabling them is just uncommenting, since they already pass against `665c64e` with the fix applied.
Two list/map limitations of `contains` I noticed while writing them, neither new in this PR (the old `==` had both):
- Field names are compared, so a list whose element is named `item` (Arrow's default, used by `DataType::new_list`) won't match Iceberg's `element`.
- A source map with `keys_sorted = true` is rejected, since Iceberg maps convert to unsorted Arrow maps.
If you think either is worth handling, I can fold them into #22 or open a separate issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#issuecomment-5882268606
Thanks for testing this so thoroughly, @mbutrovich. Agreed on the unpartitioned gap: because the sink schema is the input's own schema, `check_not_null_constraints` never runs. I'll open an issue that covers both sides, the unpartitioned path writing NULLs into required columns and the partitioned path rejecting nullable sources at plan time, with passing the table's Arrow schema (plus the partition column) as the sink schema as the likely fix for both.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4128890243
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,142 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
+ }
- let input = Arc::new(EmptyExec::new(arrow_schema));
+ #[test]
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ #[test]
+ fn test_schema_validation_nested_nullability() {
+ let child = |nullable| Field::new("x", DataType::Int32, nullable);
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![child(nullable)])),
+ false,
+ ),
+ ])
+ };
+ let table = |x: NestedField| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![Arc::new(x)])),
+ ),
+ ])
+ };
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // The same rule applies inside a struct.
+ let optional = table(NestedField::optional(3, "x", int.clone()));
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
+
+ let required = table(NestedField::required(3, "x", int));
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
}
Review Comment:
Thanks, good call. Adding the list and map cases turned up a bug already on `main`: iceberg-rust's `strip_metadata_from_schema` fails on any list or map column with "Field stack underflow in list", so a partitioned INSERT with such a column errors out before the schemas are compared. `MetadataStripVisitor` only pushes onto its field stack in `before_field`, which the visitor doesn't call for list elements or map keys and values. Filed as apache/iceberg-rust#3297.
The fix and a regression test are ready. I'll open that PR once my pending iceberg-rust PRs are merged (apache/iceberg-rust#3286, apache/iceberg-rust#2904), since reviews there are bandwidth-limited.
In abee042 the four cases live in a shared `assert_nested_nullability` helper used by the struct test, and the list and map tests are [staged but commented out](https://github.com/apache/datafusion-iceberg/blob/abee0425e1c7af8ff0f5b323209c6d36e27fa4d1/crates/datafusion/src/physical_plan/project.rs#L831-L878) until the fix lands. They pass against `665c64e` with the fix applied, including the map case with `keys_sorted = false` (what Iceberg maps convert to), so enabling them is just uncommenting.
Two list/map limitations of `contains` I noticed while writing them, neither new in this PR (the old `==` had both):
- Field names are compared, so a list whose element is named `item` (Arrow's default, used by `DataType::new_list`) won't match Iceberg's `element`.
- A source map with `keys_sorted = true` is rejected, since Iceberg maps convert to unsorted Arrow maps.
If you think either is worth handling, I can fold them into #22 or open a separate issue.
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,142 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
+ }
- let input = Arc::new(EmptyExec::new(arrow_schema));
+ #[test]
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ #[test]
+ fn test_schema_validation_nested_nullability() {
+ let child = |nullable| Field::new("x", DataType::Int32, nullable);
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![child(nullable)])),
+ false,
+ ),
+ ])
+ };
+ let table = |x: NestedField| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![Arc::new(x)])),
+ ),
+ ])
+ };
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // The same rule applies inside a struct.
+ let optional = table(NestedField::optional(3, "x", int.clone()));
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
+
+ let required = table(NestedField::required(3, "x", int));
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
}
Review Comment:
I'm going to wait until my other pending datafusion related PR's are merged into Iceberg-rust, as they are severely bandwidth restricted on reviews:
- https://github.com/apache/iceberg-rust/pull/3286
- https://github.com/apache/iceberg-rust/pull/2904
I also have an additional 5 unrelated PR's I want to eventually merge into iceberg-rust that I have written down as well.
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,142 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
+ }
- let input = Arc::new(EmptyExec::new(arrow_schema));
+ #[test]
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ #[test]
+ fn test_schema_validation_nested_nullability() {
+ let child = |nullable| Field::new("x", DataType::Int32, nullable);
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![child(nullable)])),
+ false,
+ ),
+ ])
+ };
+ let table = |x: NestedField| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![Arc::new(x)])),
+ ),
+ ])
+ };
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // The same rule applies inside a struct.
+ let optional = table(NestedField::optional(3, "x", int.clone()));
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
+
+ let required = table(NestedField::required(3, "x", int));
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
}
Review Comment:
Filed apache/iceberg-rust#3297 for the `strip_metadata_from_schema` bug. The fix plus a regression test is ready locally, and I'll open the PR after apache/iceberg-rust#3286 and apache/iceberg-rust#2904 are in.
The staged list and map tests are [here](https://github.com/apache/datafusion-iceberg/blob/abee0425e1c7af8ff0f5b323209c6d36e27fa4d1/crates/datafusion/src/physical_plan/project.rs#L831-L878). Once datafusion-iceberg is on an iceberg-rust rev with the fix, enabling them is just uncommenting, since they already pass against `665c64e` with the fix applied.
Two list/map limitations of `contains` I noticed while writing them, neither new in this PR (the old `==` had both):
- Field names are compared, so a list whose element is named `item` (Arrow's default, used by `DataType::new_list`) won't match Iceberg's `element`.
- A source map with `keys_sorted = true` is rejected, since Iceberg maps convert to unsorted Arrow maps.
If you think either is worth handling, I can fold them into #22 or open a separate issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4158409347
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,203 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
}
#[test]
- fn test_schema_validation_with_metadata_differences() {
- use std::collections::HashMap;
-
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
- .unwrap()
- .build()
- .unwrap();
+ /// Checks the nullability rule for a nested field: `input(nullable)` has a
+ /// required `id` and a column `c` whose nested int is `nullable`, and
+ /// `nested_table(required)` has the matching table column.
+ fn assert_nested_nullability(
+ input: impl Fn(bool) -> DataType,
+ nested_table: impl Fn(bool) -> Type,
+ ) {
+ let input = |nullable| {
Review Comment:
Renamed in 26be163.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4128920899
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,142 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
+ }
- let input = Arc::new(EmptyExec::new(arrow_schema));
+ #[test]
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ #[test]
+ fn test_schema_validation_nested_nullability() {
+ let child = |nullable| Field::new("x", DataType::Int32, nullable);
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![child(nullable)])),
+ false,
+ ),
+ ])
+ };
+ let table = |x: NestedField| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![Arc::new(x)])),
+ ),
+ ])
+ };
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // The same rule applies inside a struct.
+ let optional = table(NestedField::optional(3, "x", int.clone()));
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
+
+ let required = table(NestedField::required(3, "x", int));
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
}
Review Comment:
I'm going to wait until my other pending datafusion related PR's are merged into Iceberg-rust, as they are severely bandwidth restricted on reviews:
- https://github.com/apache/iceberg-rust/pull/3286
- https://github.com/apache/iceberg-rust/pull/2904
I also have an additional 5 unrelated PR's I want to eventually merge into iceberg-rust that I have written down as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4128890243
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,142 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
+ }
- let input = Arc::new(EmptyExec::new(arrow_schema));
+ #[test]
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ #[test]
+ fn test_schema_validation_nested_nullability() {
+ let child = |nullable| Field::new("x", DataType::Int32, nullable);
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![child(nullable)])),
+ false,
+ ),
+ ])
+ };
+ let table = |x: NestedField| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![Arc::new(x)])),
+ ),
+ ])
+ };
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // The same rule applies inside a struct.
+ let optional = table(NestedField::optional(3, "x", int.clone()));
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
+
+ let required = table(NestedField::required(3, "x", int));
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
}
Review Comment:
Thanks, good call. Adding the list and map cases turned up a bug that's already on `main`: iceberg-rust's `strip_metadata_from_schema` fails on any list or map column with "Field stack underflow in list", so a partitioned INSERT with such a column errors out before the schemas are even compared. `MetadataStripVisitor` only pushes onto its field stack in `before_field`, which the visitor doesn't call for list elements or map keys and values.
I have a fix for iceberg-rust ready and will open a PR there. In abee042 the four cases now live in a shared `assert_nested_nullability` helper, used by the struct test, with the list and map tests added but commented out until that fix lands. I ran them against the patched iceberg-rust and they pass, including the map case with `keys_sorted = false`, which is what Iceberg maps convert to.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4128920899
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,142 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
+ }
- let input = Arc::new(EmptyExec::new(arrow_schema));
+ #[test]
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ #[test]
+ fn test_schema_validation_nested_nullability() {
+ let child = |nullable| Field::new("x", DataType::Int32, nullable);
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![child(nullable)])),
+ false,
+ ),
+ ])
+ };
+ let table = |x: NestedField| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![Arc::new(x)])),
+ ),
+ ])
+ };
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // The same rule applies inside a struct.
+ let optional = table(NestedField::optional(3, "x", int.clone()));
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
+
+ let required = table(NestedField::required(3, "x", int));
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
}
Review Comment:
I'm going to wait until my other pending datafusion related PR's are merged into Iceberg-rust, as they are severely bandwidth restricted on reviews:
- https://github.com/apache/iceberg-rust/pull/3286
- https://github.com/apache/iceberg-rust/pull/2904
I also have an additional 6 unrelated PR's I want to eventually merge into iceberg-rust that I have written down as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#issuecomment-5882414395
@mbutrovich Opened #22 for the unpartitioned gap, with your repro. It covers both sides:
- **Unpartitioned:** `IcebergWriteExec` passes the input's schema as the sink schema to `execute_input_stream`, so `check_not_null_constraints` never runs and NULLs land in required columns as `0`.
- **Partitioned:** a nullable source column into a required one is rejected at plan time, even when it holds no NULLs.
I'd like to keep this PR to the narrower fix, accepting NOT NULL sources into optional columns, and leave the plan-time rejection as it is. Once #22 passes the table's schema as the sink schema, that rejection can be dropped in favour of DataFusion's runtime check, and both paths will behave the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#issuecomment-5936729142
Thank you for taking the time out of your busy schedule to deeply review my change, and for supporting the ballista-iceberg initiative.
I need to spend more time digesting your comment, before I can give a more proper response (if I need to make one).
Also your book is amazing, thank you for all your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4128890243
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,142 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
+ }
- let input = Arc::new(EmptyExec::new(arrow_schema));
+ #[test]
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ #[test]
+ fn test_schema_validation_nested_nullability() {
+ let child = |nullable| Field::new("x", DataType::Int32, nullable);
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![child(nullable)])),
+ false,
+ ),
+ ])
+ };
+ let table = |x: NestedField| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![Arc::new(x)])),
+ ),
+ ])
+ };
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // The same rule applies inside a struct.
+ let optional = table(NestedField::optional(3, "x", int.clone()));
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
+
+ let required = table(NestedField::required(3, "x", int));
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
}
Review Comment:
Thanks, good call. Adding the list and map cases turned up a bug that's already on `main`: iceberg-rust's `strip_metadata_from_schema` fails on any list or map column with "Field stack underflow in list", so a partitioned INSERT with such a column errors out before the schemas are even compared. `MetadataStripVisitor` only pushes onto its field stack in `before_field`, which the visitor doesn't call for list elements or map keys and values.
I have a fix for iceberg-rust ready and will open a PR there. In abee042 the four cases now live in a shared `assert_nested_nullability` helper, used by the struct test, with the list and map tests added but commented out until that fix lands. I ran them against the patched iceberg-rust and they pass, including the map case with `keys_sorted = false`, which is what Iceberg maps convert to.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4158150085
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,203 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
}
#[test]
- fn test_schema_validation_with_metadata_differences() {
- use std::collections::HashMap;
-
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
- .unwrap()
- .build()
- .unwrap();
+ /// Checks the nullability rule for a nested field: `input(nullable)` has a
+ /// required `id` and a column `c` whose nested int is `nullable`, and
+ /// `nested_table(required)` has the matching table column.
+ fn assert_nested_nullability(
+ input: impl Fn(bool) -> DataType,
+ nested_table: impl Fn(bool) -> Type,
+ ) {
+ let input = |nullable| {
Review Comment:
Nit: this closure shadows the `input` parameter it calls, which took me a second read. Maybe rename the parameter to `input_type`?
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,203 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
}
#[test]
- fn test_schema_validation_with_metadata_differences() {
- use std::collections::HashMap;
-
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
- .unwrap()
- .build()
- .unwrap();
+ /// Checks the nullability rule for a nested field: `input(nullable)` has a
+ /// required `id` and a column `c` whose nested int is `nullable`, and
+ /// `nested_table(required)` has the matching table column.
+ fn assert_nested_nullability(
+ input: impl Fn(bool) -> DataType,
+ nested_table: impl Fn(bool) -> Type,
+ ) {
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("c", input(nullable), false),
+ ])
+ };
+ let table = |required| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "c", nested_table(required)),
+ ])
+ };
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ let optional = table(false);
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- HashMap::new(),
- )
- .unwrap();
+ let required = table(true);
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ #[test]
+ fn test_schema_validation_struct_nullability() {
+ let int = Type::Primitive(PrimitiveType::Int);
+ assert_nested_nullability(
+ |nullable| {
+ DataType::Struct(Fields::from(vec![Field::new(
+ "x",
+ DataType::Int32,
+ nullable,
+ )]))
+ },
+ |required| {
+ let x = NestedField::new(3, "x", int.clone(), required);
+ Type::Struct(StructType::new(vec![Arc::new(x)]))
+ },
+ );
+ }
- // Create Arrow schema with metadata (should be ignored in comparison)
- let mut metadata = HashMap::new();
- metadata.insert("extra".to_string(), "metadata".to_string());
+ // TODO: enable once iceberg-rust's `strip_metadata_from_schema` handles lists
Review Comment:
Could these be `#[ignore = "apache/iceberg-rust#3297"]` tests instead of comments? They'd keep compiling, and enabling them would just mean removing the attribute.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4158271984
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,203 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
}
#[test]
- fn test_schema_validation_with_metadata_differences() {
- use std::collections::HashMap;
-
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
- .unwrap()
- .build()
- .unwrap();
+ /// Checks the nullability rule for a nested field: `input(nullable)` has a
+ /// required `id` and a column `c` whose nested int is `nullable`, and
+ /// `nested_table(required)` has the matching table column.
+ fn assert_nested_nullability(
+ input: impl Fn(bool) -> DataType,
+ nested_table: impl Fn(bool) -> Type,
+ ) {
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("c", input(nullable), false),
+ ])
+ };
+ let table = |required| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "c", nested_table(required)),
+ ])
+ };
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ let optional = table(false);
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- HashMap::new(),
- )
- .unwrap();
+ let required = table(true);
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ #[test]
+ fn test_schema_validation_struct_nullability() {
+ let int = Type::Primitive(PrimitiveType::Int);
+ assert_nested_nullability(
+ |nullable| {
+ DataType::Struct(Fields::from(vec![Field::new(
+ "x",
+ DataType::Int32,
+ nullable,
+ )]))
+ },
+ |required| {
+ let x = NestedField::new(3, "x", int.clone(), required);
+ Type::Struct(StructType::new(vec![Arc::new(x)]))
+ },
+ );
+ }
- // Create Arrow schema with metadata (should be ignored in comparison)
- let mut metadata = HashMap::new();
- metadata.insert("extra".to_string(), "metadata".to_string());
+ // TODO: enable once iceberg-rust's `strip_metadata_from_schema` handles lists
Review Comment:
I actually opened the PR on Iceberg-rust to let us enable these tests.
It's getting better traction than I thought thanks to compheads help, so I think we should just merge these tests in enabled when the iceberg-rust changes are in.
https://github.com/apache/iceberg-rust/pull/3303
I'll set the PR to draft as well in the meantime. We can also wait for #35 to get merged in before too :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#discussion_r4126428819
##########
crates/datafusion/src/physical_plan/project.rs:
##########
@@ -675,215 +688,142 @@ mod tests {
assert_eq!(city_partition.value(1), "Los Angeles");
}
- #[test]
- fn test_schema_validation_matching_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
+ /// A table with `fields`, partitioned by identity on its `id` column.
+ fn table_partitioned_by_id(fields: Vec<NestedField>) -> Table {
+ let table_schema = Schema::builder()
+ .with_fields(fields.into_iter().map(Arc::new))
+ .build()
+ .unwrap();
let partition_spec = PartitionSpec::builder(table_schema.clone())
.add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
.build()
.unwrap();
-
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
-
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
+ let sort_order = SortOrder::builder().build(&table_schema).unwrap();
+ let metadata = TableMetadataBuilder::new(
+ table_schema,
partition_spec,
sort_order,
"/test/table".to_string(),
FormatVersion::V2,
- std::collections::HashMap::new(),
+ HashMap::new(),
)
- .unwrap();
-
- let table_metadata = table_metadata_builder.build().unwrap();
-
- // Create Arrow schema matching the table schema
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32, false),
- Field::new("name", DataType::Utf8, false),
- ]));
-
- let input = Arc::new(EmptyExec::new(arrow_schema));
-
- let table = Table::builder()
- .metadata(table_metadata.metadata)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ Table::builder()
+ .metadata(metadata)
.identifier(TableIdent::from_strs(["test", "table"]).unwrap())
.file_io(FileIO::new_with_fs())
.metadata_location("/test/metadata.json")
.runtime(test_runtime())
.build()
- .unwrap();
-
- let result = project_with_partition(input, &table);
- assert!(result.is_ok(), "Schema validation should pass");
- }
-
- #[test]
- fn test_schema_validation_mismatched_schemas() {
- use iceberg::TableIdent;
- use iceberg::io::FileIO;
- use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-
- let table_schema = Arc::new(
- Schema::builder()
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "name",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap(),
- );
-
- let partition_spec = PartitionSpec::builder(table_schema.clone())
- .add_partition_field("id", "id_partition", Transform::Identity)
.unwrap()
- .build()
- .unwrap();
+ }
- let sort_order = iceberg::spec::SortOrder::builder()
- .build(&table_schema)
- .unwrap();
+ /// Required `id: int` and `name: string` columns.
+ fn id_and_name_table() -> Table {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)),
+ ])
+ }
- let table_metadata_builder = iceberg::spec::TableMetadataBuilder::new(
- (*table_schema).clone(),
- partition_spec,
- sort_order,
- "/test/table".to_string(),
- FormatVersion::V2,
- std::collections::HashMap::new(),
- )
- .unwrap();
+ fn input_of(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(ArrowSchema::new(fields))))
+ }
- let table_metadata = table_metadata_builder.build().unwrap();
+ const INCOMPATIBLE: &str = "Input schema is not compatible with Iceberg table schema";
- // Create Arrow schema with different field name (mismatched)
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ #[test]
+ fn test_schema_validation_matching_schemas() {
+ let input = input_of(vec![
Field::new("id", DataType::Int32, false),
- Field::new("different_name", DataType::Utf8, false), // Wrong field name
- ]));
+ Field::new("name", DataType::Utf8, false),
+ ]);
+ assert!(project_with_partition(input, &id_and_name_table()).is_ok());
+ }
- let input = Arc::new(EmptyExec::new(arrow_schema));
+ #[test]
+ fn test_schema_validation_mismatched_schemas() {
+ let input = input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("different_name", DataType::Utf8, false),
+ ]);
+ let err = project_with_partition(input, &id_and_name_table())
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let table = Table::builder()
- .metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test", "table"]).unwrap())
- .file_io(FileIO::new_with_fs())
- .metadata_location("/test/metadata.json")
- .runtime(test_runtime())
- .build()
- .unwrap();
+ #[test]
+ fn test_schema_validation_nullability() {
+ let id = |nullable| input_of(vec![Field::new("id", DataType::Int32, nullable)]);
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // A non-nullable input fits an optional column, e.g. an INSERT from a
+ // NOT NULL source.
+ let optional =
+ table_partitioned_by_id(vec![NestedField::optional(1, "id", int.clone())]);
+ assert!(project_with_partition(id(false), &optional).is_ok());
+ assert!(project_with_partition(id(true), &optional).is_ok());
+
+ // A nullable input could write nulls into a required column.
+ let required = table_partitioned_by_id(vec![NestedField::required(1, "id", int)]);
+ assert!(project_with_partition(id(false), &required).is_ok());
+ let err = project_with_partition(id(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
+ }
- let result = project_with_partition(input, &table);
- assert!(
- result.is_err(),
- "Schema validation should fail for mismatched schemas"
- );
- assert!(
- result
- .unwrap_err()
- .to_string()
- .contains("Input schema does not match Iceberg table schema")
- );
+ #[test]
+ fn test_schema_validation_nested_nullability() {
+ let child = |nullable| Field::new("x", DataType::Int32, nullable);
+ let input = |nullable| {
+ input_of(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![child(nullable)])),
+ false,
+ ),
+ ])
+ };
+ let table = |x: NestedField| {
+ table_partitioned_by_id(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)),
+ NestedField::required(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![Arc::new(x)])),
+ ),
+ ])
+ };
+ let int = Type::Primitive(PrimitiveType::Int);
+
+ // The same rule applies inside a struct.
+ let optional = table(NestedField::optional(3, "x", int.clone()));
+ assert!(project_with_partition(input(false), &optional).is_ok());
+ assert!(project_with_partition(input(true), &optional).is_ok());
+
+ let required = table(NestedField::required(3, "x", int));
+ assert!(project_with_partition(input(false), &required).is_ok());
+ let err = project_with_partition(input(true), &required)
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(INCOMPATIBLE), "{err}");
}
Review Comment:
The new doc says the rule holds "at any nesting depth", and this test covers a struct field. Could we add the same four cases for a list element and a map value? Lists and maps go through their own arms of Arrow's [`DataType::contains`](https://github.com/apache/arrow-rs/blob/f90e061326bd821a7af09281d9e92de6f3b603d9/arrow-schema/src/datatype.rs#L817-L838), and the map arm also compares the `keys_sorted` flag, so a struct test doesn't cover them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: accept NOT NULL input columns in partitioned INSERT [datafusion-iceberg]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on PR #19:
URL: https://github.com/apache/datafusion-iceberg/pull/19#issuecomment-6351238543
The upstream fix on iceberg-rust got merged in a few weeks ago
github.com/apache/iceberg-rust/pull/3303
Unfortunately #35 is still open without an open PR :cry: .
Whenever I / somebody else gets to it, I can look to address Andrew's comments and get this merged in.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org