comphead commented on code in PR #5654:
URL: https://github.com/apache/datafusion-comet/pull/5654#discussion_r4099367757
##########
native/core/src/parquet/cast_column.rs:
##########
@@ -270,12 +284,22 @@ impl PhysicalExpr for CometCastColumnExpr {
let input_physical_field = self.input_physical_field.data_type();
let target_field = self.target_field.data_type();
+ // Relabeling only swaps metadata, so it is right when every requested
field reads
+ // the file field at its own position. A mapping that reorders fields
(ids resolved
+ // to other positions) has to go through the nested conversion below.
+ let positional = self
Review Comment:
This gate is the whole #6192 fix, and it does not need a `FieldMapping`.
Checking ids in the positional walk that `types_differ_only_in_field_names`
already does is enough, with `use_field_id` read from `parquet_options`:
```rust
// Struct arm, per (pf, lf) pair
&& (!use_field_id || field_id(lf).is_none() || field_id(lf) == field_id(pf))
```
Separately, `positional` and `types_differ_only_in_field_names` depend only
on fields fixed at construction, yet both walk the type tree on every batch.
One `bool` computed in `with_parquet_options` would cover both. And since
`field_mapping` is always set together with `parquet_options`, the two could be
one `Option`.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -345,14 +356,37 @@ fn remap_physical_schema(
.with_metadata(field.metadata().clone()),
);
}
+ return Arc::clone(field);
}
}
+ // Shield: any remaining physical field whose name would hit an
ID-bearing
Review Comment:
I don't think this ordering is reachable from Spark. It only matters when an
id-bearing and an id-less requested field fold to the same name, and
`DataSource.resolveRelation` rejects that read schema with
`COLUMN_ALREADY_EXISTS` (`checkSchemaColumnNameDuplication`, in 3.5.9, 4.0.4
and 4.1.3). In case-sensitive mode the shield only fires on identical names,
which Spark rejects in any mode. By my reading, the case-sensitive Kappa read
already returns `(NULL, 7)` on `main`. I would keep `main`'s order and drop the
case-insensitive Kappa and placeholder tests.
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -162,43 +169,330 @@ impl SparkParquetOptions {
/// Spark-compatible cast implementation. Defers to DataFusion's cast where
that is known
/// to be compatible, and returns an error when a not supported and not
DF-compatible cast
-/// is requested.
+/// is requested. Resolves the nested field mapping for this one value; a
per-file caller
+/// resolves once and uses [`spark_parquet_convert_with_mapping`] for every
batch.
pub fn spark_parquet_convert(
arg: ColumnarValue,
data_type: &DataType,
parquet_options: &SparkParquetOptions,
+) -> DataFusionResult<ColumnarValue> {
+ let mapping =
+ resolve_field_mapping(&arg.data_type(), data_type,
parquet_options).map_err(spark_error)?;
+ spark_parquet_convert_with_mapping(arg, data_type, &mapping,
parquet_options)
+}
+
+/// [`spark_parquet_convert`] with a mapping already resolved for the value's
type.
+pub(crate) fn spark_parquet_convert_with_mapping(
+ arg: ColumnarValue,
+ data_type: &DataType,
+ mapping: &FieldMapping,
+ parquet_options: &SparkParquetOptions,
) -> DataFusionResult<ColumnarValue> {
match arg {
- ColumnarValue::Array(array) =>
Ok(ColumnarValue::Array(parquet_convert_array(
+ ColumnarValue::Array(array) => Ok(ColumnarValue::Array(convert_array(
array,
data_type,
+ mapping,
parquet_options,
+ None,
)?)),
ColumnarValue::Scalar(scalar) => {
// Note that normally CAST(scalar) should be fold in Spark JVM
side. However, for
// some cases e.g., scalar subquery, Spark will not fold it, so we
need to handle it
// here.
let array = scalar.to_array()?;
let scalar = ScalarValue::try_from_array(
- &parquet_convert_array(array, data_type, parquet_options)?,
+ &convert_array(array, data_type, mapping, parquet_options,
None)?,
0,
)?;
Ok(ColumnarValue::Scalar(scalar))
}
}
}
-fn parquet_convert_array(
- array: ArrayRef,
+/// Wrap a [`SparkError`] the way every native operator surfaces it to the JVM.
+pub(crate) fn spark_error(error: SparkError) -> DataFusionError {
Review Comment:
`impl From<SparkError> for DataFusionError` in `native/common/src/error.rs`
already does this, so the call sites can use `.into()`. In the same vein,
`field_id` becomes `pub(crate)` here while `schema_adapter.rs` keeps its
identical `parse_field_id`. One of the two could go.
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -162,43 +169,330 @@ impl SparkParquetOptions {
/// Spark-compatible cast implementation. Defers to DataFusion's cast where
that is known
/// to be compatible, and returns an error when a not supported and not
DF-compatible cast
-/// is requested.
+/// is requested. Resolves the nested field mapping for this one value; a
per-file caller
+/// resolves once and uses [`spark_parquet_convert_with_mapping`] for every
batch.
pub fn spark_parquet_convert(
arg: ColumnarValue,
data_type: &DataType,
parquet_options: &SparkParquetOptions,
+) -> DataFusionResult<ColumnarValue> {
+ let mapping =
+ resolve_field_mapping(&arg.data_type(), data_type,
parquet_options).map_err(spark_error)?;
+ spark_parquet_convert_with_mapping(arg, data_type, &mapping,
parquet_options)
+}
+
+/// [`spark_parquet_convert`] with a mapping already resolved for the value's
type.
+pub(crate) fn spark_parquet_convert_with_mapping(
+ arg: ColumnarValue,
+ data_type: &DataType,
+ mapping: &FieldMapping,
+ parquet_options: &SparkParquetOptions,
) -> DataFusionResult<ColumnarValue> {
match arg {
- ColumnarValue::Array(array) =>
Ok(ColumnarValue::Array(parquet_convert_array(
+ ColumnarValue::Array(array) => Ok(ColumnarValue::Array(convert_array(
array,
data_type,
+ mapping,
parquet_options,
+ None,
)?)),
ColumnarValue::Scalar(scalar) => {
// Note that normally CAST(scalar) should be fold in Spark JVM
side. However, for
// some cases e.g., scalar subquery, Spark will not fold it, so we
need to handle it
// here.
let array = scalar.to_array()?;
let scalar = ScalarValue::try_from_array(
- &parquet_convert_array(array, data_type, parquet_options)?,
+ &convert_array(array, data_type, mapping, parquet_options,
None)?,
0,
)?;
Ok(ColumnarValue::Scalar(scalar))
}
}
}
-fn parquet_convert_array(
- array: ArrayRef,
+/// Wrap a [`SparkError`] the way every native operator surfaces it to the JVM.
+pub(crate) fn spark_error(error: SparkError) -> DataFusionError {
+ DataFusionError::External(Box::new(error))
+}
+
+/// Outcome of matching one requested id or name against a struct's file
fields: the last
+/// file field that matched and whether more than one did. A plain `Copy`
value, so resolving
+/// a wide struct allocates nothing per id or per name; the matched names are
only gathered
+/// when an ambiguity is reported.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct FieldMatch {
Review Comment:
The later index that `also` records is never read, because both ambiguous
branches return an error. An `Option<usize>` entry does the same in one pass
without this struct, `record_field_match` or
`field_match_records_ambiguity_without_allocating`. It is also all the nested
duplicate-id fix needs in `match_struct_fields` on `main`:
```rust
// None marks an id that more than one file field carries
map.entry(id).and_modify(|m| *m = None).or_insert(Some(i));
// ...
(true, Some(id)) => match from_id_to_index.get(&id) {
Some(None) => Err(/* DuplicateFieldByFieldId, matched fields as "[x, y]"
*/),
index => Ok(index.copied().flatten()),
},
```
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -162,43 +169,330 @@ impl SparkParquetOptions {
/// Spark-compatible cast implementation. Defers to DataFusion's cast where
that is known
/// to be compatible, and returns an error when a not supported and not
DF-compatible cast
-/// is requested.
+/// is requested. Resolves the nested field mapping for this one value; a
per-file caller
+/// resolves once and uses [`spark_parquet_convert_with_mapping`] for every
batch.
pub fn spark_parquet_convert(
arg: ColumnarValue,
data_type: &DataType,
parquet_options: &SparkParquetOptions,
+) -> DataFusionResult<ColumnarValue> {
+ let mapping =
+ resolve_field_mapping(&arg.data_type(), data_type,
parquet_options).map_err(spark_error)?;
+ spark_parquet_convert_with_mapping(arg, data_type, &mapping,
parquet_options)
+}
+
+/// [`spark_parquet_convert`] with a mapping already resolved for the value's
type.
+pub(crate) fn spark_parquet_convert_with_mapping(
+ arg: ColumnarValue,
+ data_type: &DataType,
+ mapping: &FieldMapping,
+ parquet_options: &SparkParquetOptions,
) -> DataFusionResult<ColumnarValue> {
match arg {
- ColumnarValue::Array(array) =>
Ok(ColumnarValue::Array(parquet_convert_array(
+ ColumnarValue::Array(array) => Ok(ColumnarValue::Array(convert_array(
array,
data_type,
+ mapping,
parquet_options,
+ None,
)?)),
ColumnarValue::Scalar(scalar) => {
// Note that normally CAST(scalar) should be fold in Spark JVM
side. However, for
// some cases e.g., scalar subquery, Spark will not fold it, so we
need to handle it
// here.
let array = scalar.to_array()?;
let scalar = ScalarValue::try_from_array(
- &parquet_convert_array(array, data_type, parquet_options)?,
+ &convert_array(array, data_type, mapping, parquet_options,
None)?,
0,
)?;
Ok(ColumnarValue::Scalar(scalar))
}
}
}
-fn parquet_convert_array(
- array: ArrayRef,
+/// Wrap a [`SparkError`] the way every native operator surfaces it to the JVM.
+pub(crate) fn spark_error(error: SparkError) -> DataFusionError {
+ DataFusionError::External(Box::new(error))
+}
+
+/// Outcome of matching one requested id or name against a struct's file
fields: the last
+/// file field that matched and whether more than one did. A plain `Copy`
value, so resolving
+/// a wide struct allocates nothing per id or per name; the matched names are
only gathered
+/// when an ambiguity is reported.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct FieldMatch {
+ pub(crate) index: usize,
+ pub(crate) ambiguous: bool,
+}
+
+impl FieldMatch {
+ pub(crate) fn new(index: usize, ambiguous: bool) -> Self {
+ Self { index, ambiguous }
+ }
+
+ /// The first file field carrying this id or name.
+ pub(crate) fn first(index: usize) -> Self {
+ Self::new(index, false)
+ }
+
+ /// A further file field carrying the same id or name: the later index
wins, as Spark's
+ /// `toMap` does for exact names, and the entry turns ambiguous.
+ pub(crate) fn also(self, index: usize) -> Self {
+ Self::new(index, true)
+ }
+}
+
+/// Record file field `index` under `key`, keeping the entry `Copy`-sized
however many fields
+/// share the key.
+pub(crate) fn record_field_match<K: Hash + Eq>(
+ matches: &mut HashMap<K, FieldMatch>,
+ key: K,
+ index: usize,
+) {
+ matches
+ .entry(key)
+ .and_modify(|m| *m = m.also(index))
+ .or_insert_with(|| FieldMatch::first(index));
+}
+
+/// Names of the fields carrying `id`, for the duplicate-id error message.
Bracketed and
+/// comma-joined the way Spark's `matchIdField` renders the list, so the
message reads
+/// `Found duplicate field(s) "1": [x, y] in id mapping mode` on both sides.
+pub(crate) fn field_names_with_id(fields: &Fields, id: i32) -> String {
+ let names = fields
+ .iter()
+ .filter(|f| field_id(f) == Some(id))
+ .map(|f| f.name().as_str())
+ .collect::<Vec<_>>()
+ .join(", ");
+ format!("[{names}]")
+}
+
+/// Which file field supplies each requested field, resolved once per file and
reused for
+/// every batch. Follows the requested type as Spark's `clipParquetSchema`
does: a struct
+/// lists one source per requested field, a list in any Arrow representation
or a map carries
+/// the mapping of its element or key and value types, and anything else is a
leaf converted
+/// by type.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub(crate) enum FieldMapping {
+ Struct(Vec<StructFieldSource>),
+ List(Box<FieldMapping>),
+ Map(Box<FieldMapping>, Box<FieldMapping>),
+ Leaf,
+}
+
+/// The file field behind one requested struct field.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub(crate) struct StructFieldSource {
+ /// Index of the file field supplying the requested field; `None`
null-fills it.
+ pub(crate) from_index: Option<usize>,
+ /// Mapping of the requested field's own type.
+ pub(crate) nested: FieldMapping,
+}
+
+impl FieldMapping {
+ /// Mapping of a list's element type. A `Leaf` list converts its elements
by type alone,
+ /// as the adapter hands one to every column whose type holds no struct.
+ pub(crate) fn list_element(&self) -> DataFusionResult<&FieldMapping> {
+ match self {
+ FieldMapping::List(inner) => Ok(inner),
+ FieldMapping::Leaf => Ok(&FieldMapping::Leaf),
+ other => Err(DataFusionError::Internal(format!(
+ "list column resolved to a non-list field mapping: {other:?}"
+ ))),
+ }
+ }
+
+ /// Mappings of a map's key and value types; see
[`FieldMapping::list_element`].
+ pub(crate) fn map_entries(&self) -> DataFusionResult<(&FieldMapping,
&FieldMapping)> {
+ match self {
+ FieldMapping::Map(key, value) => Ok((key, value)),
+ FieldMapping::Leaf => Ok((&FieldMapping::Leaf,
&FieldMapping::Leaf)),
+ other => Err(DataFusionError::Internal(format!(
+ "map column resolved to a non-map field mapping: {other:?}"
+ ))),
+ }
+ }
+
+ /// True when every requested field reads the file field at its own
position, so a
+ /// metadata-only relabel of the file array already yields the requested
layout.
+ pub(crate) fn is_positional(&self) -> bool {
+ match self {
+ FieldMapping::Struct(sources) => sources
+ .iter()
+ .enumerate()
+ .all(|(i, s)| s.from_index == Some(i) &&
s.nested.is_positional()),
+ FieldMapping::List(inner) => inner.is_positional(),
+ FieldMapping::Map(key, value) => key.is_positional() &&
value.is_positional(),
+ FieldMapping::Leaf => true,
+ }
+ }
+}
+
+/// The element field of a list in any Arrow representation. One place decides
which types
+/// are lists, so the mapping resolver, the struct-holding walk in the schema
adapter, and
+/// [`convert_array`] agree.
+pub(crate) fn list_element_field(data_type: &DataType) -> Option<&FieldRef> {
+ match data_type {
+ DataType::List(f)
+ | DataType::LargeList(f)
+ | DataType::FixedSizeList(f, _)
+ | DataType::ListView(f)
+ | DataType::LargeListView(f) => Some(f),
+ _ => None,
+ }
+}
+
+/// Resolve how `to_type` reads from `from_type`, recursing through struct,
list, and map
+/// types. Raises the ambiguity Spark reports from `clipParquetGroupFields`
when a requested
+/// id or case-insensitive name matches more than one file field at any level.
+pub(crate) fn resolve_field_mapping(
+ from_type: &DataType,
to_type: &DataType,
parquet_options: &SparkParquetOptions,
-) -> DataFusionResult<ArrayRef> {
- parquet_convert_array_impl(array, to_type, parquet_options, None)
+) -> Result<FieldMapping, SparkError> {
+ use DataType::*;
+ // Dictionary encoding is a physical detail: resolve against the value
type it wraps.
+ // Parquet dictionary encoding only wraps a leaf type, so this mirrors the
adapter's
+ // conversion check and never changes which mapping is built.
+ if let Dictionary(_, value_type) = from_type {
+ return resolve_field_mapping(value_type, to_type, parquet_options);
+ }
+ // The element mapping is the same whichever list representation either
side uses.
+ if let (Some(from_item), Some(to_item)) =
+ (list_element_field(from_type), list_element_field(to_type))
+ {
+ return Ok(FieldMapping::List(Box::new(resolve_field_mapping(
+ from_item.data_type(),
+ to_item.data_type(),
+ parquet_options,
+ )?)));
+ }
+ match (from_type, to_type) {
+ (Struct(from_fields), Struct(to_fields)) => {
+ resolve_struct_mapping(from_fields, to_fields, parquet_options)
+ }
+ (Map(from_entries, from_ordered), Map(to_entries, to_ordered))
+ if from_ordered == to_ordered =>
+ {
+ match (from_entries.data_type(), to_entries.data_type()) {
+ (Struct(from_kv), Struct(to_kv)) if from_kv.len() == 2 &&
to_kv.len() == 2 => {
+ let key = resolve_field_mapping(
+ from_kv[0].data_type(),
+ to_kv[0].data_type(),
+ parquet_options,
+ )?;
+ let value = resolve_field_mapping(
+ from_kv[1].data_type(),
+ to_kv[1].data_type(),
+ parquet_options,
+ )?;
+ Ok(FieldMapping::Map(Box::new(key), Box::new(value)))
+ }
+ _ => Ok(FieldMapping::Leaf),
+ }
+ }
+ _ => Ok(FieldMapping::Leaf),
+ }
}
-fn parquet_convert_array_impl(
+/// Match `to` (requested) struct fields to `from` (file) fields. Mirrors
Spark's
+/// `clipParquetGroupFields`: when the requested struct carries Parquet field
ids anywhere,
+/// id-bearing requested fields match only by id and the rest by name;
otherwise every field
+/// matches by name.
+fn resolve_struct_mapping(
+ from_fields: &Fields,
+ to_fields: &Fields,
+ parquet_options: &SparkParquetOptions,
+) -> Result<FieldMapping, SparkError> {
+ let should_match_by_id =
+ parquet_options.use_field_id && to_fields.iter().any(|f|
field_id(f).is_some());
+
+ let mut id_matches: HashMap<i32, FieldMatch> = HashMap::new();
+ if should_match_by_id {
+ for (i, field) in from_fields.iter().enumerate() {
+ if let Some(id) = field_id(field) {
+ record_field_match(&mut id_matches, id, i);
+ }
+ }
+ }
+
+ // Fold the file and requested names once via the same
`toLowerCase(Locale.ROOT)` the
+ // top-level schema adapter uses, so nested case-insensitive matching
agrees with it.
+ let mut all_names: Vec<&str> = Vec::with_capacity(from_fields.len() +
to_fields.len());
+ all_names.extend(from_fields.iter().map(|f| f.name().as_str()));
+ all_names.extend(to_fields.iter().map(|f| f.name().as_str()));
+ let all_folded = fold_names(&all_names, parquet_options.case_sensitive)
+ .map_err(|e| SparkError::Internal(e.to_string()))?;
+ let (from_folded, to_folded) = all_folded.split_at(from_fields.len());
+
+ let mut name_matches: HashMap<&str, FieldMatch> = HashMap::new();
+ for (i, folded) in from_folded.iter().enumerate() {
+ record_field_match(&mut name_matches, folded.as_str(), i);
+ }
+
+ let mut sources = Vec::with_capacity(to_fields.len());
+ for (to_pos, to_field) in to_fields.iter().enumerate() {
+ let from_index = match (should_match_by_id, field_id(to_field)) {
+ // A missing id match is a missing column, never a name match.
+ (true, Some(id)) => match id_matches.get(&id) {
+ Some(m) if m.ambiguous => {
+ return Err(SparkError::DuplicateFieldByFieldId {
+ required_id: id,
+ matched_fields: field_names_with_id(from_fields, id),
+ });
+ }
+ Some(m) => Some(m.index),
+ None => None,
+ },
+ _ => match name_matches.get(to_folded[to_pos].as_str()) {
+ // Spark's `matchCaseInsensitiveField` raises
`_LEGACY_ERROR_TEMP_2093` for a
+ // requested name that folds onto more than one file field,
whether the siblings
+ // differ by case or are byte-identical. In case-sensitive
mode the fold is
+ // identity, so a collision means byte-identical siblings.
Spark's
+ // `matchCaseSensitiveField` builds its map with `toMap` there
and the last field
+ // wins silently. Comet refuses instead of picking one, with
the error every
+ // other path raises for a duplicate the decoder cannot
represent.
+ Some(m) if m.ambiguous => {
+ let matched: Vec<&str> = from_folded
+ .iter()
+ .zip(from_fields.iter())
+ .filter(|(folded, _)| *folded == &to_folded[to_pos])
+ .map(|(_, f)| f.name().as_str())
+ .collect();
+ if parquet_options.case_sensitive {
+ return
Err(SparkError::Internal(duplicate_parquet_field_message(
Review Comment:
Returning `SparkError` from the resolver, so errors can be cached per field,
changes two error classes. Here, the case-sensitive nested duplicate becomes
`SparkError::Internal`, which `ShimSparkErrorConverter` turns into
`SparkException.internalError`, Spark's `INTERNAL_ERROR` (SQLSTATE `XX000`).
Spark reserves that for engine bugs, and `main` raises an execution error with
the same text. At L426, a JNI failure while folding names is flattened into a
string, so the Java exception that #5845 made propagate is lost. The tests only
match message substrings, so neither shows up. Keeping `DataFusionResult` and
not caching errors avoids both.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -3518,11 +3856,187 @@ mod test {
let physical = Arc::new(Schema::new(vec![
Field::new("a", DataType::Int32, true).with_metadata(id_meta("9"))
]));
- let (remapped, _name_map) =
+ let (remapped, _name_map, _) =
super::remap_physical_schema(&logical, &physical, true, true,
false).unwrap();
assert_eq!(remapped.field(0).name(), "a");
}
+ /// Build a nullable Int64 field carrying a Parquet field ID.
+ fn field_with_id(name: &str, id: i32) -> Field {
+ Field::new(name, DataType::Int64,
true).with_metadata(id_meta(&id.to_string()))
+ }
+
+ /// Write a Parquet file from `file_schema`/`columns`, then scan it with
+ /// `required_schema` through the Spark expression adapter and return the
first batch.
+ async fn scan_with_adapter(
+ file_schema: SchemaRef,
+ columns: Vec<Arc<dyn arrow::array::Array>>,
+ required_schema: SchemaRef,
+ spark_parquet_options: SparkParquetOptions,
+ ) -> Result<RecordBatch, DataFusionError> {
+ scan_with_defaults(
+ file_schema,
+ columns,
+ required_schema,
+ spark_parquet_options,
+ None,
+ )
+ .await
+ }
+
+ /// `scan_with_adapter` with column defaults for fields missing from the
file.
+ async fn scan_with_defaults(
Review Comment:
This repeats `scan_parquet` apart from the defaults and the writer
properties. Could `scan_parquet` take those as optional arguments instead?
Likewise, `field_with_id` here and in `parquet_support.rs`, and
`int_field_with_id` in `cast_column.rs`, overlap the existing `id_meta` and
`struct_type_with_field_id` helpers.
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -162,43 +169,330 @@ impl SparkParquetOptions {
/// Spark-compatible cast implementation. Defers to DataFusion's cast where
that is known
/// to be compatible, and returns an error when a not supported and not
DF-compatible cast
-/// is requested.
+/// is requested. Resolves the nested field mapping for this one value; a
per-file caller
+/// resolves once and uses [`spark_parquet_convert_with_mapping`] for every
batch.
pub fn spark_parquet_convert(
arg: ColumnarValue,
data_type: &DataType,
parquet_options: &SparkParquetOptions,
+) -> DataFusionResult<ColumnarValue> {
+ let mapping =
+ resolve_field_mapping(&arg.data_type(), data_type,
parquet_options).map_err(spark_error)?;
+ spark_parquet_convert_with_mapping(arg, data_type, &mapping,
parquet_options)
+}
+
+/// [`spark_parquet_convert`] with a mapping already resolved for the value's
type.
+pub(crate) fn spark_parquet_convert_with_mapping(
+ arg: ColumnarValue,
+ data_type: &DataType,
+ mapping: &FieldMapping,
+ parquet_options: &SparkParquetOptions,
) -> DataFusionResult<ColumnarValue> {
match arg {
- ColumnarValue::Array(array) =>
Ok(ColumnarValue::Array(parquet_convert_array(
+ ColumnarValue::Array(array) => Ok(ColumnarValue::Array(convert_array(
array,
data_type,
+ mapping,
parquet_options,
+ None,
)?)),
ColumnarValue::Scalar(scalar) => {
// Note that normally CAST(scalar) should be fold in Spark JVM
side. However, for
// some cases e.g., scalar subquery, Spark will not fold it, so we
need to handle it
// here.
let array = scalar.to_array()?;
let scalar = ScalarValue::try_from_array(
- &parquet_convert_array(array, data_type, parquet_options)?,
+ &convert_array(array, data_type, mapping, parquet_options,
None)?,
0,
)?;
Ok(ColumnarValue::Scalar(scalar))
}
}
}
-fn parquet_convert_array(
- array: ArrayRef,
+/// Wrap a [`SparkError`] the way every native operator surfaces it to the JVM.
+pub(crate) fn spark_error(error: SparkError) -> DataFusionError {
+ DataFusionError::External(Box::new(error))
+}
+
+/// Outcome of matching one requested id or name against a struct's file
fields: the last
+/// file field that matched and whether more than one did. A plain `Copy`
value, so resolving
+/// a wide struct allocates nothing per id or per name; the matched names are
only gathered
+/// when an ambiguity is reported.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct FieldMatch {
+ pub(crate) index: usize,
+ pub(crate) ambiguous: bool,
+}
+
+impl FieldMatch {
+ pub(crate) fn new(index: usize, ambiguous: bool) -> Self {
+ Self { index, ambiguous }
+ }
+
+ /// The first file field carrying this id or name.
+ pub(crate) fn first(index: usize) -> Self {
+ Self::new(index, false)
+ }
+
+ /// A further file field carrying the same id or name: the later index
wins, as Spark's
+ /// `toMap` does for exact names, and the entry turns ambiguous.
+ pub(crate) fn also(self, index: usize) -> Self {
+ Self::new(index, true)
+ }
+}
+
+/// Record file field `index` under `key`, keeping the entry `Copy`-sized
however many fields
+/// share the key.
+pub(crate) fn record_field_match<K: Hash + Eq>(
+ matches: &mut HashMap<K, FieldMatch>,
+ key: K,
+ index: usize,
+) {
+ matches
+ .entry(key)
+ .and_modify(|m| *m = m.also(index))
+ .or_insert_with(|| FieldMatch::first(index));
+}
+
+/// Names of the fields carrying `id`, for the duplicate-id error message.
Bracketed and
+/// comma-joined the way Spark's `matchIdField` renders the list, so the
message reads
+/// `Found duplicate field(s) "1": [x, y] in id mapping mode` on both sides.
+pub(crate) fn field_names_with_id(fields: &Fields, id: i32) -> String {
+ let names = fields
+ .iter()
+ .filter(|f| field_id(f) == Some(id))
+ .map(|f| f.name().as_str())
+ .collect::<Vec<_>>()
+ .join(", ");
+ format!("[{names}]")
+}
+
+/// Which file field supplies each requested field, resolved once per file and
reused for
+/// every batch. Follows the requested type as Spark's `clipParquetSchema`
does: a struct
+/// lists one source per requested field, a list in any Arrow representation
or a map carries
+/// the mapping of its element or key and value types, and anything else is a
leaf converted
+/// by type.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub(crate) enum FieldMapping {
+ Struct(Vec<StructFieldSource>),
+ List(Box<FieldMapping>),
+ Map(Box<FieldMapping>, Box<FieldMapping>),
+ Leaf,
+}
+
+/// The file field behind one requested struct field.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub(crate) struct StructFieldSource {
+ /// Index of the file field supplying the requested field; `None`
null-fills it.
+ pub(crate) from_index: Option<usize>,
+ /// Mapping of the requested field's own type.
+ pub(crate) nested: FieldMapping,
+}
+
+impl FieldMapping {
+ /// Mapping of a list's element type. A `Leaf` list converts its elements
by type alone,
+ /// as the adapter hands one to every column whose type holds no struct.
+ pub(crate) fn list_element(&self) -> DataFusionResult<&FieldMapping> {
+ match self {
+ FieldMapping::List(inner) => Ok(inner),
+ FieldMapping::Leaf => Ok(&FieldMapping::Leaf),
+ other => Err(DataFusionError::Internal(format!(
+ "list column resolved to a non-list field mapping: {other:?}"
+ ))),
+ }
+ }
+
+ /// Mappings of a map's key and value types; see
[`FieldMapping::list_element`].
+ pub(crate) fn map_entries(&self) -> DataFusionResult<(&FieldMapping,
&FieldMapping)> {
+ match self {
+ FieldMapping::Map(key, value) => Ok((key, value)),
+ FieldMapping::Leaf => Ok((&FieldMapping::Leaf,
&FieldMapping::Leaf)),
+ other => Err(DataFusionError::Internal(format!(
+ "map column resolved to a non-map field mapping: {other:?}"
+ ))),
+ }
+ }
+
+ /// True when every requested field reads the file field at its own
position, so a
+ /// metadata-only relabel of the file array already yields the requested
layout.
+ pub(crate) fn is_positional(&self) -> bool {
+ match self {
+ FieldMapping::Struct(sources) => sources
+ .iter()
+ .enumerate()
+ .all(|(i, s)| s.from_index == Some(i) &&
s.nested.is_positional()),
+ FieldMapping::List(inner) => inner.is_positional(),
+ FieldMapping::Map(key, value) => key.is_positional() &&
value.is_positional(),
+ FieldMapping::Leaf => true,
+ }
+ }
+}
+
+/// The element field of a list in any Arrow representation. One place decides
which types
+/// are lists, so the mapping resolver, the struct-holding walk in the schema
adapter, and
+/// [`convert_array`] agree.
+pub(crate) fn list_element_field(data_type: &DataType) -> Option<&FieldRef> {
+ match data_type {
+ DataType::List(f)
+ | DataType::LargeList(f)
+ | DataType::FixedSizeList(f, _)
+ | DataType::ListView(f)
+ | DataType::LargeListView(f) => Some(f),
+ _ => None,
+ }
+}
+
+/// Resolve how `to_type` reads from `from_type`, recursing through struct,
list, and map
+/// types. Raises the ambiguity Spark reports from `clipParquetGroupFields`
when a requested
+/// id or case-insensitive name matches more than one file field at any level.
+pub(crate) fn resolve_field_mapping(
+ from_type: &DataType,
to_type: &DataType,
parquet_options: &SparkParquetOptions,
-) -> DataFusionResult<ArrayRef> {
- parquet_convert_array_impl(array, to_type, parquet_options, None)
+) -> Result<FieldMapping, SparkError> {
+ use DataType::*;
+ // Dictionary encoding is a physical detail: resolve against the value
type it wraps.
+ // Parquet dictionary encoding only wraps a leaf type, so this mirrors the
adapter's
+ // conversion check and never changes which mapping is built.
+ if let Dictionary(_, value_type) = from_type {
+ return resolve_field_mapping(value_type, to_type, parquet_options);
+ }
+ // The element mapping is the same whichever list representation either
side uses.
+ if let (Some(from_item), Some(to_item)) =
+ (list_element_field(from_type), list_element_field(to_type))
+ {
+ return Ok(FieldMapping::List(Box::new(resolve_field_mapping(
+ from_item.data_type(),
+ to_item.data_type(),
+ parquet_options,
+ )?)));
+ }
+ match (from_type, to_type) {
+ (Struct(from_fields), Struct(to_fields)) => {
+ resolve_struct_mapping(from_fields, to_fields, parquet_options)
+ }
+ (Map(from_entries, from_ordered), Map(to_entries, to_ordered))
+ if from_ordered == to_ordered =>
+ {
+ match (from_entries.data_type(), to_entries.data_type()) {
+ (Struct(from_kv), Struct(to_kv)) if from_kv.len() == 2 &&
to_kv.len() == 2 => {
+ let key = resolve_field_mapping(
+ from_kv[0].data_type(),
+ to_kv[0].data_type(),
+ parquet_options,
+ )?;
+ let value = resolve_field_mapping(
+ from_kv[1].data_type(),
+ to_kv[1].data_type(),
+ parquet_options,
+ )?;
+ Ok(FieldMapping::Map(Box::new(key), Box::new(value)))
+ }
+ _ => Ok(FieldMapping::Leaf),
+ }
+ }
+ _ => Ok(FieldMapping::Leaf),
+ }
}
-fn parquet_convert_array_impl(
+/// Match `to` (requested) struct fields to `from` (file) fields. Mirrors
Spark's
+/// `clipParquetGroupFields`: when the requested struct carries Parquet field
ids anywhere,
+/// id-bearing requested fields match only by id and the rest by name;
otherwise every field
+/// matches by name.
+fn resolve_struct_mapping(
+ from_fields: &Fields,
+ to_fields: &Fields,
+ parquet_options: &SparkParquetOptions,
+) -> Result<FieldMapping, SparkError> {
+ let should_match_by_id =
+ parquet_options.use_field_id && to_fields.iter().any(|f|
field_id(f).is_some());
+
+ let mut id_matches: HashMap<i32, FieldMatch> = HashMap::new();
+ if should_match_by_id {
+ for (i, field) in from_fields.iter().enumerate() {
+ if let Some(id) = field_id(field) {
+ record_field_match(&mut id_matches, id, i);
+ }
+ }
+ }
+
+ // Fold the file and requested names once via the same
`toLowerCase(Locale.ROOT)` the
+ // top-level schema adapter uses, so nested case-insensitive matching
agrees with it.
+ let mut all_names: Vec<&str> = Vec::with_capacity(from_fields.len() +
to_fields.len());
+ all_names.extend(from_fields.iter().map(|f| f.name().as_str()));
+ all_names.extend(to_fields.iter().map(|f| f.name().as_str()));
+ let all_folded = fold_names(&all_names, parquet_options.case_sensitive)
+ .map_err(|e| SparkError::Internal(e.to_string()))?;
+ let (from_folded, to_folded) = all_folded.split_at(from_fields.len());
+
+ let mut name_matches: HashMap<&str, FieldMatch> = HashMap::new();
+ for (i, folded) in from_folded.iter().enumerate() {
+ record_field_match(&mut name_matches, folded.as_str(), i);
+ }
+
+ let mut sources = Vec::with_capacity(to_fields.len());
+ for (to_pos, to_field) in to_fields.iter().enumerate() {
+ let from_index = match (should_match_by_id, field_id(to_field)) {
+ // A missing id match is a missing column, never a name match.
+ (true, Some(id)) => match id_matches.get(&id) {
+ Some(m) if m.ambiguous => {
+ return Err(SparkError::DuplicateFieldByFieldId {
+ required_id: id,
+ matched_fields: field_names_with_id(from_fields, id),
+ });
+ }
+ Some(m) => Some(m.index),
+ None => None,
+ },
+ _ => match name_matches.get(to_folded[to_pos].as_str()) {
+ // Spark's `matchCaseInsensitiveField` raises
`_LEGACY_ERROR_TEMP_2093` for a
+ // requested name that folds onto more than one file field,
whether the siblings
+ // differ by case or are byte-identical. In case-sensitive
mode the fold is
+ // identity, so a collision means byte-identical siblings.
Spark's
+ // `matchCaseSensitiveField` builds its map with `toMap` there
and the last field
+ // wins silently. Comet refuses instead of picking one, with
the error every
+ // other path raises for a duplicate the decoder cannot
represent.
+ Some(m) if m.ambiguous => {
+ let matched: Vec<&str> = from_folded
+ .iter()
+ .zip(from_fields.iter())
+ .filter(|(folded, _)| *folded == &to_folded[to_pos])
+ .map(|(_, f)| f.name().as_str())
+ .collect();
+ if parquet_options.case_sensitive {
+ return
Err(SparkError::Internal(duplicate_parquet_field_message(
+ to_field.name(),
+ )));
+ }
+ return Err(SparkError::duplicate_field_case_insensitive(
+ to_field.name(),
+ &matched,
+ ));
+ }
+ Some(m) => Some(m.index),
+ None => None,
+ },
+ };
+ let nested = match from_index {
+ Some(i) => resolve_field_mapping(
+ from_fields[i].data_type(),
+ to_field.data_type(),
+ parquet_options,
+ )?,
+ None => FieldMapping::Leaf,
+ };
+ sources.push(StructFieldSource { from_index, nested });
+ }
+ Ok(FieldMapping::Struct(sources))
+}
+
+/// Convert `array` to `to_type` through its resolved `mapping`.
`parent_nulls` masks the rows
+/// hidden beneath null ancestors, so only values Spark reads are checked for
overflow.
+fn convert_array(
Review Comment:
The renames (`parquet_convert_array_impl` to `convert_array`,
`parquet_convert_struct_to_struct` to `convert_struct`, while
`parquet_convert_map_to_map` keeps its name), the `convert_list` extraction,
the `has_timestamp_unit` rewrite and the reflowed arms do not change behavior,
but they add a lot of diff. Resolving once per file does save the per-batch
`match_struct_fields` work on `main` (folded names and a `Vec` per name, though
ASCII names never reach the JVM). That saving is not measured yet. Could it be
its own PR, with a case added to
`native/core/benches/parquet_timestamp_conversion.rs`?
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -1009,10 +1083,84 @@ impl PhysicalExprAdapterFactory for
SparkPhysicalExprAdapterFactory {
id_duplicate_roots,
logical_folded,
physical_folded,
+ nested_mappings,
+ root_id_ambiguities,
}))
}
}
+/// Per logical field name, the mapping of its nested type against its
physical counterpart,
+/// or the ambiguity Spark reports for it. Only fields whose type holds a
struct are listed.
+type NestedMappings = HashMap<String, Result<Arc<FieldMapping>, SparkError>>;
+
+/// Per logical field name, the `_LEGACY_ERROR_TEMP_2094` ambiguity of a root
field whose id
+/// matches more than one physical root field. Only ambiguous fields are
listed.
+type RootIdAmbiguities = HashMap<String, SparkError>;
Review Comment:
This parallels `id_duplicate_roots`: both map a logical name to a root error
that `rewrite` raises when the column is referenced. One map would do, and
`rewrite` could then check each referenced column in one pass instead of three
loops over `col_names`.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -1009,10 +1083,84 @@ impl PhysicalExprAdapterFactory for
SparkPhysicalExprAdapterFactory {
id_duplicate_roots,
logical_folded,
physical_folded,
+ nested_mappings,
+ root_id_ambiguities,
}))
}
}
+/// Per logical field name, the mapping of its nested type against its
physical counterpart,
+/// or the ambiguity Spark reports for it. Only fields whose type holds a
struct are listed.
+type NestedMappings = HashMap<String, Result<Arc<FieldMapping>, SparkError>>;
+
+/// Per logical field name, the `_LEGACY_ERROR_TEMP_2094` ambiguity of a root
field whose id
+/// matches more than one physical root field. Only ambiguous fields are
listed.
+type RootIdAmbiguities = HashMap<String, SparkError>;
+
+fn type_holds_struct(data_type: &DataType) -> bool {
+ // The resolver and converter in `parquet_support` decide which types are
lists through
+ // the same helper, so this walk descends into exactly the lists they map.
+ if let Some(element) = list_element_field(data_type) {
+ return type_holds_struct(element.data_type());
+ }
+ match data_type {
+ DataType::Struct(_) => true,
+ DataType::Map(f, _) => type_holds_struct(f.data_type()),
+ _ => false,
+ }
+}
+
+/// Resolve the nested mapping of every logical field whose type holds a
struct and that has
+/// a physical counterpart. Returns `None` when no field qualifies, so flat
reads build
+/// nothing here. Ambiguities are kept per field rather than raised: Spark
validates only
+/// the fields a read requests, and `rewrite` sees which ones those are.
+fn resolve_nested_mappings(
Review Comment:
This resolves each mapping by exact name on the adapted schema, while the
cast picks its physical field by `col.index()` in `replace_with_spark_cast` and
by folded position in `wrap_all_type_mismatches`. The two agree only because
repeated root names are rejected first. Building the mapping where
`CometCastColumnExpr` is built, from its own `(input_field, target_field)`,
keeps it once per file, since `rewrite` runs once per file. That removes
`NestedMappings`, this function, `type_holds_struct`, `field_mapping_for`,
`check_opaque_decode`, the nested block in `rewrite`, and `Leaf` standing in
for lists and maps. Resolving eagerly is not needed for errors either. With
#6004 declining repeated requested ids, an ambiguous read always has differing
types, so it always gets a cast.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -2109,6 +2221,76 @@ abstract class ParquetReadSuite extends CometTestBase {
assert(
cause.isInstanceOf[RuntimeException] &&
cause.getMessage.contains("Found duplicate field(s)"))
+ checkDuplicateFieldIdMessage(
+ spark.read.schema(readSchema).parquet(dir.getCanonicalPath),
+ """"1": [a, rand2]""")
+ }
+ }
+ }
+
+ test("duplicate exact nested names are refused when requested and skipped
otherwise") {
Review Comment:
`CometNativeReaderSuite` already covers both halves on `main`. `duplicate
Parquet field names fail clearly - *` covers the refused read (struct, array
element and map value, at two batch sizes), and `duplicate Parquet field names
- exact-name projection works in both resolver modes` covers the unique
sibling. Those tests write the file with `named_struct`, so I think this test
and `duplicate-nested-names.parquet` can both go.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -2079,10 +2079,122 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
- // Verbatim port of Spark `ParquetFieldIdIOSuite.test("multiple id
matches")` so the shim
- // error path is exercised on both 3.x and 4.x. The stock suite is the CI
signal but it
- // requires the Spark test jars and `withAllParquetReaders`; keeping a copy
here lets us
- // iterate locally.
+ // The shape a schema evolution leaves behind: a nested column dropped and
added back under
+ // its old name gets a fresh field id, so the file holds `struct<x (id 1), y
(id 2)>` while
+ // the table reads `struct<x (id 3), y (id 2)>`. The names line up at every
position, which
+ // is exactly what the metadata-only relabel shortcut in the native cast
looks for, so without
+ // the field-mapping gate the scan hands back the old x values. Spark
returns null for x and
+ // keeps y. One test per nesting, so a wrong answer names the level that
produced it.
+ // Before Spark 4.1 the vectorized reader raises on this read below a list
or map, since its
+ // column vector rejects the placeholder field the clipped schema carries
for the unmatched
+ // id, so the comparison with Spark runs from 4.1 on. The pinned rows hold
everywhere.
+ private def checkDroppedAndReAddedFieldId(
Review Comment:
These pin #6192 well. Could the three tests become one table-driven test
over `s`, `l` and `m`, with a swapped-id row (`x (id 2), y (id 1)`) added? The
swapped case is the other shape in the issue, and right now only Rust covers it.
--
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]
