andygrove commented on code in PR #7196:
URL: https://github.com/apache/datafusion-comet/pull/7196#discussion_r4370668876
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -739,17 +675,29 @@ fn check_conversion(
match (physical_type, target_type) {
(DataType::Struct(physical_fields), DataType::Struct(target_fields))
=> {
let physical_indices = match_struct_fields(physical_fields,
target_fields, options)?;
-
combine_checks(target_fields.iter().zip(physical_indices).filter_map(
- |(target_field, physical_index)| {
- let physical_field = &physical_fields[physical_index?];
- Some(check_conversion(
- physical_field.data_type(),
- target_field.data_type(),
- &format!("{column}, {}", physical_field.name()),
- options,
- ))
- },
- ))
+ combine_checks(
+ target_fields
+ .iter()
+ .zip(physical_indices.iter())
+ .filter_map(|(target_field, (physical_index, owns_value))|
{
+ let physical_field = &physical_fields[physical_index?];
+ Some(
+ check_conversion(
+ physical_field.data_type(),
+ target_field.data_type(),
+ &format!("{column}, {}",
physical_field.name()),
+ options,
+ )
+ .map(|check| {
+ if owns_value {
+ check
+ } else {
+ ConversionCheck::Accept
+ }
+ }),
Review Comment:
Masking a losing leaf's deferred rejection changes the answer from Spark's
default reader here. With `s struct<x id1 bigint, a int>` over a file `s
struct<A id1 bigint>`, Spark's vectorized reader fails with
`SchemaColumnConvertNotSupportedException: column: [s, A], physicalType: INT64,
logicalType: int` in either field order. That's because
`VectorizedColumnReader.readBatch` calls `getUpdater` for every requested leaf,
including the one whose pages the other reader already drained. Main raises the
same exception through `RejectOnNonEmpty`. With this change Comet returns `(7,
null)`, which is parquet-mr's answer. An empty file still reads in all three.
Comet otherwise follows the vectorized reader's conversion checks and
doesn't replicate parquet-mr's null-on-narrowing. So could the losing leaf keep
its deferred rejection here? It would still read NULL whenever Spark can read
the column as its type. That would flip the `Int32` cases at depth 1 and 2 in
`conflicting_field_claims_read_null_at_every_depth`, and the leaf case in
`losing_field_claims_preserve_open_errors_but_not_deferred_rejections`. The
root has the same gap on main, which I filed as #7207.
##########
native/core/src/parquet/field_resolver.rs:
##########
@@ -204,6 +241,55 @@ impl<'a> FieldResolver<'a> {
)),
}
}
+
+ /// Resolve claims in file order, with id matches ahead of shadows and
name matches.
+ /// Repeated requested ids keep the last claimant, as Spark parquet-mr
does. Duplicate file ids are
+ /// checked across the whole requested group, including unprojected root
fields (#7006).
+ /// Name ambiguity is left to the caller: the root checks referenced
columns only, whereas
+ /// nested conversion validates every requested field in
[`Self::resolve_all`].
+ pub(crate) fn file_claims(&self) -> DataFusionResult<Vec<FieldClaim>> {
+ let mut claims = vec![FieldClaim::Unchanged; self.file.len()];
+ for requested in 0..self.requested.len() {
+ match self.select(requested)? {
+ Selection::ById(matched) => {
+ if let Some(file) = matched {
+ claims[file].claim(FieldClaim::IdMatch(requested));
+ }
+ for &file in self.name_matches(requested) {
+ claims[file].claim(FieldClaim::Shadowed);
+ }
+ }
+ Selection::ByName(matches) => {
+ for &file in matches {
+ claims[file].claim(FieldClaim::NameMatch(requested));
+ }
+ }
+ }
+ }
+ Ok(claims)
+ }
+
+ /// Resolve a nested group in requested order, giving each file field to
its strongest
+ /// claimant only. Missing and losing fields read NULL. Validate
independent selections
+ /// first, in requested order, so arbitration cannot hide ambiguity or
change which of
+ /// several invalid selections reports its error first.
+ pub(crate) fn resolve_all(&self) -> DataFusionResult<Vec<Option<usize>>> {
+ let mut resolved = (0..self.requested.len())
+ .map(|requested| self.resolve(requested))
+ .collect::<DataFusionResult<Vec<_>>>()?;
+ let claims = self.file_claims()?;
Review Comment:
I ran `parquet_field_resolution` against main and this head, alternating the
two binaries six times, with LTO off and 16 codegen units for both. At 64 rows
per batch the head is still slower than main. `convert/64` averaged 15.20 µs
against 14.47 µs, between 3% and 6% slower in every pair.
`decode_and_convert/64` averaged 15.95 ms against 15.61 ms, between 1% and 3%
slower. At 8192 rows the difference was within noise. So the gain in your reply
looks like it's measured against the earlier head rather than main.
The work that's new on every batch is the claim vector and a name lookup for
each id-bearing field. Could a group without competing claims skip that, so it
costs what it did before?
##########
native/core/src/parquet/field_resolver.rs:
##########
@@ -47,6 +47,43 @@ pub(crate) enum Selection<'a> {
ByName(&'a [usize]),
}
+/// The strongest claim on one file field. Root remapping and nested
conversion give a file
+/// field to this claimant only. Repeated id claims keep the last requested
field, matching
+/// Spark parquet-mr's `ParquetRowConverter` id-to-index `toMap`. Equal name
claims keep the
+/// earliest requested field, retaining the root remap's name-matching
behavior.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum FieldClaim {
+ /// The requested field selects this file field by Parquet field id.
+ IdMatch(usize),
+ /// An id-bearing requested field has this file field's folded name. It
must not reach this
+ /// field by name when its id is missing or selects a different field.
Only an id match can
+ /// override this claim. The root hides it with a placeholder; nested
fields read NULL.
+ Shadowed,
+ /// The requested field selects this file field by folded name.
+ NameMatch(usize),
+ /// No requested field claims this file field. The root keeps its original
name.
+ Unchanged,
+}
+
+impl FieldClaim {
+ fn strength(self) -> u8 {
+ match self {
+ Self::IdMatch(_) => 3,
+ Self::Shadowed => 2,
+ Self::NameMatch(_) => 1,
+ Self::Unchanged => 0,
+ }
+ }
+
+ fn claim(&mut self, claim: Self) {
+ if claim.strength() > self.strength()
+ || matches!((*self, claim), (Self::IdMatch(_), Self::IdMatch(_)))
Review Comment:
A follow-up makes sense to me, and I filed #7206 for it. The name-match
version of this shape is still wrong after the new fallback, and on main too.
With `[x id1, a, z id2]` over a file `[A id1, z id2]`, `select("a", "z")`
returns `[7, 9]` in both Spark readers and `[null, 9]` in Comet, because the
unread `x` still takes `A` by id. No id repeats there, so the fallback can't
see it. Limiting claims to the required roots, as in your PoC, should fix both
shapes, and then the fallback could go.
--
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]
