Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions datafusion-executor/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

47 changes: 35 additions & 12 deletions datafusion-executor/src/expression.rs
Original file line number Diff line number Diff line change
Expand Up @@ -390,13 +390,11 @@ fn struct_columns_from_patch(
/// types come from `output_type`, which must be a struct holding only primitive fields (matching
/// the kernel evaluator, which supports only primitive targets).
///
/// Each field extracts its value with `cast(get_field(map, name), T)`. For a numeric or temporal
/// type the raw value is first wrapped in `nullif(.., '')`, mapping an empty string to null before
/// the cast, so an empty string becomes null (kernel's `empty_string_partition_cast`) while an
/// unparseable value fails the cast (kernel's hard parse error). String and Binary keep the raw
/// value (empty is a valid empty string / empty bytes). A missing key or null value is already null
/// via [`get_field`]. The whole struct is nulled where the input map row is null, via `<map> IS NOT
/// NULL`.
/// Each field extracts its value with `cast(get_field(map, name), T)`. For every type except String
/// and Binary, the raw value is first wrapped in `nullif(.., '')`, mapping an empty string to null
/// before the cast. String and Binary keep the raw value because empty strings and bytes are valid.
/// A missing key or null value is already null via [`get_field`]. The whole struct is nulled where
/// the input map row is null, via `<map> IS NOT NULL`.
///
/// KNOWN DIVERGENCES from the kernel parser, confined to malformed or non-spec-compliant values
/// (spec-compliant writers never emit them):
Expand All @@ -408,14 +406,20 @@ fn struct_columns_from_patch(
/// value's scale to match the target's exactly (and hard-errors otherwise).
///
/// # Errors
Comment thread
DrakeLin marked this conversation as resolved.
/// Returns an error when `output_type` is absent, not a struct, or has a non-primitive field, or
/// from lowering the map expression.
///
/// Returns an error when options are configured, `output_type` is absent, not a struct, or has a
/// non-primitive field, or from lowering the map expression.
fn map_to_struct_to_df_expr(
map_to_struct: &MapToStructExpression,
input_schema: &StructType,
output_type: Option<&KernelDataType>,
) -> DeltaResult<DFExpr> {
let target = require_struct_output(output_type, "MapToStruct")?;
if !can_lower_map_to_struct_options(map_to_struct) {
return Err(Error::unsupported(
"DataFusion execution of MapToStruct with configured options",
));
}
let map = to_df_expr(&map_to_struct.map_expr, input_schema, None)?;

let mut args = Vec::with_capacity(target.num_fields() * 2);
Expand Down Expand Up @@ -445,6 +449,10 @@ fn map_to_struct_to_df_expr(
Ok(struct_null_when_not(map.is_not_null(), named_struct(args)))
}

fn can_lower_map_to_struct_options(map_to_struct: &MapToStructExpression) -> bool {
Comment thread
DrakeLin marked this conversation as resolved.
Comment thread
DrakeLin marked this conversation as resolved.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit1 can_lower_map_to_struct_options here and can_visit_map_to_struct in ffi/src/expressions/engine_visitor.rs are one-line wrappers whose whole body is map_to_struct.options.is_default(), restating the same "adapter supports only default options" concept under two names in two crates. MapToStructOptions::is_default() is already public for exactly this cross-crate use.

Raised by: architecture-reviewer
Suggested fix: call map_to_struct.options.is_default() directly at each guard site, or express "needs engine support beyond default" as one method on MapToStructExpression that both adapters call.

map_to_struct.options.is_default()
}

/// Lowers a `ParseJson` (parse a JSON-string column into a struct) to a call of the
/// [`ParseJsonUdf`] scalar UDF, which delegates to kernel's own JSON parser. Unlike the
/// struct-shaped arms, `ParseJson` is self-typed -- it carries its target `output_schema` -- so it
Expand Down Expand Up @@ -547,7 +555,7 @@ mod tests {
use datafusion::physical_expr::execution_props::ExecutionProps;
use delta_kernel::expressions::{
col, lit, null_lit, ColumnName as KernelColumnName, Expression as KernelExpr,
ExpressionStructPatch, ExpressionStructPatchBuilder,
ExpressionStructPatch, ExpressionStructPatchBuilder, MapToStructOptions,
};
use delta_kernel::schema::{schema, schema_ref, ArrayType, DataType, MapType, StructType};
use rstest::rstest;
Expand Down Expand Up @@ -1001,7 +1009,7 @@ mod tests {
/// Lowers a `MapToStruct` over `pv` targeting `output_schema` and renders it as a `Display`
/// string.
fn lower_map_to_struct(output_schema: StructType) -> String {
let kernel = KernelExpr::map_to_struct(col!("pv"));
let kernel = KernelExpr::map_to_struct(col!("pv"), MapToStructOptions::default());
let target: DataType = output_schema.into();
to_df_expr(&kernel, &pv_map_schema(), Some(&target))
.unwrap()
Expand Down Expand Up @@ -1063,13 +1071,28 @@ mod tests {
#[case] output_type: Option<DataType>,
#[case] expected_message: &str,
) {
let kernel = KernelExpr::map_to_struct(col!("pv"));
let kernel = KernelExpr::map_to_struct(col!("pv"), MapToStructOptions::default());
let err = to_df_expr(&kernel, &pv_map_schema(), output_type.as_ref())
.unwrap_err()
.to_string();
assert!(err.contains(expected_message), "{err}");
}

#[test]
fn configured_map_to_struct_is_unsupported() {
let target = DataType::from(schema! { nullable "ts": TIMESTAMP });
let kernel = KernelExpr::map_to_struct(
col!("pv"),
MapToStructOptions::default().with_timestamp_timezone("America/Los_Angeles"),
);

let error = to_df_expr(&kernel, &pv_map_schema(), Some(&target))
.unwrap_err()
.to_string();

assert!(error.contains("MapToStruct with configured options"));
}

// === ParseJson Shared Helpers ===

/// Input schema for JSON tests: `{ j: string }`.
Expand Down
101 changes: 94 additions & 7 deletions ffi/src/expressions/engine_visitor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -174,9 +174,9 @@ pub struct EngineExpressionVisitor {
/// `child_list_id`. The `output_schema` handle specifies the schema to parse the JSON
/// into.
pub visit_parse_json: VisitParseJsonFn,
/// Visits the `MapToStruct` expression belonging to the list identified by `sibling_list_id`.
/// The sub-expression (map column) will be in a _one_ item list identified by `child_list_id`.
/// The output struct schema is determined by the evaluator's result type.
/// Visits a `MapToStruct` expression with default options. Expressions with configured options
/// are reported through `visit_unknown` without visiting the child expression. The
/// sub-expression is in the one-item list identified by `child_list_id`.
pub visit_map_to_struct: VisitUnaryFn,
/// Visits the `LessThan` binary operator belonging to the list identified by
/// `sibling_list_id`. The operands will be in a _two_ item list identified by
Expand Down Expand Up @@ -696,7 +696,10 @@ fn visit_expression_impl(
schema_handle
);
}
Expression::MapToStruct(MapToStructExpression { map_expr }) => {
Expression::MapToStruct(map_to_struct) if !can_visit_map_to_struct(map_to_struct) => {
visit_unknown(visitor, sibling_list_id, "configured_map_to_struct")
Comment thread
DrakeLin marked this conversation as resolved.
}
Expression::MapToStruct(MapToStructExpression { map_expr, .. }) => {
let child_list_id = call!(visitor, make_field_list, 1);
visit_expression_impl(visitor, map_expr, child_list_id);
call!(visitor, visit_map_to_struct, sibling_list_id, child_list_id);
Expand All @@ -711,6 +714,10 @@ fn visit_expression_impl(
}
}

fn can_visit_map_to_struct(map_to_struct: &MapToStructExpression) -> bool {
map_to_struct.options.is_default()
}

fn visit_predicate_impl(
visitor: &mut EngineExpressionVisitor,
predicate: &Predicate,
Expand Down Expand Up @@ -771,7 +778,7 @@ fn visit_predicate_internal(predicate: &Predicate, visitor: &mut EngineExpressio

#[cfg(test)]
mod tests {
use delta_kernel::expressions::{lit, Expression, Scalar};
use delta_kernel::expressions::{lit, Expression, MapToStructOptions, Scalar};
use rstest::rstest;

use super::*;
Expand All @@ -791,6 +798,14 @@ mod tests {
sibling_list_id: usize,
parts: Vec<String>,
},
Unknown {
sibling_list_id: usize,
name: String,
},
MapToStruct {
sibling_list_id: usize,
child_list_id: usize,
},
}

#[derive(Default)]
Expand Down Expand Up @@ -847,6 +862,30 @@ mod tests {
});
}

extern "C" fn visit_unknown_name(
data: *mut c_void,
sibling_list_id: usize,
name: KernelStringSlice,
) {
let builder = unsafe { &mut *(data as *mut TestExpressionBuilder) };
let name = unsafe { String::try_from_slice(&name) }.unwrap();
builder.events.push(LiteralEvent::Unknown {
sibling_list_id,
name,
});
}

extern "C" fn visit_map_to_struct(
data: *mut c_void,
sibling_list_id: usize,
child_list_id: usize,
) {
let builder = unsafe { &mut *(data as *mut TestExpressionBuilder) };
builder.events.push(LiteralEvent::MapToStruct {
sibling_list_id,
child_list_id,
});
}
macro_rules! ignore_fn {
($fn_name:ident $(, $arg_type:ty)*) => {
extern "C" fn $fn_name(
Expand Down Expand Up @@ -907,7 +946,7 @@ mod tests {
visit_is_null: ignore_child_list,
visit_to_json: ignore_child_list,
visit_parse_json: ignore_parse_json,
visit_map_to_struct: ignore_child_list,
visit_map_to_struct,
visit_lt: ignore_child_list,
visit_gt: ignore_child_list,
visit_eq: ignore_child_list,
Expand All @@ -925,7 +964,7 @@ mod tests {
visit_field_patch: ignore_field_patch,
visit_opaque_expr: ignore_opaque_expr,
visit_opaque_pred: ignore_opaque_pred,
visit_unknown: ignore_string_slice,
visit_unknown: visit_unknown_name,
}
}

Expand Down Expand Up @@ -992,4 +1031,52 @@ mod tests {
assert_eq!(top_level_id, 0);
assert_eq!(builder.events, vec![expected]);
}

#[test]
fn timezone_aware_map_to_struct_visits_unknown() {
let expression = Expression::map_to_struct(
Expression::column(["partitionValues"]),
MapToStructOptions::default().with_timestamp_timezone("America/Los_Angeles"),
);
let mut builder = TestExpressionBuilder::default();
let mut visitor = test_visitor(&mut builder);

let top_level_id = visit_expression_internal(&expression, &mut visitor);

assert_eq!(top_level_id, 0);
assert_eq!(
builder.events,
vec![LiteralEvent::Unknown {
sibling_list_id: 0,
name: "configured_map_to_struct".to_string(),
}]
);
}

#[test]
fn default_map_to_struct_visits_child_then_map_to_struct() {
let expression = Expression::map_to_struct(
Expression::column(["partitionValues"]),
MapToStructOptions::default(),
);
let mut builder = TestExpressionBuilder::default();
let mut visitor = test_visitor(&mut builder);

let top_level_id = visit_expression_internal(&expression, &mut visitor);

assert_eq!(top_level_id, 0);
assert_eq!(
builder.events,
vec![
LiteralEvent::Column {
sibling_list_id: 1,
parts: vec!["partitionValues".to_string()],
},
LiteralEvent::MapToStruct {
sibling_list_id: 0,
child_list_id: 1,
},
]
);
}
}
7 changes: 5 additions & 2 deletions ffi/src/expressions/kernel_visitor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ use std::sync::Arc;
use delta_kernel::engine::arrow_expression::opaque::ArrowOpaquePredicate;
use delta_kernel::expressions::{
lit, null_lit, BinaryExpressionOp, BinaryPredicateOp, ColumnName, Expression,
JunctionPredicateOp, Predicate, Scalar, UnaryPredicateOp,
JunctionPredicateOp, MapToStructOptions, Predicate, Scalar, UnaryPredicateOp,
};
use delta_kernel::schema::{DataType, PrimitiveType};
use delta_kernel::DeltaResult;
Expand Down Expand Up @@ -692,7 +692,10 @@ pub extern "C" fn visit_expression_map_to_struct(
child_expr: usize,
) -> usize {
unwrap_kernel_expression(state, child_expr).map_or(0, |expr| {
wrap_expression(state, Expression::map_to_struct(expr))
wrap_expression(
state,
Expression::map_to_struct(expr, MapToStructOptions::default()),
)
})
}

Expand Down
9 changes: 5 additions & 4 deletions ffi/src/test_ffi.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,9 @@ use std::sync::Arc;

use delta_kernel::expressions::{
col, column_name, column_pred, lit, null_lit, ArrayData, BinaryExpressionOp, BinaryPredicateOp,
Expression as Expr, ExpressionStructPatchBuilder, MapData, OpaqueExpressionOp,
OpaquePredicateOp, Predicate as Pred, Scalar, ScalarExpressionEvaluator, StructData,
Expression as Expr, ExpressionStructPatchBuilder, MapData, MapToStructOptions,
OpaqueExpressionOp, OpaquePredicateOp, Predicate as Pred, Scalar, ScalarExpressionEvaluator,
StructData,
};
use delta_kernel::kernel_predicates::{
DirectDataSkippingPredicateEvaluator, DirectPredicateEvaluator,
Expand Down Expand Up @@ -158,7 +159,7 @@ pub unsafe extern "C" fn get_testing_kernel_expression() -> Handle<SharedExpress
Expr::struct_from([lit(5_i32), lit(20_i64)]),
Expr::opaque(OpaqueTestOp("foo".to_string()), vec![lit(42), lit(1.111)]),
Expr::unknown("mystery"),
Expr::map_to_struct(col!("pv")),
Expr::map_to_struct(col!("pv"), MapToStructOptions::default()),
Expr::coalesce([col!("col"), lit(0_i32)]),
Expr::array([lit(1_i32), lit(2_i32)]),
];
Expand Down Expand Up @@ -254,7 +255,7 @@ pub unsafe extern "C" fn get_simple_testing_kernel_expression() -> Handle<Shared
Expr::binary(BinaryExpressionOp::Multiply, lit(5), lit(6)),
Expr::binary(BinaryExpressionOp::Divide, lit(100), lit(4)),
Expr::struct_from([lit(1_i32), lit(2_i64), lit(3.0_f64)]),
Expr::map_to_struct(col!("partitionValues")),
Expr::map_to_struct(col!("partitionValues"), MapToStructOptions::default()),
];
Arc::new(Expr::struct_from(sub_exprs)).into()
}
Expand Down
3 changes: 2 additions & 1 deletion kernel/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ pre-release-hook = [
delta_kernel_derive = { path = "../derive-macros", version = "0.28.0" }
bytes = "1.10"
chrono = "0.4.41"
chrono-tz = { version = "0.10.4", optional = true }
crc = "3.2.2"
indexmap = "2.10.0"
itertools = "0.14"
Expand Down Expand Up @@ -109,7 +110,7 @@ need-arrow = [] # need-arrow is a marker that the feature needs arrow dep
arrow-58 = ["dep:arrow_58", "dep:parquet_58", "dep:object_store_13"]
arrow-59 = ["dep:arrow_59", "dep:parquet_59", "dep:object_store_13"]
arrow-conversion = ["need-arrow"]
arrow-expression = ["need-arrow"]
arrow-expression = ["need-arrow", "dep:chrono-tz"]

# Schema diffing functionality (experimental)
schema-diff = []
Expand Down
8 changes: 8 additions & 0 deletions kernel/proto/expressions.proto
Original file line number Diff line number Diff line change
Expand Up @@ -150,8 +150,16 @@ message ParseJsonExpression {
delta.kernel.schema.StructType output_schema = 2;
}

message MapToStructOptions {
// Optional connector-supplied IANA timezone or normalized fixed offset (`+HH:MM` or `-HH:MM`)
// for offset-less TIMESTAMP values. Kernel never infers this from the host or table metadata. A
// timezone or offset in a value takes precedence; absence means UTC.
optional string timestamp_timezone = 1;
}

message MapToStructExpression {
Expression map_expr = 1;
MapToStructOptions options = 2;
}

// `nullability_predicate` is optional: when set and it evaluates to false/null, the whole
Expand Down
8 changes: 6 additions & 2 deletions kernel/src/checkpoint/checkpoint_transform.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
use std::sync::{Arc, LazyLock};

use crate::actions::{ADD_NAME, STATS_PARSED as STATS_PARSED_FIELD};
use crate::expressions::{col, Expression, ExpressionRef, UnaryExpressionOp};
use crate::expressions::{col, Expression, ExpressionRef, MapToStructOptions, UnaryExpressionOp};
use crate::schema::{DataType, SchemaRef, SchemaStructPatchBuilder, StructField, StructType};
use crate::struct_patch::ProjectionStructPatchBuilder;
use crate::table_properties::TableProperties;
Expand Down Expand Up @@ -211,7 +211,11 @@ fn build_stats_parsed_expr(stats_schema: &SchemaRef) -> ExpressionRef {
fn build_partition_values_parsed_expr() -> ExpressionRef {
Arc::new(Expression::coalesce([
col!(ADD_NAME, PARTITION_VALUES_PARSED_FIELD),
Expression::map_to_struct(col!(ADD_NAME, PARTITION_VALUES_FIELD)),
// Checkpoint construction has no reader timezone, so use UTC.
Comment thread
DrakeLin marked this conversation as resolved.
Expression::map_to_struct(
Comment thread
DrakeLin marked this conversation as resolved.
col!(ADD_NAME, PARTITION_VALUES_FIELD),
MapToStructOptions::default(),
Comment thread
DrakeLin marked this conversation as resolved.
),
]))
}

Expand Down
Loading
Loading