use std::collections::HashMap;
use std::sync::Arc;
use arrow::array::ArrayData;
use arrow_array::{
Array, ArrayRef, BooleanArray, FixedSizeListArray, GenericListArray, RecordBatch,
RecordBatchOptions, StructArray,
};
use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema, SchemaRef};
use lance_core::datatypes::LANCE_FIELD_ID_KEY;
use lance_core::{Error, Result};
use super::TOMBSTONE;
pub(super) fn field_id_of(field: &ArrowField) -> Option<i32> {
field
.metadata()
.get(LANCE_FIELD_ID_KEY)
.and_then(|v| v.parse::<i32>().ok())
.filter(|id| *id >= 0)
}
pub(super) fn relabel_to(column: &ArrayRef, data_type: &DataType) -> Result<ArrayRef> {
if column.data_type() == data_type {
return Ok(column.clone());
}
relabel_data(&column.to_data(), data_type).map(arrow_array::make_array)
}
fn relabel_data(data: &ArrayData, data_type: &DataType) -> Result<ArrayData> {
let children = children_of(data_type);
let child_data = match children {
Some(fields) if fields.len() == data.child_data().len() => data
.child_data()
.iter()
.zip(fields.iter())
.map(|(child, field)| relabel_data(child, field.data_type()))
.collect::<Result<Vec<_>>>()?,
_ => data.child_data().to_vec(),
};
data.clone()
.into_builder()
.data_type(data_type.clone())
.child_data(child_data)
.build()
.map_err(|e| {
Error::invalid_input(format!(
"a {} column cannot be read as {data_type}: {e}",
data.data_type()
))
})
}
pub(super) fn without_field_ids_in(data_type: &DataType) -> DataType {
let one = ArrowSchema::new(vec![ArrowField::new("", data_type.clone(), true)]);
without_field_ids(&one).field(0).data_type().clone()
}
pub(super) fn without_field_id(field: &ArrowField) -> ArrowField {
let mut field = field.clone();
field.metadata_mut().remove(LANCE_FIELD_ID_KEY);
field
}
pub(super) fn without_field_ids(schema: &ArrowSchema) -> ArrowSchema {
fn strip(field: &ArrowField) -> ArrowField {
let field = without_field_id(field);
match field.data_type() {
DataType::Struct(children) => {
let children: Vec<ArrowField> = children.iter().map(|c| strip(c)).collect();
field.with_data_type(DataType::Struct(children.into()))
}
DataType::List(element) => field
.clone()
.with_data_type(DataType::List(Arc::new(strip(element)))),
DataType::LargeList(element) => field
.clone()
.with_data_type(DataType::LargeList(Arc::new(strip(element)))),
DataType::FixedSizeList(element, size) => {
let size = *size;
field
.clone()
.with_data_type(DataType::FixedSizeList(Arc::new(strip(element)), size))
}
DataType::Map(entries, sorted) => {
let sorted = *sorted;
field
.clone()
.with_data_type(DataType::Map(Arc::new(strip(entries)), sorted))
}
_ => field,
}
}
let fields: Vec<ArrowField> = schema.fields().iter().map(|f| strip(f)).collect();
ArrowSchema::new_with_metadata(fields, schema.metadata().clone())
}
#[derive(Debug, Clone)]
enum Source {
Take(usize),
Nested(usize, Vec<Self>, DataType),
Null(DataType),
Live,
}
pub struct Plan {
target: SchemaRef,
sources: Vec<Source>,
identity: bool,
}
impl Plan {
pub(crate) fn emitting_plain_schema(mut self) -> Self {
fn strip(source: &mut Source) {
match source {
Source::Nested(_, children, data_type) => {
*data_type = without_field_ids_in(data_type);
children.iter_mut().for_each(strip);
}
Source::Null(data_type) => *data_type = without_field_ids_in(data_type),
_ => {}
}
}
self.sources.iter_mut().for_each(strip);
self.target = Arc::new(without_field_ids(&self.target));
self
}
pub(crate) fn resolve(
source: &ArrowSchema,
target: &SchemaRef,
pk_columns: &[String],
) -> Result<Self> {
let claimed = claimed_by_id(source.fields(), target.fields());
let sources = target
.fields()
.iter()
.map(|field| resolve_field(field, source.fields(), &claimed, pk_columns, true))
.collect::<Result<Vec<_>>>()?;
let identity = source.fields() == target.fields();
Ok(Self {
target: Arc::clone(target),
sources,
identity,
})
}
pub(crate) fn target(&self) -> &SchemaRef {
&self.target
}
pub(crate) fn is_identity(&self) -> bool {
self.identity
}
pub(crate) fn apply(&self, batch: &RecordBatch) -> Result<RecordBatch> {
let rows = batch.num_rows();
let columns = self
.sources
.iter()
.zip(self.target.fields())
.map(|(source, field)| take_column(source, batch.columns(), rows, field.name()))
.collect::<Result<Vec<_>>>()?;
RecordBatch::try_new_with_options(
Arc::clone(&self.target),
columns,
&RecordBatchOptions::new().with_row_count(Some(rows)),
)
.map_err(|e| Error::invalid_input(format!("reconcile a batch to the schema: {e}")))
}
}
fn resolve_field(
field: &ArrowField,
source_fields: &arrow_schema::Fields,
claimed: &[bool],
pk_columns: &[String],
top_level: bool,
) -> Result<Source> {
let name = field.name();
let by_id = field_id_of(field).and_then(|id| {
source_fields
.iter()
.position(|f| field_id_of(f) == Some(id))
});
let by_name = || {
source_fields
.iter()
.position(|f| f.name() == name)
.filter(|i| !claimed[*i])
};
let index = match field_id_of(field) {
Some(_) => by_id.or_else(|| {
source_fields
.iter()
.all(|f| field_id_of(f).is_none())
.then(by_name)
.flatten()
}),
None => by_name(),
};
let Some(index) = index else {
if top_level && name == TOMBSTONE {
return Ok(Source::Live);
}
if pk_columns.iter().any(|c| c == name) {
return Err(Error::invalid_input(format!(
"batch is missing primary key column `{name}` declared by the schema"
)));
}
return Ok(Source::Null(field.data_type().clone()));
};
let source = &source_fields[index];
if is_nested(field.data_type()) {
return Ok(Source::Nested(
index,
resolve_children(source, field)?,
field.data_type().clone(),
));
}
if source.data_type() == field.data_type() {
return Ok(Source::Take(index));
}
if !matches!(
(source.data_type(), field.data_type()),
(DataType::Struct(_), DataType::Struct(_))
) {
return Err(Error::invalid_input(format!(
"column `{name}` is stored as {} and the schema declares {}; a column's type \
cannot change on a table with a MemWAL",
source.data_type(),
field.data_type()
)));
}
unreachable!("a struct is resolved above and any other mismatch is rejected")
}
fn is_nested(data_type: &DataType) -> bool {
children_of(data_type).is_some()
}
fn children_of(data_type: &DataType) -> Option<arrow_schema::Fields> {
match data_type {
DataType::Struct(children) => Some(children.clone()),
DataType::List(element) | DataType::LargeList(element) => {
Some(vec![element.as_ref().clone()].into())
}
DataType::FixedSizeList(element, _) => Some(vec![element.as_ref().clone()].into()),
DataType::Map(entries, _) => Some(vec![entries.as_ref().clone()].into()),
_ => None,
}
}
fn resolve_children(source: &ArrowField, field: &ArrowField) -> Result<Vec<Source>> {
let (Some(source_children), Some(target_children)) = (
children_of(source.data_type()),
children_of(field.data_type()),
) else {
return Err(Error::invalid_input(format!(
"column `{}` is stored as {} and the schema declares {}; a column's type \
cannot change on a table with a MemWAL",
field.name(),
source.data_type(),
field.data_type()
)));
};
let claimed = claimed_by_id(&source_children, &target_children);
target_children
.iter()
.map(|child| resolve_field(child, &source_children, &claimed, &[], false))
.collect()
}
fn claimed_by_id(source: &arrow_schema::Fields, target: &arrow_schema::Fields) -> Vec<bool> {
let by_id: HashMap<i32, usize> = source
.iter()
.enumerate()
.filter_map(|(i, f)| field_id_of(f).map(|id| (id, i)))
.collect();
let mut claimed = vec![false; source.len()];
for field in target {
if let Some(i) = field_id_of(field).and_then(|id| by_id.get(&id)) {
claimed[*i] = true;
}
}
claimed
}
fn rebuild_list<O: arrow_array::OffsetSizeTrait>(
column: &ArrayRef,
element_source: &Source,
target: &DataType,
name: &str,
) -> Result<ArrayRef> {
let list = column
.as_any()
.downcast_ref::<GenericListArray<O>>()
.ok_or_else(|| Error::invalid_input(format!("column `{name}` is not a list")))?;
let Some(element) = children_of(target).and_then(|c| c.first().cloned()) else {
unreachable!("a list target has an element");
};
let child = take_column(
element_source,
std::slice::from_ref(list.values()),
list.values().len(),
element.name(),
)?;
Ok(Arc::new(
GenericListArray::<O>::try_new(
element,
list.offsets().clone(),
child,
list.nulls().cloned(),
)
.map_err(|e| Error::invalid_input(format!("rebuild list column `{name}`: {e}")))?,
))
}
fn take_column(source: &Source, columns: &[ArrayRef], rows: usize, name: &str) -> Result<ArrayRef> {
match source {
Source::Take(i) => Ok(Arc::clone(&columns[*i])),
Source::Nested(i, children, to @ DataType::List(_)) => {
rebuild_list::<i32>(&columns[*i], &children[0], to, name)
}
Source::Nested(i, children, to @ DataType::LargeList(_)) => {
rebuild_list::<i64>(&columns[*i], &children[0], to, name)
}
Source::Nested(i, children, to @ DataType::Map(_, sorted)) => {
let map = columns[*i]
.as_any()
.downcast_ref::<arrow_array::MapArray>()
.ok_or_else(|| Error::invalid_input(format!("column `{name}` is not a map")))?;
let Some(entries) = children_of(to).and_then(|c| c.first().cloned()) else {
unreachable!("a map target has an entries field");
};
let stored_entries: ArrayRef = Arc::new(map.entries().clone());
let rebuilt = take_column(
&children[0],
std::slice::from_ref(&stored_entries),
map.entries().len(),
entries.name(),
)?;
let rebuilt = rebuilt
.as_any()
.downcast_ref::<StructArray>()
.ok_or_else(|| {
Error::invalid_input(format!("map column `{name}` entries are not a struct"))
})?
.clone();
Ok(Arc::new(
arrow_array::MapArray::try_new(
entries,
map.offsets().clone(),
rebuilt,
map.nulls().cloned(),
*sorted,
)
.map_err(|e| Error::invalid_input(format!("rebuild map column `{name}`: {e}")))?,
))
}
Source::Nested(i, children, to @ DataType::FixedSizeList(_, _)) => {
let DataType::FixedSizeList(_, size) = to else {
unreachable!("matched above");
};
let list = columns[*i]
.as_any()
.downcast_ref::<FixedSizeListArray>()
.ok_or_else(|| {
Error::invalid_input(format!("column `{name}` is not a fixed-size list"))
})?;
let Some(element) = children_of(to).and_then(|c| c.first().cloned()) else {
unreachable!("a fixed-size list target has an element");
};
let child = take_column(
&children[0],
std::slice::from_ref(list.values()),
list.values().len(),
element.name(),
)?;
Ok(Arc::new(
FixedSizeListArray::try_new(element, *size, child, list.nulls().cloned()).map_err(
|e| {
Error::invalid_input(format!(
"rebuild fixed-size list column `{name}`: {e}"
))
},
)?,
))
}
Source::Nested(i, children, to) => {
let DataType::Struct(target_children) = to else {
unreachable!("Nested is only built for a struct or list target");
};
let struct_array = columns[*i]
.as_any()
.downcast_ref::<StructArray>()
.ok_or_else(|| Error::invalid_input(format!("column `{name}` is not a struct")))?;
let built = children
.iter()
.zip(target_children)
.map(|(child, field)| {
take_column(child, struct_array.columns(), rows, field.name())
})
.collect::<Result<Vec<_>>>()?;
Ok(Arc::new(
StructArray::try_new(
target_children.clone(),
built,
struct_array.nulls().cloned(),
)
.map_err(|e| {
Error::invalid_input(format!("rebuild struct column `{name}`: {e}"))
})?,
))
}
Source::Null(ty) => Ok(arrow_array::new_null_array(ty, rows)),
Source::Live => Ok(Arc::new(BooleanArray::from(vec![false; rows]))),
}
}
#[cfg(test)]
mod relabel_tests {
use super::*;
use arrow_array::{
Array, FixedSizeListArray, Int64Array, LargeListArray, ListArray, StructArray,
};
use arrow_buffer::{NullBuffer, OffsetBuffer};
use arrow_schema::Fields;
#[test]
fn every_type_lance_gives_children_to_is_one_we_recurse_into() {
let item = || Arc::new(ArrowField::new("item", DataType::Int64, true));
let a_struct = || Fields::from(vec![ArrowField::new("a", DataType::Int64, true)]);
let entries = Arc::new(ArrowField::new(
"entries",
DataType::Struct(Fields::from(vec![
ArrowField::new("key", DataType::Int64, false),
ArrowField::new("value", DataType::Int64, true),
])),
false,
));
let candidates = [
DataType::Int64,
DataType::Struct(a_struct()),
DataType::List(item()),
DataType::LargeList(item()),
DataType::FixedSizeList(item(), 2),
DataType::FixedSizeList(
Arc::new(ArrowField::new("item", DataType::Struct(a_struct()), true)),
2,
),
DataType::Map(entries, false),
DataType::ListView(item()),
DataType::LargeListView(item()),
DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
DataType::RunEndEncoded(
Arc::new(ArrowField::new("run_ends", DataType::Int32, false)),
item(),
),
];
for data_type in candidates {
let field = ArrowField::new("c", data_type.clone(), true);
let Ok(lance) = lance_core::datatypes::Field::try_from(&field) else {
continue;
};
if !lance.children.is_empty() {
assert!(
is_nested(&data_type),
"Lance gives {data_type:?} children, so reconcile must recurse into it"
);
}
}
}
fn stamped(name: &str, data_type: DataType, id: i32) -> ArrowField {
ArrowField::new(name, data_type, true).with_metadata(
[(LANCE_FIELD_ID_KEY.to_string(), id.to_string())]
.into_iter()
.collect(),
)
}
fn strip_and_check(column: ArrayRef) -> ArrayRef {
let plain = without_field_ids_in(column.data_type());
let out = relabel_to(&column, &plain).expect("relabel");
assert_eq!(out.data_type(), &plain, "every level is relabelled");
assert_eq!(out.len(), column.len(), "row count is preserved");
assert_eq!(
out.null_count(),
column.null_count(),
"validity is preserved"
);
out
}
#[test]
fn relabel_reaches_a_nested_child() {
let inner_stamped = stamped("b", DataType::Int64, 3);
let middle_stamped = stamped(
"inner",
DataType::Struct(Fields::from(vec![inner_stamped.clone()])),
2,
);
let outer_stamped = DataType::Struct(Fields::from(vec![middle_stamped.clone()]));
let leaf = Arc::new(Int64Array::from(vec![Some(7)])) as ArrayRef;
let middle = StructArray::new(
Fields::from(vec![inner_stamped]),
vec![Arc::clone(&leaf)],
None,
);
let outer = Arc::new(StructArray::new(
Fields::from(vec![middle_stamped]),
vec![Arc::new(middle) as ArrayRef],
None,
)) as ArrayRef;
assert_eq!(outer.data_type(), &outer_stamped);
let plain = without_field_ids_in(&outer_stamped);
let relabelled = relabel_to(&outer, &plain).expect("relabel a nested column");
assert_eq!(relabelled.data_type(), &plain, "every level is relabelled");
let as_struct = relabelled
.as_any()
.downcast_ref::<StructArray>()
.expect("a struct");
let middle = as_struct
.column(0)
.as_any()
.downcast_ref::<StructArray>()
.expect("a nested struct");
let values = middle
.column(0)
.as_any()
.downcast_ref::<Int64Array>()
.expect("the leaf");
assert_eq!(values.value(0), 7);
}
#[test]
fn a_renamed_field_inside_a_map_resolves_by_id() {
let entry = |value: &str| {
ArrowField::new(
"entries",
DataType::Struct(Fields::from(vec![
stamped("key", DataType::Int64, 2),
stamped(value, DataType::Int64, 3),
])),
false,
)
};
let keys = Arc::new(Int64Array::from(vec![1, 2])) as ArrayRef;
let values = Arc::new(Int64Array::from(vec![10, 20])) as ArrayRef;
let entries = StructArray::new(
Fields::from(vec![
stamped("key", DataType::Int64, 2),
stamped("old", DataType::Int64, 3),
]),
vec![keys, values],
None,
);
let column = Arc::new(
arrow_array::MapArray::try_new(
Arc::new(entry("old")),
arrow_buffer::OffsetBuffer::new(vec![0, 2].into()),
entries,
None,
false,
)
.unwrap(),
) as ArrayRef;
let source = ArrowSchema::new(vec![stamped(
"m",
DataType::Map(Arc::new(entry("old")), false),
1,
)]);
let target: SchemaRef = Arc::new(ArrowSchema::new(vec![stamped(
"m",
DataType::Map(Arc::new(entry("new")), false),
1,
)]));
let batch = RecordBatch::try_new(Arc::new(source.clone()), vec![column]).unwrap();
let plan = Plan::resolve(&source, &target, &[]).expect("a rename inside a map resolves");
let out = plan.apply(&batch).expect("apply");
let map = out
.column(0)
.as_any()
.downcast_ref::<arrow_array::MapArray>()
.expect("a map");
assert_eq!(
map.entries().column_names(),
vec!["key", "new"],
"the renamed entry arrives under its new name"
);
let vals = map
.entries()
.column(1)
.as_any()
.downcast_ref::<Int64Array>()
.expect("values");
assert_eq!(
(vals.value(0), vals.value(1)),
(10, 20),
"carrying its values"
);
assert_eq!(map.value_length(0), 2, "and its offsets");
}
#[test]
fn a_struct_child_named_like_the_tombstone_is_an_ordinary_field() {
let kept = stamped("kept", DataType::Int64, 2);
let source = ArrowSchema::new(vec![
stamped("id", DataType::Int64, 0),
stamped(
"info",
DataType::Struct(Fields::from(vec![kept.clone()])),
1,
),
]);
let target = Arc::new(ArrowSchema::new(vec![
stamped("id", DataType::Int64, 0),
stamped(
"info",
DataType::Struct(Fields::from(vec![
kept.clone(),
stamped(TOMBSTONE, DataType::Int64, 3),
])),
1,
),
]));
let batch = RecordBatch::try_new(
Arc::new(source.clone()),
vec![
Arc::new(Int64Array::from(vec![1])) as ArrayRef,
Arc::new(StructArray::new(
Fields::from(vec![kept]),
vec![Arc::new(Int64Array::from(vec![7])) as ArrayRef],
None,
)),
],
)
.expect("a batch under the source schema");
let out = Plan::resolve(&source, &target, &["id".to_string()])
.expect("a nested field named like the tombstone resolves")
.emitting_plain_schema()
.apply(&batch)
.expect("and applies");
let info = out
.column_by_name("info")
.expect("info")
.as_any()
.downcast_ref::<StructArray>()
.expect("a struct");
assert!(
info.column_by_name(TOMBSTONE)
.expect("the child")
.is_null(0),
"a child the generation never stored is null, whatever it is called"
);
}
#[test]
fn a_null_parent_and_a_null_child_both_survive() {
let child = stamped("b", DataType::Int64, 3);
let values = Arc::new(Int64Array::from(vec![Some(1), None, Some(3)])) as ArrayRef;
let column = Arc::new(StructArray::new(
Fields::from(vec![child]),
vec![values],
Some(NullBuffer::from(vec![true, true, false])),
)) as ArrayRef;
let out = strip_and_check(column);
let out = out.as_any().downcast_ref::<StructArray>().expect("struct");
assert!(out.is_null(2), "the null parent stays null");
let inner = out
.column(0)
.as_any()
.downcast_ref::<Int64Array>()
.expect("the child");
assert_eq!(inner.value(0), 1);
assert!(inner.is_null(1), "the null child stays null");
}
#[test]
fn a_lists_offsets_and_validity_survive() {
let element = stamped(
"item",
DataType::Struct(Fields::from(vec![stamped("b", DataType::Int64, 4)])),
3,
);
let leaf = Arc::new(Int64Array::from(vec![Some(1), None, Some(3)])) as ArrayRef;
let inner = Arc::new(StructArray::new(
Fields::from(vec![stamped("b", DataType::Int64, 4)]),
vec![leaf],
None,
)) as ArrayRef;
let column = Arc::new(ListArray::new(
Arc::new(element),
OffsetBuffer::new(vec![0, 2, 2, 3].into()),
inner,
Some(NullBuffer::from(vec![true, true, false])),
)) as ArrayRef;
let out = strip_and_check(column);
let out = out.as_any().downcast_ref::<ListArray>().expect("list");
assert_eq!(out.value_length(0), 2, "the first row keeps two elements");
assert_eq!(out.value_length(1), 0, "the empty list stays empty");
assert!(out.is_null(2), "the null list stays null");
}
#[test]
fn a_large_lists_offsets_survive() {
let element = stamped("item", DataType::Int64, 3);
let values = Arc::new(Int64Array::from(vec![1, 2, 3])) as ArrayRef;
let column = Arc::new(LargeListArray::new(
Arc::new(element),
OffsetBuffer::new(vec![0i64, 2, 3].into()),
values,
None,
)) as ArrayRef;
let out = strip_and_check(column);
let out = out
.as_any()
.downcast_ref::<LargeListArray>()
.expect("a large list, not a list");
assert_eq!(out.value_length(0), 2);
assert_eq!(out.value_length(1), 1);
}
#[test]
fn a_fixed_size_lists_width_survives() {
let element = stamped("item", DataType::Int64, 3);
let values = Arc::new(Int64Array::from(vec![1, 2, 3, 4])) as ArrayRef;
let column =
Arc::new(FixedSizeListArray::new(Arc::new(element), 2, values, None)) as ArrayRef;
let out = strip_and_check(column);
assert!(
matches!(out.data_type(), DataType::FixedSizeList(_, 2)),
"the width is part of the type, got {:?}",
out.data_type()
);
}
#[test]
fn a_slice_keeps_its_offset() {
let child = stamped("b", DataType::Int64, 3);
let values = Arc::new(Int64Array::from(vec![1, 2, 3, 4])) as ArrayRef;
let whole = StructArray::new(Fields::from(vec![child]), vec![values], None);
let column = Arc::new(whole.slice(2, 2)) as ArrayRef;
let out = strip_and_check(column);
let out = out.as_any().downcast_ref::<StructArray>().expect("struct");
let inner = out
.column(0)
.as_any()
.downcast_ref::<Int64Array>()
.expect("the child");
assert_eq!(
(0..out.len()).map(|i| inner.value(i)).collect::<Vec<_>>(),
vec![3, 4],
"the slice reads its own rows, not the array's first ones"
);
}
#[test]
fn an_empty_column_is_still_relabelled() {
let child = stamped("b", DataType::Int64, 3);
let column = Arc::new(StructArray::new(
Fields::from(vec![child]),
vec![Arc::new(Int64Array::from(Vec::<i64>::new())) as ArrayRef],
None,
)) as ArrayRef;
let out = strip_and_check(column);
assert_eq!(out.len(), 0);
}
#[test]
fn a_column_already_in_the_target_shape_is_returned_as_it_stands() {
let plain = ArrowField::new("b", DataType::Int64, true);
let column = Arc::new(StructArray::new(
Fields::from(vec![plain]),
vec![Arc::new(Int64Array::from(vec![1, 2])) as ArrayRef],
None,
)) as ArrayRef;
let out = relabel_to(&column, column.data_type()).expect("relabel");
assert_eq!(out.data_type(), column.data_type());
assert_eq!(out.len(), 2);
}
}