yihua commented on code in PR #868:
URL: https://github.com/apache/hudi-rs/pull/868#discussion_r4420383724
##########
crates/core/src/write/mod.rs:
##########
@@ -109,6 +109,154 @@ pub(crate) fn align_batch_to_schema(
arrow_array::RecordBatch::try_new(target.clone(),
batch.columns().to_vec()).ok()
}
+/// The write's batches in the shape the rest of the write path compares
exactly;
+/// see [`normalize_write_batch`].
+///
+/// Call after the timeline reload, so the schema is the table's current one.
+/// The schema is read only when a batch carries a list column, which is the
+/// only shape this changes.
+pub(crate) async fn normalize_for_write(
Review Comment:
Fixed in 8ed81801: list items are now named `element` on every write,
including the first, so the files never mix item names, and every test read now
goes through both readers and asserts one schema across batches.
##########
crates/core/src/write/rewrite/mod.rs:
##########
@@ -889,6 +894,7 @@ pub async fn dynamic_partition_overwrite_batches(
) -> Result<WriteResult> {
crate::write::ensure_supported_key_generator(&table.hudi_configs)?;
table.reload_timeline_for_write().await?;
+ let batches = &crate::write::normalize_for_write(table, batches).await?;
Review Comment:
Done, added `test_dynamic_partition_overwrite_accepts_list_columns`.
##########
crates/core/tests/table_write_list_columns_tests.rs:
##########
@@ -0,0 +1,350 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+//! Writes carrying list columns, the shape embedding vectors arrive in.
+
+use std::sync::Arc;
+
+use arrow::array::{
+ Array, ArrayRef, FixedSizeListArray, Float32Array, Int64Array, ListArray,
StringArray,
+};
+use arrow::datatypes::{DataType, Field, Schema};
+use arrow::record_batch::RecordBatch;
+use hudi_core::config::table::TableTypeValue;
+use hudi_core::table::{ReadOptions, Table};
+use tempfile::tempdir;
+
+fn embedding_values(rows: &[(&str, f32)]) -> Float32Array {
+ Float32Array::from(
+ rows.iter()
+ .flat_map(|(_, v)| [*v, *v + 1.0])
+ .collect::<Vec<_>>(),
+ )
+}
+
+/// `id`, `embedding: list<item: float32>`, `v`, with the item field named the
way
+/// PyArrow names it.
+fn list_batch(rows: &[(&str, f32)], version: i64) -> RecordBatch {
+ let item = Arc::new(Field::new("item", DataType::Float32, true));
+ let offsets =
arrow::buffer::OffsetBuffer::from_lengths(rows.iter().map(|_| 2));
+ let embedding = ListArray::new(
+ item.clone(),
+ offsets,
+ Arc::new(embedding_values(rows)),
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("embedding", DataType::List(item), true),
+ Field::new("v", DataType::Int64, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(
+ rows.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
+ )),
+ Arc::new(embedding) as ArrayRef,
+ Arc::new(Int64Array::from(vec![version; rows.len()])),
+ ],
+ )
+ .unwrap()
+}
+
+/// The same rows with `embedding: fixed_size_list<item: float32>[2]`.
+fn fixed_batch(rows: &[(&str, f32)], version: i64) -> RecordBatch {
+ let item = Arc::new(Field::new("item", DataType::Float32, true));
+ let embedding =
+ FixedSizeListArray::new(item.clone(), 2,
Arc::new(embedding_values(rows)), None);
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("embedding", DataType::FixedSizeList(item, 2), true),
+ Field::new("v", DataType::Int64, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(
+ rows.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
+ )),
+ Arc::new(embedding) as ArrayRef,
+ Arc::new(Int64Array::from(vec![version; rows.len()])),
+ ],
+ )
+ .unwrap()
+}
+
+/// `(id, first embedding value)` for every row, sorted by id.
+async fn read_embeddings(table: &Table) -> Vec<(String, f32)> {
+ let batches = table.read(&ReadOptions::new()).await.unwrap();
+ let mut rows = Vec::new();
+ for batch in &batches {
+ let ids = batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let embeddings = batch
+ .column_by_name("embedding")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .expect("embedding reads back as a variable-size list");
+ for i in 0..batch.num_rows() {
+ let values = embeddings.value(i);
+ let first = values
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap()
+ .value(0);
+ rows.push((ids.value(i).to_string(), first));
+ }
+ }
+ rows.sort_by(|a, b| a.0.cmp(&b.0));
+ rows
+}
+
+#[tokio::test]
+async fn
test_upsert_accepts_list_column_whose_item_name_differs_from_table_schema() {
+ let dir = tempdir().unwrap();
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("chunks")
+ .with_record_key_fields(["id"])
+ .with_ordering_fields(["v"])
+ .create()
+ .await
+ .unwrap();
+ table
+ .append([list_batch(&[("a", 1.0), ("b", 2.0)], 1)])
+ .await
+ .unwrap();
+ // The table schema now comes from Avro and names the item `element`; the
+ // incoming batch still names it `item`.
+ let result = table
+ .upsert([list_batch(&[("b", 20.0), ("c", 3.0)], 2)])
+ .await
+ .unwrap();
+ assert_eq!((result.num_updates, result.num_inserts), (1, 1));
+ assert_eq!(
+ read_embeddings(&table).await,
+ vec![("a".into(), 1.0), ("b".into(), 20.0), ("c".into(), 3.0)]
+ );
+
+ let result = table.delete("id = 'a'").await.unwrap();
+ assert_eq!(result.num_deletes, 1);
+ assert_eq!(
+ read_embeddings(&table).await,
+ vec![("b".into(), 20.0), ("c".into(), 3.0)]
+ );
+}
+
+#[tokio::test]
+async fn test_fixed_size_list_column_is_written_as_list() {
+ let dir = tempdir().unwrap();
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("chunks")
+ .with_record_key_fields(["id"])
+ .with_ordering_fields(["v"])
+ .create()
+ .await
+ .unwrap();
+ table
+ .append([fixed_batch(&[("a", 1.0), ("b", 2.0)], 1)])
+ .await
+ .unwrap();
+ let schema = table.get_schema().await.unwrap();
+ assert!(matches!(
+ schema.field_with_name("embedding").unwrap().data_type(),
+ DataType::List(_)
+ ));
+ let result = table
+ .upsert([fixed_batch(&[("b", 20.0)], 2)])
+ .await
+ .unwrap();
+ assert_eq!(result.num_updates, 1);
+ assert_eq!(
+ read_embeddings(&table).await,
+ vec![("a".into(), 1.0), ("b".into(), 20.0)]
+ );
+}
+
+async fn new_table(dir: &tempfile::TempDir, table_type: TableTypeValue) ->
Table {
+ Table::create(dir.path().to_str().unwrap())
+ .with_table_name("chunks")
+ .with_table_type(table_type)
+ .with_record_key_fields(["id"])
+ .with_ordering_fields(["v"])
+ .create()
+ .await
+ .unwrap()
+}
+
+#[tokio::test]
+async fn test_update_and_multi_batch_upsert_accept_list_item_name_mismatch() {
+ let dir = tempdir().unwrap();
+ let mut table = new_table(&dir, TableTypeValue::CopyOnWrite).await;
+ table
+ .append([list_batch(&[("a", 1.0), ("b", 2.0)], 1)])
+ .await
+ .unwrap();
+ // Two batches in one write, both named `item` against a table that says
`element`.
+ let result = table
+ .upsert([list_batch(&[("a", 10.0)], 2), fixed_batch(&[("c", 3.0)], 2)])
+ .await
+ .unwrap();
+ assert_eq!((result.num_updates, result.num_inserts), (1, 1));
+ // SET on the list column itself through a one-row batch named `item`.
+ let set = list_batch(&[("ignored", 7.0)], 3).project(&[1]).unwrap();
+ let result = table.update("id = 'b'", set).await.unwrap();
+ assert_eq!(result.num_updates, 1);
+ assert_eq!(
+ read_embeddings(&table).await,
+ vec![("a".into(), 10.0), ("b".into(), 7.0), ("c".into(), 3.0)]
+ );
+}
+
+#[tokio::test]
+async fn test_merge_on_read_upsert_accepts_list_column() {
+ let dir = tempdir().unwrap();
+ let mut table = new_table(&dir, TableTypeValue::MergeOnRead).await;
+ table
+ .append([list_batch(&[("a", 1.0), ("b", 2.0)], 1)])
+ .await
+ .unwrap();
+ let result = table
+ .upsert([fixed_batch(&[("b", 20.0), ("c", 3.0)], 2)])
+ .await
+ .unwrap();
+ assert_eq!((result.num_updates, result.num_inserts), (1, 1));
+ assert_eq!(
+ read_embeddings(&table).await,
+ vec![("a".into(), 1.0), ("b".into(), 20.0), ("c".into(), 3.0)]
+ );
+}
+
+#[tokio::test]
+async fn
test_nested_fixed_size_list_is_widened_and_struct_child_names_stay_strict() {
+ use arrow::array::StructArray;
+ let dir = tempdir().unwrap();
+ let mut table = new_table(&dir, TableTypeValue::CopyOnWrite).await;
+ // `payload: struct<vec: fixed_size_list<float32>[2]>`
+ let item = Arc::new(Field::new("item", DataType::Float32, true));
+ let vec_field = Arc::new(Field::new(
+ "vec",
+ DataType::FixedSizeList(item.clone(), 2),
+ true,
+ ));
+ let payload = StructArray::from(vec![(
+ vec_field.clone(),
+ Arc::new(FixedSizeListArray::new(
+ item,
+ 2,
+ Arc::new(Float32Array::from(vec![1.0, 2.0])),
+ None,
+ )) as ArrayRef,
+ )]);
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("payload", DataType::Struct(vec![vec_field].into()), true),
+ Field::new("v", DataType::Int64, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(vec!["a"])),
+ Arc::new(payload) as ArrayRef,
+ Arc::new(Int64Array::from(vec![1])),
+ ],
+ )
+ .unwrap();
+ table.append([batch]).await.unwrap();
+ let stored = table.get_schema().await.unwrap();
+ let DataType::Struct(children) =
stored.field_with_name("payload").unwrap().data_type() else {
+ panic!("payload must stay a struct");
+ };
+ assert!(matches!(children[0].data_type(), DataType::List(_)));
+
+ // Same shape but a differently named struct child: not the same column.
+ let other = Arc::new(Field::new(
+ "other",
+ DataType::List(Arc::new(Field::new("item", DataType::Float32, true))),
+ true,
+ ));
+ let payload = StructArray::from(vec![(
+ other.clone(),
+ Arc::new(ListArray::new(
+ Arc::new(Field::new("item", DataType::Float32, true)),
+ arrow::buffer::OffsetBuffer::from_lengths([2]),
+ Arc::new(Float32Array::from(vec![3.0, 4.0])),
+ None,
+ )) as ArrayRef,
+ )]);
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("payload", DataType::Struct(vec![other].into()), true),
+ Field::new("v", DataType::Int64, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(vec!["a"])),
+ Arc::new(payload) as ArrayRef,
+ Arc::new(Int64Array::from(vec![2])),
+ ],
+ )
+ .unwrap();
+ let err = table.upsert([batch]).await.unwrap_err().to_string();
+ assert!(err.contains("does not match"), "{err}");
Review Comment:
Done, it now matches `CoreError::Schema` and still checks the message.
##########
crates/core/src/write/mod.rs:
##########
@@ -109,6 +109,154 @@ pub(crate) fn align_batch_to_schema(
arrow_array::RecordBatch::try_new(target.clone(),
batch.columns().to_vec()).ok()
}
+/// The write's batches in the shape the rest of the write path compares
exactly;
+/// see [`normalize_write_batch`].
+///
+/// Call after the timeline reload, so the schema is the table's current one.
+/// The schema is read only when a batch carries a list column, which is the
+/// only shape this changes.
+pub(crate) async fn normalize_for_write(
+ table: &crate::table::Table,
+ batches: &[arrow_array::RecordBatch],
+) -> Result<Vec<arrow_array::RecordBatch>> {
+ let has_list = batches.iter().any(|b| {
+ b.schema()
+ .fields()
+ .iter()
+ .any(|f| contains_list(f.data_type()))
+ });
+ let table_schema = if has_list
+ && table
+ .timeline
+ .get_latest_commit_timestamp_as_option()
+ .is_some()
+ {
+ Some(table.get_schema().await?)
+ } else {
+ None
+ };
+ batches
+ .iter()
+ .cloned()
+ .map(|batch| normalize_write_batch(batch, table_schema.as_ref()))
+ .collect()
+}
+
+fn contains_list(dt: &arrow_schema::DataType) -> bool {
+ use arrow_schema::DataType;
+ match dt {
+ DataType::List(_) | DataType::LargeList(_) |
DataType::FixedSizeList(_, _) => true,
+ DataType::Struct(fields) => fields.iter().any(|f|
contains_list(f.data_type())),
+ DataType::Map(entries, _) => contains_list(entries.data_type()),
+ _ => false,
Review Comment:
Addressed: `contains_list` is gone along with the table schema read.
##########
crates/core/src/write/mod.rs:
##########
@@ -109,6 +109,154 @@ pub(crate) fn align_batch_to_schema(
arrow_array::RecordBatch::try_new(target.clone(),
batch.columns().to_vec()).ok()
}
+/// The write's batches in the shape the rest of the write path compares
exactly;
+/// see [`normalize_write_batch`].
+///
+/// Call after the timeline reload, so the schema is the table's current one.
+/// The schema is read only when a batch carries a list column, which is the
+/// only shape this changes.
+pub(crate) async fn normalize_for_write(
+ table: &crate::table::Table,
+ batches: &[arrow_array::RecordBatch],
+) -> Result<Vec<arrow_array::RecordBatch>> {
+ let has_list = batches.iter().any(|b| {
+ b.schema()
+ .fields()
+ .iter()
+ .any(|f| contains_list(f.data_type()))
+ });
+ let table_schema = if has_list
+ && table
+ .timeline
+ .get_latest_commit_timestamp_as_option()
+ .is_some()
+ {
+ Some(table.get_schema().await?)
+ } else {
+ None
+ };
+ batches
+ .iter()
+ .cloned()
+ .map(|batch| normalize_write_batch(batch, table_schema.as_ref()))
+ .collect()
+}
+
+fn contains_list(dt: &arrow_schema::DataType) -> bool {
+ use arrow_schema::DataType;
+ match dt {
+ DataType::List(_) | DataType::LargeList(_) |
DataType::FixedSizeList(_, _) => true,
+ DataType::Struct(fields) => fields.iter().any(|f|
contains_list(f.data_type())),
+ DataType::Map(entries, _) => contains_list(entries.data_type()),
+ _ => false,
+ }
+}
+
+/// Bring a batch to the shape the rest of the write path compares exactly.
+///
+/// Fixed-size lists become lists at every nesting level, since Avro has only
+/// `array` and the readers hand lists back as `List`. Then, when the table
+/// already has a schema, a column whose type matches the table's up to list
+/// item names is cast to the table's type: a list read back from Avro names
+/// its item `element` while PyArrow names it `item`, and the cast only
+/// relabels. Columns that differ in any other way are left as they are, so
+/// the schema checks downstream reject them with their usual messages.
+fn normalize_write_batch(
+ batch: arrow_array::RecordBatch,
+ table_schema: Option<&arrow_schema::Schema>,
+) -> Result<arrow_array::RecordBatch> {
+ use arrow_schema::Field;
+ let mut changed = false;
+ let mut fields = Vec::with_capacity(batch.num_columns());
+ let mut columns = Vec::with_capacity(batch.num_columns());
+ for (field, column) in batch.schema().fields().iter().zip(batch.columns())
{
+ let widened = widen_fixed_size_lists(field.data_type());
+ let target = match table_schema.and_then(|s|
s.field_with_name(field.name()).ok()) {
+ Some(table_field) if same_up_to_list_item_names(&widened,
table_field.data_type()) => {
+ table_field.data_type().clone()
+ }
+ _ => widened,
+ };
+ if &target == field.data_type() {
+ fields.push(field.clone());
+ columns.push(column.clone());
+ continue;
+ }
+ changed = true;
+ let casted = arrow::compute::cast(column.as_ref(),
&target).map_err(|e| {
+ CoreError::Schema(format!(
+ "cannot convert column '{}' from {} to {target} for writing:
{e}",
+ field.name(),
+ field.data_type()
+ ))
+ })?;
+ fields.push(std::sync::Arc::new(
+ Field::new(field.name(), target, field.is_nullable())
+ .with_metadata(field.metadata().clone()),
+ ));
+ columns.push(casted);
+ }
+ if !changed {
+ return Ok(batch);
+ }
+ let schema = std::sync::Arc::new(arrow_schema::Schema::new_with_metadata(
+ fields,
+ batch.schema().metadata().clone(),
+ ));
+ arrow_array::RecordBatch::try_new(schema, columns)
+ .map_err(|e| CoreError::Schema(format!("cannot rebuild batch for
writing: {e}")))
+}
+
+/// `dt` with every `FixedSizeList`, at any depth, replaced by a `List` of the
+/// same item.
+fn widen_fixed_size_lists(dt: &arrow_schema::DataType) ->
arrow_schema::DataType {
+ use arrow_schema::{DataType, Field};
+ let widen_field = |f: &std::sync::Arc<Field>| {
+ std::sync::Arc::new(
+ Field::new(
+ f.name(),
+ widen_fixed_size_lists(f.data_type()),
+ f.is_nullable(),
+ )
+ .with_metadata(f.metadata().clone()),
+ )
+ };
+ match dt {
+ DataType::FixedSizeList(item, _) | DataType::List(item) => {
+ DataType::List(widen_field(item))
+ }
+ DataType::LargeList(item) => DataType::LargeList(widen_field(item)),
+ DataType::Struct(fields) =>
DataType::Struct(fields.iter().map(widen_field).collect()),
+ DataType::Map(entries, sorted) => DataType::Map(widen_field(entries),
*sorted),
+ other => other.clone(),
Review Comment:
Done, `normalize_list_type` (the renamed function) and
`normalize_write_batch` now have unit tests.
##########
crates/core/src/write/mod.rs:
##########
@@ -109,6 +109,154 @@ pub(crate) fn align_batch_to_schema(
arrow_array::RecordBatch::try_new(target.clone(),
batch.columns().to_vec()).ok()
}
+/// The write's batches in the shape the rest of the write path compares
exactly;
+/// see [`normalize_write_batch`].
+///
+/// Call after the timeline reload, so the schema is the table's current one.
+/// The schema is read only when a batch carries a list column, which is the
+/// only shape this changes.
+pub(crate) async fn normalize_for_write(
+ table: &crate::table::Table,
+ batches: &[arrow_array::RecordBatch],
+) -> Result<Vec<arrow_array::RecordBatch>> {
+ let has_list = batches.iter().any(|b| {
+ b.schema()
+ .fields()
+ .iter()
+ .any(|f| contains_list(f.data_type()))
+ });
+ let table_schema = if has_list
+ && table
+ .timeline
+ .get_latest_commit_timestamp_as_option()
+ .is_some()
+ {
+ Some(table.get_schema().await?)
+ } else {
+ None
+ };
+ batches
+ .iter()
+ .cloned()
+ .map(|batch| normalize_write_batch(batch, table_schema.as_ref()))
+ .collect()
+}
+
+fn contains_list(dt: &arrow_schema::DataType) -> bool {
+ use arrow_schema::DataType;
+ match dt {
+ DataType::List(_) | DataType::LargeList(_) |
DataType::FixedSizeList(_, _) => true,
+ DataType::Struct(fields) => fields.iter().any(|f|
contains_list(f.data_type())),
+ DataType::Map(entries, _) => contains_list(entries.data_type()),
+ _ => false,
+ }
+}
+
+/// Bring a batch to the shape the rest of the write path compares exactly.
+///
+/// Fixed-size lists become lists at every nesting level, since Avro has only
+/// `array` and the readers hand lists back as `List`. Then, when the table
+/// already has a schema, a column whose type matches the table's up to list
+/// item names is cast to the table's type: a list read back from Avro names
+/// its item `element` while PyArrow names it `item`, and the cast only
+/// relabels. Columns that differ in any other way are left as they are, so
+/// the schema checks downstream reject them with their usual messages.
+fn normalize_write_batch(
+ batch: arrow_array::RecordBatch,
+ table_schema: Option<&arrow_schema::Schema>,
+) -> Result<arrow_array::RecordBatch> {
+ use arrow_schema::Field;
+ let mut changed = false;
+ let mut fields = Vec::with_capacity(batch.num_columns());
+ let mut columns = Vec::with_capacity(batch.num_columns());
+ for (field, column) in batch.schema().fields().iter().zip(batch.columns())
{
+ let widened = widen_fixed_size_lists(field.data_type());
+ let target = match table_schema.and_then(|s|
s.field_with_name(field.name()).ok()) {
+ Some(table_field) if same_up_to_list_item_names(&widened,
table_field.data_type()) => {
+ table_field.data_type().clone()
+ }
+ _ => widened,
+ };
+ if &target == field.data_type() {
+ fields.push(field.clone());
+ columns.push(column.clone());
+ continue;
+ }
+ changed = true;
+ let casted = arrow::compute::cast(column.as_ref(),
&target).map_err(|e| {
+ CoreError::Schema(format!(
+ "cannot convert column '{}' from {} to {target} for writing:
{e}",
+ field.name(),
+ field.data_type()
+ ))
+ })?;
+ fields.push(std::sync::Arc::new(
+ Field::new(field.name(), target, field.is_nullable())
+ .with_metadata(field.metadata().clone()),
+ ));
+ columns.push(casted);
+ }
+ if !changed {
+ return Ok(batch);
+ }
+ let schema = std::sync::Arc::new(arrow_schema::Schema::new_with_metadata(
+ fields,
+ batch.schema().metadata().clone(),
+ ));
+ arrow_array::RecordBatch::try_new(schema, columns)
+ .map_err(|e| CoreError::Schema(format!("cannot rebuild batch for
writing: {e}")))
+}
+
+/// `dt` with every `FixedSizeList`, at any depth, replaced by a `List` of the
+/// same item.
+fn widen_fixed_size_lists(dt: &arrow_schema::DataType) ->
arrow_schema::DataType {
+ use arrow_schema::{DataType, Field};
+ let widen_field = |f: &std::sync::Arc<Field>| {
+ std::sync::Arc::new(
+ Field::new(
+ f.name(),
+ widen_fixed_size_lists(f.data_type()),
+ f.is_nullable(),
+ )
+ .with_metadata(f.metadata().clone()),
+ )
+ };
+ match dt {
+ DataType::FixedSizeList(item, _) | DataType::List(item) => {
+ DataType::List(widen_field(item))
+ }
+ DataType::LargeList(item) => DataType::LargeList(widen_field(item)),
+ DataType::Struct(fields) =>
DataType::Struct(fields.iter().map(widen_field).collect()),
+ DataType::Map(entries, sorted) => DataType::Map(widen_field(entries),
*sorted),
+ other => other.clone(),
+ }
+}
+
+/// Whether two types are equal except for the names of list item fields.
+///
+/// Struct children must match by name as well as type: a positional cast
+/// between differently named children would silently reinterpret data.
+fn same_up_to_list_item_names(a: &arrow_schema::DataType, b:
&arrow_schema::DataType) -> bool {
+ use arrow_schema::DataType;
+ let same_item = |x: &arrow_schema::Field, y: &arrow_schema::Field| {
+ x.is_nullable() == y.is_nullable()
+ && same_up_to_list_item_names(x.data_type(), y.data_type())
+ };
+ match (a, b) {
+ (DataType::List(x), DataType::List(y))
+ | (DataType::LargeList(x), DataType::LargeList(y)) => same_item(x, y),
+ (DataType::Struct(xs), DataType::Struct(ys)) => {
+ xs.len() == ys.len()
+ && xs
+ .iter()
+ .zip(ys.iter())
+ .all(|(x, y)| x.name() == y.name() && same_item(x, y))
+ }
+ (DataType::Map(x, xs), DataType::Map(y, ys)) => xs == ys &&
same_item(x, y),
+ _ => a == b,
Review Comment:
Addressed: `same_up_to_list_item_names` is gone, since writes no longer
compare against the table schema.
--
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]
