linliu-code commented on code in PR #711:
URL: https://github.com/apache/hudi-rs/pull/711#discussion_r3928690950
##########
python/src/datafusion_internal.rs:
##########
@@ -40,23 +40,26 @@ fn extract_codec(session: Bound<PyAny>) ->
PyResult<FFI_LogicalExtensionCodec> {
} else {
session
};
- let capsule = capsule_obj.downcast::<PyCapsule>()?;
- if let Some(name) = capsule.name()? {
- let name = name
- .to_str()
- .map_err(|e| PyValueError::new_err(format!("{e}")))?;
- if name != "datafusion_logical_extension_codec" {
- return Err(PyValueError::new_err(format!(
- "Expected PyCapsule name 'datafusion_logical_extension_codec',
got '{name}'"
- )));
- }
- }
- let codec = unsafe { capsule.reference::<FFI_LogicalExtensionCodec>() };
+ let capsule = capsule_obj.cast::<PyCapsule>()?;
+ // `pointer_checked` is the name check and the pointer read in one step: it
+ // refuses any capsule not carrying exactly this name, which is what makes
+ // the cast below sound. An unnamed capsule no longer passes, where the
+ // previous name-then-read pair let one through.
+ let codec = capsule
+ .pointer_checked(Some(c"datafusion_logical_extension_codec"))
+ .map_err(|e| {
+ PyValueError::new_err(format!(
+ "Expected a PyCapsule named
'datafusion_logical_extension_codec': {e}"
+ ))
+ })?;
+ // SAFETY: DataFusion puts that name only on a capsule holding an
Review Comment:
The name check proves the capsule holds an `FFI_LogicalExtensionCodec`, but
not one of a matching major version — the capsule name is not versioned, and
`datafusion-ffi = "54.1"` is a caret req while the `datafusion==54.0.0` pin
lives only in an optional extra. A user who installs a datafusion-python built
on ffi 55 into the same env hands over the same capsule name with a different
struct layout, and the cast then reads function pointers at the wrong offsets.
`FFI_LogicalExtensionCodec` carries a `version` fn returning the major version
for exactly this handshake — worth calling it, or softening the comment to say
the version match is assumed? (Pre-existing, not introduced here.)
##########
crates/datafusion/tests/plan_tests.rs:
##########
@@ -148,26 +148,37 @@ async fn verify_plan(
plan.contains("SortExec: TopK(fetch=10)"),
"Plan should contain TopK sort"
);
+ // The projection, struct field access included, is pushed into the Parquet
+ // source rather than planned as its own `ProjectionExec`, so it is
asserted
+ // wherever it lands. The index in `structField@N` is the field's position
in
+ // the file schema and differs per table, hence the two-part match.
assert!(
- plan.contains(&format!(
- "ProjectionExec: expr=[id@0 as id, name@1 as name, isActive@2 as
isActive, \
- get_field(structField@3, field2) as
{table_name}.structField[field2]]"
- )),
- "Plan should contain expected projection"
+ plan.contains("get_field(structField@")
Review Comment:
This drops `id, name, isActive` from the assertion, so a projection-pushdown
regression that read every file column would still pass here — `verify_data`
only sees the query's output columns, not what the scan read, and
`get_field(structField@` also appears inside `predicate=`. Since only the `@N`
index varies per table, could the first half keep the anchor,
`plan.contains("projection=[id, name, isActive, get_field(structField@")`, with
the existing tail match unchanged?
Same thought on the `", field2) > 30"` fragment just below (and on line
403): anchoring it to `predicate=` would stop it also matching inside
`pruning_predicate=`.
##########
crates/datafusion/tests/plan_tests.rs:
##########
@@ -148,26 +148,37 @@ async fn verify_plan(
plan.contains("SortExec: TopK(fetch=10)"),
"Plan should contain TopK sort"
);
+ // The projection, struct field access included, is pushed into the Parquet
+ // source rather than planned as its own `ProjectionExec`, so it is
asserted
+ // wherever it lands. The index in `structField@N` is the field's position
in
+ // the file schema and differs per table, hence the two-part match.
assert!(
- plan.contains(&format!(
- "ProjectionExec: expr=[id@0 as id, name@1 as name, isActive@2 as
isActive, \
- get_field(structField@3, field2) as
{table_name}.structField[field2]]"
- )),
- "Plan should contain expected projection"
+ plan.contains("get_field(structField@")
+ && plan.contains(&format!(", field2) as
{table_name}.structField[field2]")),
+ "Plan should project the struct field"
);
- // With pushdown_filters enabled, simple predicates (id % 2 = 0, name !=
Alice)
- // are pushed into the Parquet source. Only non-pushable predicates like
- // struct field access remain in FilterExec.
+ // Simple predicates (id % 2 = 0, name != Alice) and the struct field
access
+ // alike are pushed into the Parquet source.
assert!(
- plan.contains("get_field(structField@3, field2) > 30"),
- "Plan should contain struct field filter (either in FilterExec or
DataSourceExec)"
+ plan.contains(", field2) > 30"),
+ "Plan should contain struct field filter"
);
assert!(
-
plan.contains(&format!("input_partitions={planned_input_partitioned}")),
- "Plan should contain expected
input_partitions={planned_input_partitioned}"
+ plan.contains(&format!(
+ "file_groups={{{planned_input_partitioned} {}:",
+ file_group_noun(*planned_input_partitioned)
+ )),
+ "Plan should scan {planned_input_partitioned} file group(s)"
);
}
+/// How the plan spells the file-group count: `hoodie.read.input.partitions`
+/// decides how many groups the scan is split into, and DataFusion renders one
+/// group in the singular.
+fn file_group_noun(count: i32) -> &'static str {
Review Comment:
nit: this helper can go — `"file_groups={2 group"` is already a prefix of
`"file_groups={2 groups:"`, so
`plan.contains(&format!("file_groups={{{planned_input_partitioned} group"))`
covers both the singular and plural rendering on its own.
--
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]
