moomindani commented on code in PR #2804:
URL: https://github.com/apache/iceberg-rust/pull/2804#discussion_r3828103702
##########
crates/integrations/datafusion/src/table/mod.rs:
##########
@@ -236,6 +251,77 @@ impl TableProvider for IcebergTableProvider {
}
}
+/// Collects the `write-default` values of a schema's top-level columns as
DataFusion
+/// expressions, keyed by column name.
+///
+/// Per the spec, writers must use `write-default` for columns that are not
supplied;
+/// DataFusion's insert planner consults these defaults for columns omitted
from an
+/// `INSERT` and falls back to `NULL` otherwise. Defaults of types that cannot
be
+/// expressed as a DataFusion scalar are skipped.
+fn column_defaults_from_schema(schema: &IcebergSchema) -> HashMap<String,
Expr> {
+ schema
+ .as_struct()
+ .fields()
+ .iter()
+ .filter_map(|field| {
+ let literal = field.write_default.as_ref()?;
+ let scalar = literal_to_scalar_value(&field.field_type, literal)?;
+ Some((field.name.clone(), Expr::Literal(scalar, None)))
+ })
+ .collect()
+}
+
+/// Converts an Iceberg literal of the given type into a DataFusion
[`ScalarValue`].
+///
+/// Returns `None` for combinations that have no scalar representation; the
insert
+/// planner casts the resulting expression to the target arrow type, so minor
+/// representation differences (e.g. timezone strings) are reconciled
downstream.
+fn literal_to_scalar_value(field_type: &Type, literal: &Literal) ->
Option<ScalarValue> {
Review Comment:
Thanks for flagging this. I looked into the reuse path before deciding, and
the composition would regress today.
`type_to_arrow_type` maps `Time → Time64(Microsecond)`, `Uuid →
FixedSizeBinary(16)`, `Fixed(len) → FixedSizeBinary(len)` and `Binary →
LargeBinary`. But `create_primitive_array_single_element`
(`crates/iceberg/src/arrow/value.rs:627`) only has arms for Boolean, Int32,
Date32, Int64, Timestamp(us|ns, tz), Float32, Float64, Utf8, `Binary`,
Decimal128 and Struct(None), and falls through to `Err("Unsupported constant
type combination")`. There is no `Time64` arm, no `FixedSizeBinary` arm and no
`LargeBinary` arm. Since `column_defaults_from_schema` uses `filter_map`, those
four types would silently lose their write-default and fall back to NULL.
So the drift runs in both directions: this match handles `Time` precisely
because core does not, and the same holds for `Uuid`, `Fixed` and `Binary`.
I kept the local match and made the duplication explicit in the doc comment.
What I did take from your comment is the arrow-type alignment — see the two
replies below: `Binary` now produces `LargeBinary`, and `Fixed(len)` is sized
from the declared width, so the scalars carry exactly the types
`type_to_arrow_type` assigns and the downstream cast has nothing left to
reconcile. A new unit test asserts `scalar.data_type() ==
type_to_arrow_type(field_type)` for all 16 supported combinations, so a future
divergence fails the build.
Happy to do the core side as a follow-up if you would like the single source
of truth: add the `Time64` / `FixedSizeBinary` / `LargeBinary` arms to
`create_primitive_array_single_element`, promote it to `pub`, then reduce this
function to your three lines. That expands iceberg-core's public surface, so it
seemed better as its own PR than folded in here — but I am happy either way,
and would defer to @CTTY / @blackmwk on whether core should export it.
##########
crates/integrations/datafusion/src/table/mod.rs:
##########
@@ -236,6 +251,77 @@ impl TableProvider for IcebergTableProvider {
}
}
+/// Collects the `write-default` values of a schema's top-level columns as
DataFusion
+/// expressions, keyed by column name.
+///
+/// Per the spec, writers must use `write-default` for columns that are not
supplied;
+/// DataFusion's insert planner consults these defaults for columns omitted
from an
+/// `INSERT` and falls back to `NULL` otherwise. Defaults of types that cannot
be
+/// expressed as a DataFusion scalar are skipped.
+fn column_defaults_from_schema(schema: &IcebergSchema) -> HashMap<String,
Expr> {
+ schema
+ .as_struct()
+ .fields()
+ .iter()
+ .filter_map(|field| {
+ let literal = field.write_default.as_ref()?;
+ let scalar = literal_to_scalar_value(&field.field_type, literal)?;
+ Some((field.name.clone(), Expr::Literal(scalar, None)))
+ })
+ .collect()
+}
+
+/// Converts an Iceberg literal of the given type into a DataFusion
[`ScalarValue`].
+///
+/// Returns `None` for combinations that have no scalar representation; the
insert
+/// planner casts the resulting expression to the target arrow type, so minor
+/// representation differences (e.g. timezone strings) are reconciled
downstream.
+fn literal_to_scalar_value(field_type: &Type, literal: &Literal) ->
Option<ScalarValue> {
+ let Type::Primitive(primitive_type) = field_type else {
+ return None;
+ };
+ let Literal::Primitive(primitive) = literal else {
+ return None;
+ };
+ Some(match (primitive_type, primitive) {
+ (PrimitiveType::Boolean, PrimitiveLiteral::Boolean(v)) =>
ScalarValue::Boolean(Some(*v)),
+ (PrimitiveType::Int, PrimitiveLiteral::Int(v)) =>
ScalarValue::Int32(Some(*v)),
+ (PrimitiveType::Long, PrimitiveLiteral::Long(v)) =>
ScalarValue::Int64(Some(*v)),
+ (PrimitiveType::Float, PrimitiveLiteral::Float(v)) =>
ScalarValue::Float32(Some(v.0)),
+ (PrimitiveType::Double, PrimitiveLiteral::Double(v)) =>
ScalarValue::Float64(Some(v.0)),
+ (PrimitiveType::String, PrimitiveLiteral::String(v)) =>
ScalarValue::Utf8(Some(v.clone())),
+ (PrimitiveType::Date, PrimitiveLiteral::Int(v)) =>
ScalarValue::Date32(Some(*v)),
+ (PrimitiveType::Time, PrimitiveLiteral::Long(v)) => {
+ ScalarValue::Time64Microsecond(Some(*v))
+ }
+ (PrimitiveType::Timestamp, PrimitiveLiteral::Long(v)) => {
+ ScalarValue::TimestampMicrosecond(Some(*v), None)
+ }
+ (PrimitiveType::Timestamptz, PrimitiveLiteral::Long(v)) => {
+ ScalarValue::TimestampMicrosecond(Some(*v),
Some(UTC_TIME_ZONE.into()))
+ }
+ (PrimitiveType::TimestampNs, PrimitiveLiteral::Long(v)) => {
+ ScalarValue::TimestampNanosecond(Some(*v), None)
+ }
+ (PrimitiveType::TimestamptzNs, PrimitiveLiteral::Long(v)) => {
+ ScalarValue::TimestampNanosecond(Some(*v),
Some(UTC_TIME_ZONE.into()))
+ }
+ (PrimitiveType::Decimal { precision, scale },
PrimitiveLiteral::Int128(v)) => {
+ ScalarValue::Decimal128(Some(*v), *precision as u8, *scale as i8)
+ }
+ (PrimitiveType::Binary, PrimitiveLiteral::Binary(v)) => {
+ ScalarValue::Binary(Some(v.clone()))
+ }
+ (PrimitiveType::Fixed(_), PrimitiveLiteral::Binary(v)) => {
Review Comment:
Fixed. It now sizes the scalar from the declared `Fixed(len)` and skips the
default when the value's length contradicts the declaration:
```rust
let width = i32::try_from(*len).ok()?;
if v.len() != usize::try_from(*len).ok()? {
return None;
}
ScalarValue::FixedSizeBinary(width, Some(v.clone()))
```
Covered by `test_literal_to_scalar_value_skips_fixed_default_of_wrong_width`.
##########
crates/integrations/datafusion/src/table/mod.rs:
##########
@@ -236,6 +251,77 @@ impl TableProvider for IcebergTableProvider {
}
}
+/// Collects the `write-default` values of a schema's top-level columns as
DataFusion
+/// expressions, keyed by column name.
+///
+/// Per the spec, writers must use `write-default` for columns that are not
supplied;
+/// DataFusion's insert planner consults these defaults for columns omitted
from an
+/// `INSERT` and falls back to `NULL` otherwise. Defaults of types that cannot
be
+/// expressed as a DataFusion scalar are skipped.
+fn column_defaults_from_schema(schema: &IcebergSchema) -> HashMap<String,
Expr> {
+ schema
+ .as_struct()
+ .fields()
+ .iter()
+ .filter_map(|field| {
+ let literal = field.write_default.as_ref()?;
+ let scalar = literal_to_scalar_value(&field.field_type, literal)?;
+ Some((field.name.clone(), Expr::Literal(scalar, None)))
+ })
+ .collect()
+}
+
+/// Converts an Iceberg literal of the given type into a DataFusion
[`ScalarValue`].
+///
+/// Returns `None` for combinations that have no scalar representation; the
insert
+/// planner casts the resulting expression to the target arrow type, so minor
+/// representation differences (e.g. timezone strings) are reconciled
downstream.
+fn literal_to_scalar_value(field_type: &Type, literal: &Literal) ->
Option<ScalarValue> {
+ let Type::Primitive(primitive_type) = field_type else {
+ return None;
+ };
+ let Literal::Primitive(primitive) = literal else {
+ return None;
+ };
+ Some(match (primitive_type, primitive) {
+ (PrimitiveType::Boolean, PrimitiveLiteral::Boolean(v)) =>
ScalarValue::Boolean(Some(*v)),
+ (PrimitiveType::Int, PrimitiveLiteral::Int(v)) =>
ScalarValue::Int32(Some(*v)),
+ (PrimitiveType::Long, PrimitiveLiteral::Long(v)) =>
ScalarValue::Int64(Some(*v)),
+ (PrimitiveType::Float, PrimitiveLiteral::Float(v)) =>
ScalarValue::Float32(Some(v.0)),
+ (PrimitiveType::Double, PrimitiveLiteral::Double(v)) =>
ScalarValue::Float64(Some(v.0)),
+ (PrimitiveType::String, PrimitiveLiteral::String(v)) =>
ScalarValue::Utf8(Some(v.clone())),
+ (PrimitiveType::Date, PrimitiveLiteral::Int(v)) =>
ScalarValue::Date32(Some(*v)),
+ (PrimitiveType::Time, PrimitiveLiteral::Long(v)) => {
+ ScalarValue::Time64Microsecond(Some(*v))
+ }
+ (PrimitiveType::Timestamp, PrimitiveLiteral::Long(v)) => {
+ ScalarValue::TimestampMicrosecond(Some(*v), None)
+ }
+ (PrimitiveType::Timestamptz, PrimitiveLiteral::Long(v)) => {
+ ScalarValue::TimestampMicrosecond(Some(*v),
Some(UTC_TIME_ZONE.into()))
+ }
+ (PrimitiveType::TimestampNs, PrimitiveLiteral::Long(v)) => {
+ ScalarValue::TimestampNanosecond(Some(*v), None)
+ }
+ (PrimitiveType::TimestamptzNs, PrimitiveLiteral::Long(v)) => {
+ ScalarValue::TimestampNanosecond(Some(*v),
Some(UTC_TIME_ZONE.into()))
+ }
+ (PrimitiveType::Decimal { precision, scale },
PrimitiveLiteral::Int128(v)) => {
+ ScalarValue::Decimal128(Some(*v), *precision as u8, *scale as i8)
+ }
+ (PrimitiveType::Binary, PrimitiveLiteral::Binary(v)) => {
Review Comment:
Fixed — it emits `ScalarValue::LargeBinary` now, matching
`type_to_arrow_type`. The new
`test_literal_to_scalar_value_matches_column_arrow_type` asserts the scalar's
arrow type equals `type_to_arrow_type(field_type)` for every supported
primitive, so this cannot drift back unnoticed.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
