use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use crate::metadata_provider::DuckLakeTableColumn;
use crate::{DuckLakeError, Result};
use arrow::datatypes::{DataType, Field, IntervalUnit, Schema, TimeUnit};
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use parquet::file::metadata::ParquetMetaData;
pub(crate) const MAX_NESTED_TYPE_DEPTH: usize = 128;
pub fn ducklake_to_arrow_type(ducklake_type: &str) -> Result<DataType> {
ducklake_to_arrow_type_inner(ducklake_type, 0)
}
fn ducklake_to_arrow_type_inner(ducklake_type: &str, depth: usize) -> Result<DataType> {
if depth > MAX_NESTED_TYPE_DEPTH {
return Err(DuckLakeError::UnsupportedType(format!(
"Type exceeds maximum nesting depth {MAX_NESTED_TYPE_DEPTH}"
)));
}
let type_str = ducklake_type.trim();
if let Some(decimal_params) = parse_decimal(type_str)? {
return Ok(decimal_params);
}
if let Some(list_type) = parse_list_type(type_str, depth)? {
return Ok(list_type);
}
if let Some(struct_type) = parse_struct_type(type_str, depth)? {
return Ok(struct_type);
}
if let Some(map_type) = parse_map_type(type_str, depth)? {
return Ok(map_type);
}
match type_str.to_ascii_lowercase().as_str() {
"boolean" | "bool" => Ok(DataType::Boolean),
"int8" | "tinyint" => Ok(DataType::Int8),
"int16" | "smallint" => Ok(DataType::Int16),
"int32" | "int" | "integer" => Ok(DataType::Int32),
"int64" | "bigint" | "long" => Ok(DataType::Int64),
"uint8" | "utinyint" => Ok(DataType::UInt8),
"uint16" | "usmallint" => Ok(DataType::UInt16),
"uint32" | "uint" | "uinteger" => Ok(DataType::UInt32),
"uint64" | "ubigint" => Ok(DataType::UInt64),
"float32" | "float" | "real" => Ok(DataType::Float32),
"float64" | "double" => Ok(DataType::Float64),
"time" => Ok(DataType::Time64(TimeUnit::Microsecond)),
"date" => Ok(DataType::Date32),
"timestamp" => Ok(DataType::Timestamp(TimeUnit::Microsecond, None)),
"timestamptz" | "timestamp with time zone" => Ok(DataType::Timestamp(
TimeUnit::Microsecond,
Some("UTC".into()),
)),
"timestamptz_ns" => Ok(DataType::Timestamp(
TimeUnit::Nanosecond,
Some("UTC".into()),
)),
"timestamp_s" => Ok(DataType::Timestamp(TimeUnit::Second, None)),
"timestamp_ms" => Ok(DataType::Timestamp(TimeUnit::Millisecond, None)),
"timestamp_ns" => Ok(DataType::Timestamp(TimeUnit::Nanosecond, None)),
"interval" => Ok(DataType::Interval(IntervalUnit::MonthDayNano)),
"varchar" | "text" | "string" => Ok(DataType::Utf8View),
"json" => Ok(DataType::Utf8View),
"blob" | "binary" | "bytea" => Ok(DataType::BinaryView),
"uuid" => Ok(DataType::FixedSizeBinary(16)),
"point" | "linestring" | "polygon" | "multipoint" | "multilinestring" | "multipolygon"
| "geometrycollection" | "linestring z" | "geometry" => Ok(DataType::Binary),
"timetz" | "time with time zone" => Ok(DataType::Utf8View),
_ => Err(DuckLakeError::UnsupportedType(ducklake_type.to_string())),
}
}
pub fn arrow_to_ducklake_type(arrow_type: &DataType) -> Result<String> {
match arrow_type {
DataType::Boolean => Ok("boolean".to_string()),
DataType::Int8 => Ok("int8".to_string()),
DataType::Int16 => Ok("int16".to_string()),
DataType::Int32 => Ok("int32".to_string()),
DataType::Int64 => Ok("int64".to_string()),
DataType::UInt8 => Ok("uint8".to_string()),
DataType::UInt16 => Ok("uint16".to_string()),
DataType::UInt32 => Ok("uint32".to_string()),
DataType::UInt64 => Ok("uint64".to_string()),
DataType::Float32 => Ok("float32".to_string()),
DataType::Float64 => Ok("float64".to_string()),
DataType::Date32 | DataType::Date64 => Ok("date".to_string()),
DataType::Time32(_) | DataType::Time64(_) => Ok("time".to_string()),
DataType::Timestamp(TimeUnit::Second, None) => Ok("timestamp_s".to_string()),
DataType::Timestamp(TimeUnit::Millisecond, None) => Ok("timestamp_ms".to_string()),
DataType::Timestamp(TimeUnit::Microsecond, None) => Ok("timestamp".to_string()),
DataType::Timestamp(TimeUnit::Nanosecond, None) => Ok("timestamp_ns".to_string()),
DataType::Timestamp(TimeUnit::Nanosecond, Some(_)) => Ok("timestamptz_ns".to_string()),
DataType::Timestamp(_, Some(_)) => Ok("timestamptz".to_string()),
DataType::Interval(_) => Ok("interval".to_string()),
DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => Ok("varchar".to_string()),
DataType::Binary | DataType::LargeBinary | DataType::BinaryView => Ok("blob".to_string()),
DataType::FixedSizeBinary(16) => Ok("uuid".to_string()),
DataType::FixedSizeBinary(_) => Ok("blob".to_string()),
DataType::Decimal128(precision, scale) | DataType::Decimal256(precision, scale) => {
Ok(format!("decimal({}, {})", precision, scale))
},
DataType::Null => Ok("varchar".to_string()),
DataType::Dictionary(_, value) => arrow_to_ducklake_type(value),
DataType::List(field) | DataType::LargeList(field) => {
let inner = arrow_to_ducklake_type(field.data_type())?;
Ok(format!("list<{}>", inner))
},
DataType::FixedSizeList(field, _) => {
let inner = arrow_to_ducklake_type(field.data_type())?;
Ok(format!("list<{}>", inner))
},
DataType::Struct(fields) => {
let fields = fields
.iter()
.map(|field| {
Ok(format!(
"{}:{}",
field.name(),
arrow_to_ducklake_type(field.data_type())?
))
})
.collect::<Result<Vec<_>>>()?;
Ok(format!("struct<{}>", fields.join(",")))
},
DataType::Map(entries, _) => {
let DataType::Struct(fields) = entries.data_type() else {
return Err(DuckLakeError::UnsupportedType(
"Arrow map entries must be a struct".to_string(),
));
};
let [key, value] = fields.as_ref() else {
return Err(DuckLakeError::UnsupportedType(
"Arrow maps must have key and value fields".to_string(),
));
};
Ok(format!(
"map<{},{}>",
arrow_to_ducklake_type(key.data_type())?,
arrow_to_ducklake_type(value.data_type())?
))
},
other => Err(DuckLakeError::UnsupportedType(format!(
"Arrow type '{}' has no DuckLake equivalent",
other
))),
}
}
const DECIMAL_MAX_PRECISION: u8 = 76;
fn validate_decimal_precision_scale(precision: u8, scale: i8, type_str: &str) -> Result<()> {
if precision == 0 {
return Err(DuckLakeError::UnsupportedType(format!(
"Decimal precision must be >= 1, got 0 in type '{}'",
type_str
)));
}
if precision > DECIMAL_MAX_PRECISION {
return Err(DuckLakeError::UnsupportedType(format!(
"Decimal precision must be <= {}, got {} in type '{}'",
DECIMAL_MAX_PRECISION, precision, type_str
)));
}
if scale >= 0 && scale as u8 > precision {
return Err(DuckLakeError::UnsupportedType(format!(
"Decimal scale ({}) must not exceed precision ({}) in type '{}'",
scale, precision, type_str
)));
}
Ok(())
}
fn decimal_data_type(precision: u8, scale: i8) -> DataType {
if precision > 38 {
DataType::Decimal256(precision, scale)
} else {
DataType::Decimal128(precision, scale)
}
}
fn parse_decimal(type_str: &str) -> Result<Option<DataType>> {
let normalized = type_str.to_ascii_lowercase();
if !normalized.starts_with("decimal") && !normalized.starts_with("numeric") {
return Ok(None);
}
let start = match type_str.find('(') {
Some(s) => s,
None => return Ok(None),
};
let end = match type_str.find(')') {
Some(e) => e,
None => return Ok(None),
};
let params = &type_str[start + 1..end];
let parts: Vec<&str> = params.split(',').map(|s| s.trim()).collect();
match parts.len() {
1 => {
let precision: u8 = parts[0].parse().map_err(|_| {
DuckLakeError::UnsupportedType(format!(
"Invalid decimal precision '{}' in type '{}'",
parts[0], type_str
))
})?;
validate_decimal_precision_scale(precision, 0, type_str)?;
Ok(Some(decimal_data_type(precision, 0)))
},
2 => {
let precision: u8 = parts[0].parse().map_err(|_| {
DuckLakeError::UnsupportedType(format!(
"Invalid decimal precision '{}' in type '{}'",
parts[0], type_str
))
})?;
let scale: i8 = parts[1].parse().map_err(|_| {
DuckLakeError::UnsupportedType(format!(
"Invalid decimal scale '{}' in type '{}'",
parts[1], type_str
))
})?;
validate_decimal_precision_scale(precision, scale, type_str)?;
Ok(Some(decimal_data_type(precision, scale)))
},
n => Err(DuckLakeError::UnsupportedType(format!(
"Invalid decimal type: expected at most 2 parameters (precision, scale), got {} in type '{}'",
n, type_str
))),
}
}
fn parse_list_type(type_str: &str, depth: usize) -> Result<Option<DataType>> {
let normalized = type_str.to_ascii_lowercase();
let inner = if normalized.starts_with("list<") || normalized.starts_with("array<") {
let start = type_str.find('<').unwrap();
if !type_str.ends_with('>') {
return Ok(None);
}
&type_str[start + 1..type_str.len() - 1]
} else if let Some(stripped) = type_str.strip_suffix("[]") {
stripped
} else {
return Ok(None);
};
let inner = inner.trim();
if inner.is_empty() {
return Err(DuckLakeError::UnsupportedType(format!(
"List type '{}' has empty element type",
type_str
)));
}
let element_type = ducklake_to_arrow_type_inner(inner, depth + 1)?;
Ok(Some(DataType::List(Arc::new(Field::new(
"item",
element_type,
true,
)))))
}
fn parse_struct_type(type_str: &str, depth: usize) -> Result<Option<DataType>> {
if !type_str.to_ascii_lowercase().starts_with("struct<") {
return Ok(None);
}
if !type_str.ends_with('>') {
return Err(DuckLakeError::UnsupportedType(format!(
"Struct type '{type_str}' is missing its closing '>'"
)));
}
let inner = &type_str[7..type_str.len() - 1];
if inner.is_empty() {
return Ok(Some(DataType::Struct(Vec::<Arc<Field>>::new().into())));
}
let fields = split_top_level(inner, ',')?
.into_iter()
.map(|field| {
let Some(separator) = top_level_separator(field, ':') else {
return Err(DuckLakeError::UnsupportedType(format!(
"Struct field '{field}' must use name:type syntax"
)));
};
let name = field[..separator].trim();
let type_name = field[separator + 1..].trim();
if name.is_empty() || type_name.is_empty() {
return Err(DuckLakeError::UnsupportedType(format!(
"Struct field '{field}' must have a name and type"
)));
}
Ok(Arc::new(Field::new(
name,
ducklake_to_arrow_type_inner(type_name, depth + 1)?,
true,
)))
})
.collect::<Result<Vec<_>>>()?;
Ok(Some(DataType::Struct(fields.into())))
}
fn parse_map_type(type_str: &str, depth: usize) -> Result<Option<DataType>> {
if !type_str.to_ascii_lowercase().starts_with("map<") {
return Ok(None);
}
if !type_str.ends_with('>') {
return Err(DuckLakeError::UnsupportedType(format!(
"Map type '{type_str}' is missing its closing '>'"
)));
}
let parts = split_top_level(&type_str[4..type_str.len() - 1], ',')?;
let [key_type, value_type] = parts.as_slice() else {
return Err(DuckLakeError::UnsupportedType(format!(
"Map type '{type_str}' must have key and value types"
)));
};
let entries = DataType::Struct(
vec![
Arc::new(Field::new(
"key",
ducklake_to_arrow_type_inner(key_type, depth + 1)?,
false,
)),
Arc::new(Field::new(
"value",
ducklake_to_arrow_type_inner(value_type, depth + 1)?,
true,
)),
]
.into(),
);
Ok(Some(DataType::Map(
Arc::new(Field::new("entries", entries, false)),
false,
)))
}
fn split_top_level(value: &str, separator: char) -> Result<Vec<&str>> {
let mut parts = Vec::new();
let mut start = 0;
let mut angle_depth = 0_i32;
let mut paren_depth = 0_i32;
for (index, character) in value.char_indices() {
match character {
'<' => angle_depth += 1,
'>' => angle_depth -= 1,
'(' => paren_depth += 1,
')' => paren_depth -= 1,
_ => {},
}
if angle_depth < 0 || paren_depth < 0 {
return Err(DuckLakeError::UnsupportedType(format!(
"Unbalanced nested type '{value}'"
)));
}
if character == separator && angle_depth == 0 && paren_depth == 0 {
parts.push(value[start..index].trim());
start = index + character.len_utf8();
}
}
if angle_depth != 0 || paren_depth != 0 {
return Err(DuckLakeError::UnsupportedType(format!(
"Unbalanced nested type '{value}'"
)));
}
parts.push(value[start..].trim());
Ok(parts)
}
fn top_level_separator(value: &str, separator: char) -> Option<usize> {
let mut angle_depth = 0_i32;
let mut paren_depth = 0_i32;
for (index, character) in value.char_indices() {
match character {
'<' => angle_depth += 1,
'>' => angle_depth -= 1,
'(' => paren_depth += 1,
')' => paren_depth -= 1,
_ => {},
}
if character == separator && angle_depth == 0 && paren_depth == 0 {
return Some(index);
}
}
None
}
pub fn normalize_ducklake_type(ducklake_type: &str) -> Result<String> {
let arrow_type = ducklake_to_arrow_type(ducklake_type)?;
arrow_to_ducklake_type(&arrow_type)
}
pub fn is_promotable(from: &str, to: &str) -> bool {
let from_arrow = match ducklake_to_arrow_type(from) {
Ok(t) => t,
Err(_) => return false,
};
let to_arrow = match ducklake_to_arrow_type(to) {
Ok(t) => t,
Err(_) => return false,
};
is_arrow_promotable(&from_arrow, &to_arrow)
}
fn is_arrow_promotable(from: &DataType, to: &DataType) -> bool {
use DataType::*;
if from == to {
return true;
}
fn signed_int_rank(dt: &DataType) -> Option<u8> {
match dt {
Int8 => Some(0),
Int16 => Some(1),
Int32 => Some(2),
Int64 => Some(3),
_ => None,
}
}
fn unsigned_int_rank(dt: &DataType) -> Option<u8> {
match dt {
UInt8 => Some(0),
UInt16 => Some(1),
UInt32 => Some(2),
UInt64 => Some(3),
_ => None,
}
}
if let (Some(from_rank), Some(to_rank)) = (signed_int_rank(from), signed_int_rank(to)) {
return from_rank < to_rank;
}
if let (Some(from_rank), Some(to_rank)) = (unsigned_int_rank(from), unsigned_int_rank(to)) {
return from_rank < to_rank;
}
if matches!(from, Float32) && matches!(to, Float64) {
return true;
}
false
}
pub fn types_compatible(existing_type: &str, new_type: &str) -> bool {
let existing_normalized = match normalize_ducklake_type(existing_type) {
Ok(t) => t,
Err(_) => return false,
};
let new_normalized = match normalize_ducklake_type(new_type) {
Ok(t) => t,
Err(_) => return false,
};
if existing_normalized == new_normalized {
return true;
}
is_promotable(existing_type, new_type)
}
pub fn types_equal_canonical(a: &str, b: &str) -> bool {
match (normalize_ducklake_type(a), normalize_ducklake_type(b)) {
(Ok(na), Ok(nb)) => na == nb,
_ => false,
}
}
pub fn build_arrow_schema(columns: &[DuckLakeTableColumn]) -> Result<Schema> {
let fields: Result<Vec<Field>> = columns
.iter()
.map(|col| {
let data_type = read_compatible_data_type(&col.data_type()?, true);
Ok(Field::new(&col.column_name, data_type, col.is_nullable))
})
.collect();
Ok(Schema::new(fields?))
}
pub(crate) const ABSENT_FIELD_PREFIX: &str = "__ducklake_absent_field_";
fn read_compatible_data_type(data_type: &DataType, nullable_struct_fields: bool) -> DataType {
let rewrite_field = |field: &Arc<Field>, nullable: bool| {
Arc::new(
field
.as_ref()
.clone()
.with_data_type(read_compatible_data_type(field.data_type(), true))
.with_nullable(field.is_nullable() || nullable),
)
};
match data_type {
DataType::List(field) => DataType::List(rewrite_field(field, false)),
DataType::LargeList(field) => DataType::LargeList(rewrite_field(field, false)),
DataType::FixedSizeList(field, size) => {
DataType::FixedSizeList(rewrite_field(field, false), *size)
},
DataType::Struct(fields) => DataType::Struct(
fields
.iter()
.map(|field| rewrite_field(field, nullable_struct_fields))
.collect::<Vec<_>>()
.into(),
),
DataType::Map(entries, sorted) => DataType::Map(
Arc::new(
entries
.as_ref()
.clone()
.with_data_type(read_compatible_data_type(entries.data_type(), false)),
),
*sorted,
),
_ => data_type.clone(),
}
}
pub fn extract_parquet_field_ids(metadata: &ParquetMetaData) -> HashMap<i32, String> {
let schema_descr = metadata.file_metadata().schema_descr();
fn collect(type_: &parquet::schema::types::Type, entries: &mut Vec<(i32, String)>) {
let basic_info = type_.get_basic_info();
if basic_info.has_id() {
entries.push((basic_info.id(), type_.name().to_string()));
}
if type_.is_group() {
for child in type_.get_fields() {
collect(child, entries);
}
}
}
let mut entries = Vec::new();
for field in schema_descr.root_schema().get_fields() {
collect(field, &mut entries);
}
field_ids_dropping_duplicates(entries.into_iter())
}
fn field_ids_dropping_duplicates(
entries: impl Iterator<Item = (i32, String)>,
) -> HashMap<i32, String> {
let mut map: HashMap<i32, String> = HashMap::new();
let mut duplicates: HashSet<i32> = HashSet::new();
for (id, name) in entries {
if map.insert(id, name).is_some() {
duplicates.insert(id);
}
}
for id in duplicates {
map.remove(&id);
tracing::warn!(
field_id = id,
"parquet file has multiple top-level columns sharing this field_id; \
ignoring it — the affected column will read as NULL"
);
}
map
}
fn arrow_field_id_names(schema: &Schema) -> HashMap<i32, String> {
fn collect(field: &Field, names: &mut HashMap<i32, String>) {
if let Some(id) = field
.metadata()
.get(PARQUET_FIELD_ID_META_KEY)
.and_then(|value| value.parse().ok())
{
names.insert(id, field.name().clone());
}
match field.data_type() {
DataType::List(child)
| DataType::LargeList(child)
| DataType::FixedSizeList(child, _)
| DataType::Map(child, _) => collect(child, names),
DataType::Struct(fields) => {
for child in fields {
collect(child, names);
}
},
_ => {},
}
}
let mut names = HashMap::new();
for field in schema.fields() {
collect(field, &mut names);
}
names
}
fn data_type_has_field_ids(data_type: &DataType) -> bool {
let fields: &[Arc<Field>] = match data_type {
DataType::List(field)
| DataType::LargeList(field)
| DataType::FixedSizeList(field, _)
| DataType::Map(field, _) => std::slice::from_ref(field),
DataType::Struct(fields) => fields,
_ => return false,
};
fields.iter().any(|field| {
field.metadata().contains_key(PARQUET_FIELD_ID_META_KEY)
|| data_type_has_field_ids(field.data_type())
})
}
fn read_data_type_with_field_id_mapping(
data_type: &DataType,
nested_column_ids: &[i64],
parquet_field_ids: &HashMap<i32, String>,
arrow_field_names: &HashMap<i32, String>,
match_by_id: bool,
) -> Result<DataType> {
fn rewrite_field(
field: &Field,
ids: &mut std::slice::Iter<'_, i64>,
parquet_field_ids: &HashMap<i32, String>,
arrow_field_names: &HashMap<i32, String>,
match_by_id: bool,
) -> Result<Arc<Field>> {
let column_id = ids.next().ok_or_else(|| {
DuckLakeError::InvalidConfig(
"Nested column metadata has fewer field ids than its Arrow type".to_string(),
)
})?;
let field_id = i32::try_from(*column_id).map_err(|_| {
DuckLakeError::Internal(format!(
"column_id {column_id} for nested column '{}' exceeds i32 range for Parquet field_id",
field.name()
))
})?;
let is_absent = match_by_id && !parquet_field_ids.contains_key(&field_id);
let name = if let Some(parquet_name) = parquet_field_ids.get(&field_id) {
arrow_field_names
.get(&field_id)
.unwrap_or(parquet_name)
.clone()
} else if match_by_id {
format!("{ABSENT_FIELD_PREFIX}{field_id}")
} else {
field.name().clone()
};
let data_type = rewrite_type(
field.data_type(),
ids,
parquet_field_ids,
arrow_field_names,
match_by_id,
)?;
let field_metadata_id = (match_by_id && !is_absent).then(|| field_id.to_string());
let mut read_field = field
.clone()
.with_name(name)
.with_data_type(data_type)
.with_nullable(field.is_nullable() || is_absent);
if let Some(id) = field_metadata_id {
let mut metadata = read_field.metadata().clone();
metadata.insert(PARQUET_FIELD_ID_META_KEY.to_string(), id);
read_field = read_field.with_metadata(metadata);
}
Ok(Arc::new(read_field))
}
fn rewrite_type(
data_type: &DataType,
ids: &mut std::slice::Iter<'_, i64>,
parquet_field_ids: &HashMap<i32, String>,
arrow_field_names: &HashMap<i32, String>,
match_by_id: bool,
) -> Result<DataType> {
match data_type {
DataType::List(field) => Ok(DataType::List(rewrite_field(
field,
ids,
parquet_field_ids,
arrow_field_names,
match_by_id,
)?)),
DataType::LargeList(field) => Ok(DataType::LargeList(rewrite_field(
field,
ids,
parquet_field_ids,
arrow_field_names,
match_by_id,
)?)),
DataType::FixedSizeList(field, size) => Ok(DataType::FixedSizeList(
rewrite_field(
field,
ids,
parquet_field_ids,
arrow_field_names,
match_by_id,
)?,
*size,
)),
DataType::Struct(fields) => Ok(DataType::Struct(
fields
.iter()
.map(|field| {
rewrite_field(
field,
ids,
parquet_field_ids,
arrow_field_names,
match_by_id,
)
})
.collect::<Result<Vec<_>>>()?
.into(),
)),
DataType::Map(entries, sorted) => Ok(DataType::Map(
Arc::new(entries.as_ref().clone().with_data_type(rewrite_type(
entries.data_type(),
ids,
parquet_field_ids,
arrow_field_names,
match_by_id,
)?)),
*sorted,
)),
_ => Ok(data_type.clone()),
}
}
let mut ids = nested_column_ids.iter();
let result = rewrite_type(
data_type,
&mut ids,
parquet_field_ids,
arrow_field_names,
match_by_id,
)?;
if ids.next().is_some() {
return Err(DuckLakeError::InvalidConfig(
"Nested column metadata has more field ids than its Arrow type".to_string(),
));
}
Ok(result)
}
pub fn build_read_schema_with_field_id_mapping(
current_columns: &[DuckLakeTableColumn],
parquet_field_ids: &HashMap<i32, String>,
file_schema: Option<&Schema>,
) -> Result<(Schema, HashMap<String, String>)> {
let mut name_mapping: HashMap<String, String> = HashMap::new();
let arrow_field_names = file_schema.map(arrow_field_id_names).unwrap_or_default();
let fields: Result<Vec<Field>> = current_columns
.iter()
.map(|col| {
let mut data_type = read_compatible_data_type(&col.data_type()?, true);
let field_id = i32::try_from(col.column_id).map_err(|_| {
DuckLakeError::Internal(format!(
"column_id {} for column '{}' exceeds i32 range for Parquet field_id",
col.column_id, col.column_name
))
})?;
let (read_name, needs_rename, is_absent) =
if let Some(parquet_name) = parquet_field_ids.get(&field_id) {
if parquet_name != &col.column_name {
(parquet_name.clone(), true, false) } else {
(col.column_name.clone(), false, false)
}
} else if parquet_field_ids.is_empty() {
(col.column_name.clone(), false, false) } else {
(format!("{ABSENT_FIELD_PREFIX}{field_id}"), true, true)
};
if !is_absent && !col.nested_column_ids.is_empty() {
let match_nested_by_id = file_schema
.and_then(|schema| schema.field_with_name(&read_name).ok())
.is_some_and(|field| data_type_has_field_ids(field.data_type()));
data_type = read_data_type_with_field_id_mapping(
&data_type,
&col.nested_column_ids,
parquet_field_ids,
&arrow_field_names,
match_nested_by_id,
)?;
}
if needs_rename {
name_mapping.insert(read_name.clone(), col.column_name.clone());
}
Ok(Field::new(
read_name,
data_type,
col.is_nullable || is_absent,
))
})
.collect();
Ok((Schema::new(fields?), name_mapping))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_build_read_schema_with_renamed_columns() {
let current_columns = vec![
DuckLakeTableColumn {
column_id: 1,
column_name: "userId".to_string(), column_type: "int32".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
},
DuckLakeTableColumn {
column_id: 2,
column_name: "name".to_string(), column_type: "varchar".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
},
];
let mut parquet_field_ids = HashMap::new();
parquet_field_ids.insert(1, "user_id".to_string()); parquet_field_ids.insert(2, "name".to_string());
let (read_schema, name_mapping) =
build_read_schema_with_field_id_mapping(¤t_columns, &parquet_field_ids, None)
.unwrap();
assert_eq!(read_schema.field(0).name(), "user_id");
assert_eq!(read_schema.field(1).name(), "name");
assert_eq!(name_mapping.len(), 1);
assert_eq!(name_mapping.get("user_id"), Some(&"userId".to_string()));
}
#[test]
fn test_build_read_schema_no_rename_needed() {
let current_columns = vec![DuckLakeTableColumn {
column_id: 1,
column_name: "id".to_string(),
column_type: "int32".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
}];
let mut parquet_field_ids = HashMap::new();
parquet_field_ids.insert(1, "id".to_string());
let (read_schema, name_mapping) =
build_read_schema_with_field_id_mapping(¤t_columns, &parquet_field_ids, None)
.unwrap();
assert_eq!(read_schema.field(0).name(), "id");
assert!(name_mapping.is_empty()); }
#[test]
fn test_build_read_schema_no_field_ids() {
let current_columns = vec![DuckLakeTableColumn {
column_id: 1,
column_name: "id".to_string(),
column_type: "int32".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
}];
let parquet_field_ids = HashMap::new();
let (read_schema, name_mapping) =
build_read_schema_with_field_id_mapping(¤t_columns, &parquet_field_ids, None)
.unwrap();
assert_eq!(read_schema.field(0).name(), "id");
assert!(name_mapping.is_empty());
}
#[test]
fn test_build_read_schema_absent_field_id_reads_null() {
let current_columns = vec![
DuckLakeTableColumn {
column_id: 1,
column_name: "id".to_string(),
column_type: "int32".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
},
DuckLakeTableColumn {
column_id: 2,
column_name: "tag".to_string(),
column_type: "varchar".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
},
];
let mut parquet_field_ids = HashMap::new();
parquet_field_ids.insert(1, "id".to_string());
let (read_schema, name_mapping) =
build_read_schema_with_field_id_mapping(¤t_columns, &parquet_field_ids, None)
.unwrap();
assert_eq!(read_schema.field(0).name(), "id");
assert_ne!(read_schema.field(1).name(), "tag");
assert!(
read_schema
.field(1)
.name()
.starts_with("__ducklake_absent_field_")
);
assert!(read_schema.field(1).is_nullable());
assert_eq!(
name_mapping.get(read_schema.field(1).name()),
Some(&"tag".to_string())
);
}
#[test]
fn test_build_read_schema_keeps_new_nested_field() {
let current_columns = vec![DuckLakeTableColumn {
column_id: 1,
column_name: "payload".to_string(),
column_type: "struct<a:int32,b:varchar,c:int32>".to_string(),
is_nullable: false,
data_type: Some(DataType::Struct(
vec![
Arc::new(Field::new("a", DataType::Int32, false)),
Arc::new(Field::new("b", DataType::Utf8View, true)),
Arc::new(Field::new("c", DataType::Int32, false)),
]
.into(),
)),
nested_column_ids: vec![2, 3, 4],
}];
let parquet_field_ids = HashMap::from([
(1, "payload".to_string()),
(2, "a".to_string()),
(3, "old_b".to_string()),
]);
let id_metadata =
|id: i32| HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), id.to_string())]);
let file_schema = Schema::new(vec![
Field::new(
"payload",
DataType::Struct(
vec![
Arc::new(
Field::new("a", DataType::Int32, false).with_metadata(id_metadata(2)),
),
Arc::new(
Field::new("old_b", DataType::Utf8, true).with_metadata(id_metadata(3)),
),
]
.into(),
),
false,
)
.with_metadata(id_metadata(1)),
]);
let (read_schema, _) = build_read_schema_with_field_id_mapping(
¤t_columns,
&parquet_field_ids,
Some(&file_schema),
)
.unwrap();
let DataType::Struct(fields) = read_schema.field(0).data_type() else {
panic!("payload must remain a struct");
};
assert_eq!(fields.len(), 3);
assert_eq!(fields[0].name(), "a");
assert_eq!(fields[1].name(), "old_b");
assert_eq!(fields[2].name(), "__ducklake_absent_field_4");
assert_eq!(fields[2].data_type(), &DataType::Int32);
assert!(fields[2].is_nullable());
}
#[test]
fn test_build_read_schema_does_not_reuse_dropped_nested_name() {
let current_columns = vec![DuckLakeTableColumn {
column_id: 1,
column_name: "payload".to_string(),
column_type: "struct<c:int32>".to_string(),
is_nullable: false,
data_type: Some(DataType::Struct(
vec![Arc::new(Field::new("c", DataType::Int32, true))].into(),
)),
nested_column_ids: vec![4],
}];
let parquet_field_ids = HashMap::from([(1, "payload".to_string()), (3, "c".to_string())]);
let id_metadata =
|id: i32| HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), id.to_string())]);
let file_schema = Schema::new(vec![
Field::new(
"payload",
DataType::Struct(
vec![Arc::new(
Field::new("c", DataType::Int32, true).with_metadata(id_metadata(3)),
)]
.into(),
),
false,
)
.with_metadata(id_metadata(1)),
]);
let (read_schema, _) = build_read_schema_with_field_id_mapping(
¤t_columns,
&parquet_field_ids,
Some(&file_schema),
)
.unwrap();
let DataType::Struct(fields) = read_schema.field(0).data_type() else {
panic!("payload must remain a struct");
};
assert_eq!(fields[0].name(), "__ducklake_absent_field_4");
}
fn nested_field_ids(data_type: &DataType) -> Vec<(String, Option<String>)> {
fn walk(prefix: &str, field: &Field, out: &mut Vec<(String, Option<String>)>) {
let path = format!("{prefix}.{}", field.name());
out.push((
path.clone(),
field.metadata().get(PARQUET_FIELD_ID_META_KEY).cloned(),
));
collect(&path, field.data_type(), out);
}
fn collect(prefix: &str, data_type: &DataType, out: &mut Vec<(String, Option<String>)>) {
match data_type {
DataType::List(child)
| DataType::LargeList(child)
| DataType::FixedSizeList(child, _) => walk(prefix, child, out),
DataType::Struct(children) => {
for child in children {
walk(prefix, child, out);
}
},
DataType::Map(entries, _) => collect(prefix, entries.data_type(), out),
_ => {},
}
}
let mut out = Vec::new();
collect("", data_type, &mut out);
out
}
fn nested_test_columns() -> Vec<DuckLakeTableColumn> {
vec![
DuckLakeTableColumn {
column_id: 1,
column_name: "v".to_string(),
column_type: "list".to_string(),
is_nullable: true,
data_type: Some(DataType::List(Arc::new(Field::new(
"item",
DataType::Float32,
true,
)))),
nested_column_ids: vec![2],
},
DuckLakeTableColumn {
column_id: 3,
column_name: "s".to_string(),
column_type: "struct".to_string(),
is_nullable: true,
data_type: Some(DataType::Struct(
vec![
Arc::new(Field::new("label", DataType::Utf8View, true)),
Arc::new(Field::new("score", DataType::Int32, true)),
]
.into(),
)),
nested_column_ids: vec![4, 5],
},
DuckLakeTableColumn {
column_id: 6,
column_name: "m".to_string(),
column_type: "map".to_string(),
is_nullable: true,
data_type: Some(DataType::Map(
Arc::new(Field::new(
"entries",
DataType::Struct(
vec![
Arc::new(Field::new("key", DataType::Utf8View, false)),
Arc::new(Field::new("value", DataType::Int32, true)),
]
.into(),
),
false,
)),
false,
)),
nested_column_ids: vec![7, 8],
},
DuckLakeTableColumn {
column_id: 9,
column_name: "nn".to_string(),
column_type: "list".to_string(),
is_nullable: true,
data_type: Some(DataType::List(Arc::new(Field::new(
"item",
DataType::List(Arc::new(Field::new("item", DataType::Int32, true))),
true,
)))),
nested_column_ids: vec![10, 11],
},
]
}
fn nested_test_file_schema() -> Schema {
let id = |id: i32| HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), id.to_string())]);
Schema::new(vec![
Field::new(
"v",
DataType::List(Arc::new(
Field::new("element", DataType::Float32, true).with_metadata(id(2)),
)),
true,
)
.with_metadata(id(1)),
Field::new(
"s",
DataType::Struct(
vec![
Arc::new(Field::new("label", DataType::Utf8, true).with_metadata(id(4))),
Arc::new(Field::new("score", DataType::Int32, true).with_metadata(id(5))),
]
.into(),
),
true,
)
.with_metadata(id(3)),
Field::new(
"m",
DataType::Map(
Arc::new(Field::new(
"key_value",
DataType::Struct(
vec![
Arc::new(
Field::new("key", DataType::Utf8, false).with_metadata(id(7)),
),
Arc::new(
Field::new("value", DataType::Int32, true).with_metadata(id(8)),
),
]
.into(),
),
false,
)),
false,
),
true,
)
.with_metadata(id(6)),
Field::new(
"nn",
DataType::List(Arc::new(
Field::new(
"element",
DataType::List(Arc::new(
Field::new("element", DataType::Int32, true).with_metadata(id(11)),
)),
true,
)
.with_metadata(id(10)),
)),
true,
)
.with_metadata(id(9)),
])
}
fn nested_test_parquet_field_ids() -> HashMap<i32, String> {
HashMap::from([
(1, "v".to_string()),
(2, "element".to_string()),
(3, "s".to_string()),
(4, "label".to_string()),
(5, "score".to_string()),
(6, "m".to_string()),
(7, "key".to_string()),
(8, "value".to_string()),
(9, "nn".to_string()),
(10, "element".to_string()),
(11, "element".to_string()),
])
}
#[test]
fn test_read_schema_declares_nested_field_ids() {
let (read_schema, _) = build_read_schema_with_field_id_mapping(
&nested_test_columns(),
&nested_test_parquet_field_ids(),
Some(&nested_test_file_schema()),
)
.unwrap();
let ids: Vec<(String, Option<String>)> = read_schema
.fields()
.iter()
.flat_map(|field| nested_field_ids(field.data_type()))
.collect();
assert_eq!(
ids,
vec![
(".element".to_string(), Some("2".to_string())),
(".label".to_string(), Some("4".to_string())),
(".score".to_string(), Some("5".to_string())),
(".key".to_string(), Some("7".to_string())),
(".value".to_string(), Some("8".to_string())),
(".element".to_string(), Some("10".to_string())),
(".element.element".to_string(), Some("11".to_string())),
],
"every nested node must declare the field id the file records for it"
);
let DataType::Map(entries, _) = read_schema.field(2).data_type() else {
panic!("m must remain a map");
};
assert!(
!entries.metadata().contains_key(PARQUET_FIELD_ID_META_KEY),
"the map wrapper group has no DuckLake column and must carry no field id"
);
for field in read_schema.fields() {
assert!(
!field.metadata().contains_key(PARQUET_FIELD_ID_META_KEY),
"top-level field '{}' must not be tagged",
field.name()
);
}
}
#[test]
fn test_read_schema_omits_nested_field_ids_for_file_without_them() {
let file_schema = Schema::new(vec![
Field::new(
"v",
DataType::List(Arc::new(Field::new("item", DataType::Float32, true))),
true,
),
Field::new(
"s",
DataType::Struct(
vec![
Arc::new(Field::new("label", DataType::Utf8, true)),
Arc::new(Field::new("score", DataType::Int32, true)),
]
.into(),
),
true,
),
]);
let columns = nested_test_columns()[..2].to_vec();
let parquet_field_ids = HashMap::from([(1, "v".to_string()), (3, "s".to_string())]);
let (read_schema, _) = build_read_schema_with_field_id_mapping(
&columns,
&parquet_field_ids,
Some(&file_schema),
)
.unwrap();
let ids: Vec<(String, Option<String>)> = read_schema
.fields()
.iter()
.flat_map(|field| nested_field_ids(field.data_type()))
.collect();
assert_eq!(
ids,
vec![
(".item".to_string(), None),
(".label".to_string(), None),
(".score".to_string(), None),
],
"a file without nested field ids must not be described as having them"
);
}
#[test]
fn test_read_schema_omits_field_id_for_absent_nested_field() {
let mut parquet_field_ids = nested_test_parquet_field_ids();
parquet_field_ids.remove(&5);
let (read_schema, _) = build_read_schema_with_field_id_mapping(
&nested_test_columns()[1..2],
&parquet_field_ids,
Some(&nested_test_file_schema()),
)
.unwrap();
let DataType::Struct(fields) = read_schema.field(0).data_type() else {
panic!("s must remain a struct");
};
assert_eq!(
fields[0].metadata().get(PARQUET_FIELD_ID_META_KEY),
Some(&"4".to_string()),
"a present nested field keeps its id"
);
assert_eq!(fields[1].name(), "__ducklake_absent_field_5");
assert!(
!fields[1].metadata().contains_key(PARQUET_FIELD_ID_META_KEY),
"a nested field absent from the file has no physical node to tag"
);
}
#[test]
fn test_nested_type_depth_is_bounded() {
let nested = format!(
"{}int32{}",
"list<".repeat(MAX_NESTED_TYPE_DEPTH + 1),
">".repeat(MAX_NESTED_TYPE_DEPTH + 1)
);
let error = ducklake_to_arrow_type(&nested).unwrap_err();
assert!(error.to_string().contains("maximum nesting depth"));
}
#[test]
fn test_basic_types() {
assert_eq!(
ducklake_to_arrow_type("boolean").unwrap(),
DataType::Boolean
);
assert_eq!(ducklake_to_arrow_type("int32").unwrap(), DataType::Int32);
assert_eq!(ducklake_to_arrow_type("int64").unwrap(), DataType::Int64);
assert_eq!(
ducklake_to_arrow_type("float64").unwrap(),
DataType::Float64
);
assert_eq!(
ducklake_to_arrow_type("varchar").unwrap(),
DataType::Utf8View
);
assert_eq!(
ducklake_to_arrow_type("blob").unwrap(),
DataType::BinaryView
);
}
#[test]
fn test_string_types_map_to_utf8view() {
for t in ["varchar", "text", "string", "json", "timetz", "time with time zone"] {
assert_eq!(
ducklake_to_arrow_type(t).unwrap(),
DataType::Utf8View,
"{t} should map to Utf8View"
);
}
}
#[test]
fn test_binary_types_map_to_binaryview() {
for t in ["blob", "binary", "bytea"] {
assert_eq!(
ducklake_to_arrow_type(t).unwrap(),
DataType::BinaryView,
"{t} should map to BinaryView"
);
}
}
#[test]
fn test_geometry_stays_binary() {
for t in [
"geometry",
"point",
"linestring",
"polygon",
"multipoint",
"multilinestring",
"multipolygon",
"geometrycollection",
] {
assert_eq!(
ducklake_to_arrow_type(t).unwrap(),
DataType::Binary,
"{t} should stay Binary"
);
}
}
#[test]
fn test_uuid_stays_fixed_size_binary() {
assert_eq!(
ducklake_to_arrow_type("uuid").unwrap(),
DataType::FixedSizeBinary(16)
);
}
#[test]
fn test_view_types_write_back_to_string_and_blob() {
assert_eq!(
arrow_to_ducklake_type(&DataType::Utf8View).unwrap(),
"varchar"
);
assert_eq!(arrow_to_ducklake_type(&DataType::Utf8).unwrap(), "varchar");
assert_eq!(
arrow_to_ducklake_type(&DataType::LargeUtf8).unwrap(),
"varchar"
);
assert_eq!(
arrow_to_ducklake_type(&DataType::BinaryView).unwrap(),
"blob"
);
assert_eq!(arrow_to_ducklake_type(&DataType::Binary).unwrap(), "blob");
assert_eq!(
arrow_to_ducklake_type(&DataType::LargeBinary).unwrap(),
"blob"
);
}
#[test]
fn test_dictionary_uses_logical_value_type() {
let dictionary = DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
assert_eq!(arrow_to_ducklake_type(&dictionary).unwrap(), "varchar");
let list = DataType::List(Arc::new(Field::new("item", dictionary, true)));
assert_eq!(arrow_to_ducklake_type(&list).unwrap(), "list<varchar>");
}
#[test]
fn test_string_binary_normalize_is_stable() {
assert_eq!(normalize_ducklake_type("varchar").unwrap(), "varchar");
assert_eq!(normalize_ducklake_type("text").unwrap(), "varchar");
assert_eq!(normalize_ducklake_type("json").unwrap(), "varchar");
assert_eq!(normalize_ducklake_type("blob").unwrap(), "blob");
assert_eq!(normalize_ducklake_type("binary").unwrap(), "blob");
}
#[test]
fn test_view_type_list_children() {
assert_eq!(
ducklake_to_arrow_type("list<varchar>").unwrap(),
DataType::List(Arc::new(Field::new("item", DataType::Utf8View, true)))
);
}
#[test]
fn test_decimal_types() {
assert_eq!(
ducklake_to_arrow_type("decimal(10, 2)").unwrap(),
DataType::Decimal128(10, 2)
);
assert_eq!(
ducklake_to_arrow_type("decimal(38, 10)").unwrap(),
DataType::Decimal128(38, 10)
);
}
#[test]
fn test_decimal_single_param_over_38_uses_decimal256() {
assert_eq!(
ducklake_to_arrow_type("decimal(50)").unwrap(),
DataType::Decimal256(50, 0)
);
assert_eq!(
ducklake_to_arrow_type("decimal(38)").unwrap(),
DataType::Decimal128(38, 0)
);
assert_eq!(
ducklake_to_arrow_type("decimal(50, 10)").unwrap(),
DataType::Decimal256(50, 10)
);
}
#[test]
fn test_field_ids_dropping_duplicates() {
let unique =
field_ids_dropping_duplicates([(1, "a".to_string()), (2, "b".to_string())].into_iter());
assert_eq!(unique.get(&1), Some(&"a".to_string()));
assert_eq!(unique.get(&2), Some(&"b".to_string()));
let with_dup = field_ids_dropping_duplicates(
[(3, "a".to_string()), (3, "b".to_string()), (4, "c".to_string())].into_iter(),
);
assert_eq!(
with_dup.get(&3),
None,
"a duplicated field_id must not map to either column"
);
assert_eq!(with_dup.get(&4), Some(&"c".to_string()));
}
#[test]
fn test_temporal_types() {
assert_eq!(ducklake_to_arrow_type("date").unwrap(), DataType::Date32);
assert_eq!(
ducklake_to_arrow_type("timestamp").unwrap(),
DataType::Timestamp(TimeUnit::Microsecond, None)
);
assert_eq!(
ducklake_to_arrow_type("timestamptz").unwrap(),
DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()))
);
assert_eq!(
ducklake_to_arrow_type("timestamptz_ns").unwrap(),
DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()))
);
}
#[test]
fn test_list_type_angle_bracket() {
let result = ducklake_to_arrow_type("list<int32>").unwrap();
let expected = DataType::List(Arc::new(Field::new("item", DataType::Int32, true)));
assert_eq!(result, expected);
}
#[test]
fn test_list_type_various_elements() {
let cases = vec![
("list<varchar>", DataType::Utf8View),
("list<float64>", DataType::Float64),
("list<boolean>", DataType::Boolean),
("list<date>", DataType::Date32),
];
for (type_str, expected_inner) in cases {
let result = ducklake_to_arrow_type(type_str).unwrap();
let expected =
DataType::List(Arc::new(Field::new("item", expected_inner.clone(), true)));
assert_eq!(result, expected, "Failed for {}", type_str);
}
}
#[test]
fn test_array_type_angle_bracket() {
let result = ducklake_to_arrow_type("array<varchar>").unwrap();
let expected = DataType::List(Arc::new(Field::new("item", DataType::Utf8View, true)));
assert_eq!(result, expected);
}
#[test]
fn test_list_type_postgres_bracket_syntax() {
let cases = vec![
("varchar[]", DataType::Utf8View),
("float64[]", DataType::Float64),
("int32[]", DataType::Int32),
("boolean[]", DataType::Boolean),
("bigint[]", DataType::Int64),
("text[]", DataType::Utf8View),
("float[]", DataType::Float32),
("integer[]", DataType::Int32),
];
for (type_str, expected_inner) in cases {
let result = ducklake_to_arrow_type(type_str).unwrap();
let expected =
DataType::List(Arc::new(Field::new("item", expected_inner.clone(), true)));
assert_eq!(result, expected, "Failed for {}", type_str);
}
}
#[test]
fn test_list_type_empty_element_errors() {
assert!(ducklake_to_arrow_type("list<>").is_err());
assert!(ducklake_to_arrow_type("[]").is_err());
}
#[test]
fn test_struct_type() {
let result = ducklake_to_arrow_type("STRUCT<Price:DECIMAL(38,16),Label:VARCHAR>").unwrap();
assert_eq!(
result,
DataType::Struct(
vec![
Arc::new(Field::new("Price", DataType::Decimal128(38, 16), true,)),
Arc::new(Field::new("Label", DataType::Utf8View, true)),
]
.into(),
)
);
}
#[test]
fn test_map_type() {
let result = ducklake_to_arrow_type("map<varchar,list<int32>>").unwrap();
let DataType::Map(entries, false) = result else {
panic!("expected map");
};
let DataType::Struct(fields) = entries.data_type() else {
panic!("expected map entries struct");
};
assert_eq!(fields[0].name(), "key");
assert_eq!(fields[0].data_type(), &DataType::Utf8View);
assert_eq!(
fields[1].data_type(),
&DataType::List(Arc::new(Field::new("item", DataType::Int32, true)))
);
}
#[test]
fn test_arbitrarily_composed_nested_types() {
let type_name = "list<struct<levels:list<struct<price:decimal(38, 16)>>,attrs:map<varchar,list<int32>>>>";
let arrow = ducklake_to_arrow_type(type_name).unwrap();
assert_eq!(arrow_to_ducklake_type(&arrow).unwrap(), type_name);
let nested_lists = ducklake_to_arrow_type("int32[][]").unwrap();
assert_eq!(
arrow_to_ducklake_type(&nested_lists).unwrap(),
"list<list<int32>>"
);
}
#[test]
fn test_unknown_type_error() {
let result = ducklake_to_arrow_type("completely_unknown_type");
assert!(result.is_err());
match result {
Err(DuckLakeError::UnsupportedType(msg)) => {
assert_eq!(msg, "completely_unknown_type");
},
_ => panic!("Expected UnsupportedType error for unknown type"),
}
}
#[test]
fn test_arrow_to_ducklake_basic_types() {
assert_eq!(
arrow_to_ducklake_type(&DataType::Boolean).unwrap(),
"boolean"
);
assert_eq!(arrow_to_ducklake_type(&DataType::Int8).unwrap(), "int8");
assert_eq!(arrow_to_ducklake_type(&DataType::Int16).unwrap(), "int16");
assert_eq!(arrow_to_ducklake_type(&DataType::Int32).unwrap(), "int32");
assert_eq!(arrow_to_ducklake_type(&DataType::Int64).unwrap(), "int64");
assert_eq!(arrow_to_ducklake_type(&DataType::UInt8).unwrap(), "uint8");
assert_eq!(arrow_to_ducklake_type(&DataType::UInt16).unwrap(), "uint16");
assert_eq!(arrow_to_ducklake_type(&DataType::UInt32).unwrap(), "uint32");
assert_eq!(arrow_to_ducklake_type(&DataType::UInt64).unwrap(), "uint64");
assert_eq!(
arrow_to_ducklake_type(&DataType::Float32).unwrap(),
"float32"
);
assert_eq!(
arrow_to_ducklake_type(&DataType::Float64).unwrap(),
"float64"
);
assert_eq!(arrow_to_ducklake_type(&DataType::Utf8).unwrap(), "varchar");
assert_eq!(arrow_to_ducklake_type(&DataType::Binary).unwrap(), "blob");
}
#[test]
fn test_arrow_to_ducklake_temporal_types() {
assert_eq!(arrow_to_ducklake_type(&DataType::Date32).unwrap(), "date");
assert_eq!(arrow_to_ducklake_type(&DataType::Date64).unwrap(), "date");
assert_eq!(
arrow_to_ducklake_type(&DataType::Time64(TimeUnit::Microsecond)).unwrap(),
"time"
);
assert_eq!(
arrow_to_ducklake_type(&DataType::Timestamp(TimeUnit::Microsecond, None)).unwrap(),
"timestamp"
);
assert_eq!(
arrow_to_ducklake_type(&DataType::Timestamp(
TimeUnit::Microsecond,
Some("UTC".into())
))
.unwrap(),
"timestamptz"
);
assert_eq!(
arrow_to_ducklake_type(&DataType::Timestamp(
TimeUnit::Nanosecond,
Some("UTC".into())
))
.unwrap(),
"timestamptz_ns"
);
assert_eq!(
arrow_to_ducklake_type(&DataType::Timestamp(
TimeUnit::Nanosecond,
Some("America/New_York".into())
))
.unwrap(),
"timestamptz_ns"
);
assert_eq!(
arrow_to_ducklake_type(&DataType::Timestamp(
TimeUnit::Millisecond,
Some("UTC".into())
))
.unwrap(),
"timestamptz"
);
}
#[test]
fn test_arrow_to_ducklake_decimal() {
assert_eq!(
arrow_to_ducklake_type(&DataType::Decimal128(10, 2)).unwrap(),
"decimal(10, 2)"
);
assert_eq!(
arrow_to_ducklake_type(&DataType::Decimal256(40, 5)).unwrap(),
"decimal(40, 5)"
);
}
#[test]
fn test_arrow_to_ducklake_uuid() {
assert_eq!(
arrow_to_ducklake_type(&DataType::FixedSizeBinary(16)).unwrap(),
"uuid"
);
assert_eq!(
arrow_to_ducklake_type(&DataType::FixedSizeBinary(32)).unwrap(),
"blob"
);
}
#[test]
fn test_arrow_to_ducklake_roundtrip() {
let test_types = vec![
DataType::Boolean,
DataType::Int32,
DataType::Int64,
DataType::Float64,
DataType::Utf8View,
DataType::BinaryView,
DataType::Date32,
DataType::Timestamp(TimeUnit::Microsecond, None),
DataType::Timestamp(TimeUnit::Nanosecond, None),
DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into())),
DataType::Decimal128(10, 2),
DataType::List(Arc::new(Field::new("item", DataType::Int32, true))),
DataType::List(Arc::new(Field::new("item", DataType::Utf8View, true))),
];
for original in test_types {
let ducklake = arrow_to_ducklake_type(&original).unwrap();
let back = ducklake_to_arrow_type(&ducklake).unwrap();
assert_eq!(original, back, "Roundtrip failed for {:?}", original);
}
}
#[test]
fn test_nanosecond_timestamptz_preserves_precision() {
let ns_tz = DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()));
let ducklake = arrow_to_ducklake_type(&ns_tz).unwrap();
assert_eq!(ducklake, "timestamptz_ns");
assert_ne!(
ducklake, "timestamptz",
"ns tz-aware timestamp must not collapse to µs timestamptz"
);
let back = ducklake_to_arrow_type(&ducklake).unwrap();
assert_eq!(back, ns_tz);
assert!(!is_promotable("timestamptz", "timestamptz_ns"));
assert!(!is_promotable("timestamptz_ns", "timestamptz"));
assert!(!types_compatible("timestamptz", "timestamptz_ns"));
}
#[test]
fn test_arrow_to_ducklake_list() {
let list_type = DataType::List(Arc::new(Field::new("item", DataType::Int32, true)));
assert_eq!(arrow_to_ducklake_type(&list_type).unwrap(), "list<int32>");
let list_type = DataType::List(Arc::new(Field::new("item", DataType::Utf8, true)));
assert_eq!(arrow_to_ducklake_type(&list_type).unwrap(), "list<varchar>");
let large_list = DataType::LargeList(Arc::new(Field::new("item", DataType::Float64, true)));
assert_eq!(
arrow_to_ducklake_type(&large_list).unwrap(),
"list<float64>"
);
}
#[test]
fn test_arrow_to_ducklake_struct() {
let struct_type = DataType::Struct(vec![Field::new("a", DataType::Int32, true)].into());
assert_eq!(
arrow_to_ducklake_type(&struct_type).unwrap(),
"struct<a:int32>"
);
}
#[test]
fn test_decimal_precision_zero_rejected() {
let result = ducklake_to_arrow_type("decimal(0, 0)");
assert!(result.is_err());
match result {
Err(DuckLakeError::UnsupportedType(msg)) => {
assert!(msg.contains("precision must be >= 1"));
},
_ => panic!("Expected UnsupportedType error for precision=0"),
}
}
#[test]
fn test_decimal_precision_too_large_rejected() {
let result = ducklake_to_arrow_type("decimal(77, 0)");
assert!(result.is_err());
match result {
Err(DuckLakeError::UnsupportedType(msg)) => {
assert!(msg.contains("precision must be <= 76"));
},
_ => panic!("Expected UnsupportedType error for precision=77"),
}
}
#[test]
fn test_decimal_precision_255_rejected() {
let result = ducklake_to_arrow_type("decimal(255, 0)");
assert!(result.is_err());
match result {
Err(DuckLakeError::UnsupportedType(msg)) => {
assert!(msg.contains("precision must be <= 76"));
},
_ => panic!("Expected UnsupportedType error for precision=255"),
}
}
#[test]
fn test_decimal_scale_exceeds_precision_rejected() {
let result = ducklake_to_arrow_type("decimal(10, 11)");
assert!(result.is_err());
match result {
Err(DuckLakeError::UnsupportedType(msg)) => {
assert!(msg.contains("scale (11) must not exceed precision (10)"));
},
_ => panic!("Expected UnsupportedType error for scale > precision"),
}
}
#[test]
fn test_decimal_valid_edge_cases() {
assert_eq!(
ducklake_to_arrow_type("decimal(1, 0)").unwrap(),
DataType::Decimal128(1, 0)
);
assert_eq!(
ducklake_to_arrow_type("decimal(38, 0)").unwrap(),
DataType::Decimal128(38, 0)
);
assert_eq!(
ducklake_to_arrow_type("decimal(39, 0)").unwrap(),
DataType::Decimal256(39, 0)
);
assert_eq!(
ducklake_to_arrow_type("decimal(76, 0)").unwrap(),
DataType::Decimal256(76, 0)
);
assert_eq!(
ducklake_to_arrow_type("decimal(10, 10)").unwrap(),
DataType::Decimal128(10, 10)
);
}
#[test]
fn test_decimal_negative_precision_rejected() {
let result = ducklake_to_arrow_type("decimal(-1, 0)");
assert!(result.is_err());
}
#[test]
fn test_decimal_too_many_parameters_rejected() {
let result = ducklake_to_arrow_type("decimal(1,2,3)");
assert!(result.is_err());
match result {
Err(DuckLakeError::UnsupportedType(msg)) => {
assert!(msg.contains("expected at most 2 parameters"));
assert!(msg.contains("got 3"));
},
_ => panic!("Expected UnsupportedType error for 3 parameters"),
}
let result = ducklake_to_arrow_type("decimal(10,2,5,3)");
assert!(result.is_err());
match result {
Err(DuckLakeError::UnsupportedType(msg)) => {
assert!(msg.contains("expected at most 2 parameters"));
assert!(msg.contains("got 4"));
},
_ => panic!("Expected UnsupportedType error for 4 parameters"),
}
}
#[test]
fn test_decimal_negative_scale_valid() {
assert_eq!(
ducklake_to_arrow_type("decimal(10, -2)").unwrap(),
DataType::Decimal128(10, -2)
);
}
#[test]
fn test_build_schema_with_list_type() {
let columns = vec![
DuckLakeTableColumn {
column_id: 1,
column_name: "id".to_string(),
column_type: "int32".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
},
DuckLakeTableColumn {
column_id: 2,
column_name: "tags".to_string(),
column_type: "list<varchar>".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
},
];
let schema = build_arrow_schema(&columns).unwrap();
assert_eq!(schema.fields().len(), 2);
assert_eq!(
*schema.field(1).data_type(),
DataType::List(Arc::new(Field::new("item", DataType::Utf8View, true)))
);
}
#[test]
fn test_build_schema_with_struct_type() {
let columns = vec![DuckLakeTableColumn {
column_id: 1,
column_name: "data".to_string(),
column_type: "struct<a:int32>".to_string(),
is_nullable: true,
data_type: Some(DataType::Struct(
vec![Arc::new(Field::new("a", DataType::Int32, false))].into(),
)),
nested_column_ids: Vec::new(),
}];
let schema = build_arrow_schema(&columns).unwrap();
assert_eq!(
schema.field(0).data_type(),
&DataType::Struct(vec![Field::new("a", DataType::Int32, true)].into())
);
}
#[test]
fn test_column_id_i32_max_succeeds() {
let columns = vec![DuckLakeTableColumn {
column_id: i32::MAX as i64,
column_name: "id".to_string(),
column_type: "int32".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
}];
let mut parquet_field_ids = HashMap::new();
parquet_field_ids.insert(i32::MAX, "id".to_string());
let result = build_read_schema_with_field_id_mapping(&columns, &parquet_field_ids, None);
assert!(result.is_ok(), "column_id = i32::MAX should succeed");
}
#[test]
fn test_column_id_overflow_returns_error() {
let columns = vec![DuckLakeTableColumn {
column_id: i32::MAX as i64 + 1, column_name: "id".to_string(),
column_type: "int32".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
}];
let parquet_field_ids = HashMap::new();
let result = build_read_schema_with_field_id_mapping(&columns, &parquet_field_ids, None);
assert!(result.is_err(), "column_id > i32::MAX should fail");
match result {
Err(DuckLakeError::Internal(msg)) => {
assert!(
msg.contains("2147483648"),
"Error should contain the overflowing value: {}",
msg
);
assert!(
msg.contains("exceeds i32 range"),
"Error should explain the issue: {}",
msg
);
},
_ => panic!("Expected Internal error for column_id overflow"),
}
}
#[test]
fn test_column_id_negative_within_i32_range_succeeds() {
let columns = vec![DuckLakeTableColumn {
column_id: -1,
column_name: "id".to_string(),
column_type: "int32".to_string(),
is_nullable: true,
data_type: None,
nested_column_ids: Vec::new(),
}];
let mut parquet_field_ids = HashMap::new();
parquet_field_ids.insert(-1_i32, "id".to_string());
let result = build_read_schema_with_field_id_mapping(&columns, &parquet_field_ids, None);
assert!(
result.is_ok(),
"Negative column_id within i32 range should succeed"
);
}
#[test]
fn test_normalize_int_aliases() {
assert_eq!(normalize_ducklake_type("int").unwrap(), "int32");
assert_eq!(normalize_ducklake_type("integer").unwrap(), "int32");
assert_eq!(normalize_ducklake_type("INT").unwrap(), "int32");
assert_eq!(normalize_ducklake_type("Integer").unwrap(), "int32");
assert_eq!(normalize_ducklake_type("int32").unwrap(), "int32");
}
#[test]
fn test_normalize_bigint_aliases() {
assert_eq!(normalize_ducklake_type("bigint").unwrap(), "int64");
assert_eq!(normalize_ducklake_type("long").unwrap(), "int64");
assert_eq!(normalize_ducklake_type("BIGINT").unwrap(), "int64");
assert_eq!(normalize_ducklake_type("int64").unwrap(), "int64");
}
#[test]
fn test_normalize_string_aliases() {
assert_eq!(normalize_ducklake_type("text").unwrap(), "varchar");
assert_eq!(normalize_ducklake_type("string").unwrap(), "varchar");
assert_eq!(normalize_ducklake_type("varchar").unwrap(), "varchar");
assert_eq!(normalize_ducklake_type("TEXT").unwrap(), "varchar");
assert_eq!(normalize_ducklake_type("STRING").unwrap(), "varchar");
}
#[test]
fn test_normalize_float_aliases() {
assert_eq!(normalize_ducklake_type("float").unwrap(), "float32");
assert_eq!(normalize_ducklake_type("real").unwrap(), "float32");
assert_eq!(normalize_ducklake_type("FLOAT").unwrap(), "float32");
assert_eq!(normalize_ducklake_type("float32").unwrap(), "float32");
}
#[test]
fn test_normalize_double_aliases() {
assert_eq!(normalize_ducklake_type("double").unwrap(), "float64");
assert_eq!(normalize_ducklake_type("DOUBLE").unwrap(), "float64");
assert_eq!(normalize_ducklake_type("float64").unwrap(), "float64");
}
#[test]
fn test_normalize_bool_aliases() {
assert_eq!(normalize_ducklake_type("bool").unwrap(), "boolean");
assert_eq!(normalize_ducklake_type("boolean").unwrap(), "boolean");
assert_eq!(normalize_ducklake_type("BOOLEAN").unwrap(), "boolean");
}
#[test]
fn test_normalize_smallint_aliases() {
assert_eq!(normalize_ducklake_type("smallint").unwrap(), "int16");
assert_eq!(normalize_ducklake_type("SMALLINT").unwrap(), "int16");
assert_eq!(normalize_ducklake_type("int16").unwrap(), "int16");
}
#[test]
fn test_normalize_tinyint_aliases() {
assert_eq!(normalize_ducklake_type("tinyint").unwrap(), "int8");
assert_eq!(normalize_ducklake_type("TINYINT").unwrap(), "int8");
assert_eq!(normalize_ducklake_type("int8").unwrap(), "int8");
}
#[test]
fn test_normalize_unknown_type_errors() {
assert!(normalize_ducklake_type("foobar").is_err());
}
#[test]
fn test_promotable_same_type() {
assert!(is_promotable("int32", "int32"));
assert!(is_promotable("varchar", "varchar"));
assert!(is_promotable("float64", "float64"));
}
#[test]
fn test_promotable_signed_int_widening() {
assert!(is_promotable("int8", "int16"));
assert!(is_promotable("int8", "int32"));
assert!(is_promotable("int8", "int64"));
assert!(is_promotable("int16", "int32"));
assert!(is_promotable("int16", "int64"));
assert!(is_promotable("int32", "int64"));
}
#[test]
fn test_promotable_signed_int_narrowing_rejected() {
assert!(!is_promotable("int64", "int32"));
assert!(!is_promotable("int32", "int16"));
assert!(!is_promotable("int16", "int8"));
}
#[test]
fn test_promotable_unsigned_int_widening() {
assert!(is_promotable("uint8", "uint16"));
assert!(is_promotable("uint8", "uint32"));
assert!(is_promotable("uint8", "uint64"));
assert!(is_promotable("uint16", "uint32"));
assert!(is_promotable("uint32", "uint64"));
}
#[test]
fn test_promotable_unsigned_narrowing_rejected() {
assert!(!is_promotable("uint64", "uint32"));
assert!(!is_promotable("uint32", "uint16"));
}
#[test]
fn test_promotable_float_widening() {
assert!(is_promotable("float32", "float64"));
}
#[test]
fn test_promotable_float_narrowing_rejected() {
assert!(!is_promotable("float64", "float32"));
}
#[test]
fn test_promotable_int_to_float_excluded() {
assert!(!is_promotable("int8", "float64"));
assert!(!is_promotable("int16", "float64"));
assert!(!is_promotable("int32", "float64"));
assert!(!is_promotable("int64", "float64"));
assert!(!is_promotable("int32", "float32"));
}
#[test]
fn test_promotable_timestamp_to_timestamptz_excluded() {
assert!(!is_promotable("timestamp", "timestamptz"));
assert!(!is_promotable("timestamptz", "timestamp"));
}
#[test]
fn test_promotable_decimal_excluded() {
assert!(!is_promotable("decimal(10, 2)", "decimal(18, 4)"));
assert!(!is_promotable("decimal(10, 2)", "decimal(20, 2)"));
assert!(!is_promotable("decimal(18, 4)", "decimal(10, 2)")); assert!(is_promotable("decimal(10, 2)", "decimal(10, 2)"));
}
#[test]
fn test_promotable_incompatible_types() {
assert!(!is_promotable("int32", "varchar"));
assert!(!is_promotable("varchar", "int32"));
assert!(!is_promotable("boolean", "int32"));
assert!(!is_promotable("date", "timestamp"));
}
#[test]
fn test_promotable_unknown_types() {
assert!(!is_promotable("foobar", "int32"));
assert!(!is_promotable("int32", "foobar"));
}
#[test]
fn test_promotable_with_aliases() {
assert!(is_promotable("int", "bigint")); assert!(is_promotable("tinyint", "integer")); assert!(is_promotable("float", "double")); }
#[test]
fn test_types_compatible_same_canonical() {
assert!(types_compatible("int", "int32"));
assert!(types_compatible("int32", "int"));
assert!(types_compatible("integer", "int"));
assert!(types_compatible("text", "varchar"));
assert!(types_compatible("string", "text"));
assert!(types_compatible("bigint", "int64"));
assert!(types_compatible("float", "real"));
assert!(types_compatible("double", "float64"));
assert!(types_compatible("bool", "boolean"));
}
#[test]
fn test_types_compatible_case_insensitive() {
assert!(types_compatible("INT", "int32"));
assert!(types_compatible("VARCHAR", "text"));
assert!(types_compatible("BIGINT", "int64"));
}
#[test]
fn test_types_compatible_with_promotion() {
assert!(types_compatible("int32", "int64"));
assert!(types_compatible("float32", "float64"));
assert!(!types_compatible("timestamp", "timestamptz"));
}
#[test]
fn test_types_equal_canonical() {
assert!(types_equal_canonical("int64", "bigint"));
assert!(types_equal_canonical("bigint", "int64"));
assert!(types_equal_canonical("int", "int32"));
assert!(types_equal_canonical("text", "varchar"));
assert!(types_equal_canonical("INT64", "int64")); assert!(!types_equal_canonical("int32", "int64"));
assert!(!types_equal_canonical("float32", "float64"));
assert!(!types_equal_canonical("int32", "varchar"));
assert!(!types_equal_canonical("foobar", "int32"));
}
#[test]
fn test_types_compatible_narrowing_rejected() {
assert!(!types_compatible("int64", "int32"));
assert!(!types_compatible("float64", "float32"));
}
#[test]
fn test_types_compatible_incompatible() {
assert!(!types_compatible("int32", "varchar"));
assert!(!types_compatible("varchar", "int32"));
assert!(!types_compatible("boolean", "float64"));
}
#[test]
fn test_types_compatible_unknown() {
assert!(!types_compatible("foobar", "int32"));
assert!(!types_compatible("int32", "foobar"));
assert!(!types_compatible("foobar", "bazqux"));
}
#[test]
fn test_read_schema_relaxes_nested_nullability_like_catalog_schema() {
let columns = vec![DuckLakeTableColumn {
column_id: 1,
column_name: "s".to_string(),
column_type: "struct<a:int32,b:int32>".to_string(),
is_nullable: true,
data_type: Some(DataType::Struct(
vec![
Arc::new(Field::new("a", DataType::Int32, true)),
Arc::new(Field::new("b", DataType::Int32, false)),
]
.into(),
)),
nested_column_ids: vec![2, 3],
}];
let parquet_field_ids =
HashMap::from([(1, "s".to_string()), (2, "a".to_string()), (3, "b".to_string())]);
let (read_schema, _) =
build_read_schema_with_field_id_mapping(&columns, &parquet_field_ids, None).unwrap();
let DataType::Struct(fields) = read_schema.field(0).data_type() else {
panic!("s must remain a struct");
};
assert!(fields[1].is_nullable(), "child `b` must read as nullable");
let catalog_schema = build_arrow_schema(&columns).unwrap();
assert!(
crate::column_rename::types_equal_ignoring_field_metadata(
read_schema.field(0).data_type(),
catalog_schema.field(0).data_type(),
),
"read {:?} != catalog {:?}",
read_schema.field(0).data_type(),
catalog_schema.field(0).data_type(),
);
}
#[test]
fn test_read_schema_keeps_map_key_non_nullable() {
let entries = Field::new(
"entries",
DataType::Struct(
vec![
Arc::new(Field::new("key", DataType::Utf8, false)),
Arc::new(Field::new("value", DataType::Int32, true)),
]
.into(),
),
false,
);
let columns = vec![DuckLakeTableColumn {
column_id: 1,
column_name: "m".to_string(),
column_type: "map<varchar,int32>".to_string(),
is_nullable: true,
data_type: Some(DataType::Map(Arc::new(entries), false)),
nested_column_ids: vec![2, 3],
}];
let parquet_field_ids =
HashMap::from([(1, "m".to_string()), (2, "key".to_string()), (3, "value".to_string())]);
let (read_schema, _) =
build_read_schema_with_field_id_mapping(&columns, &parquet_field_ids, None).unwrap();
let DataType::Map(entries, _) = read_schema.field(0).data_type() else {
panic!("m must remain a map");
};
let DataType::Struct(fields) = entries.data_type() else {
panic!("map entries must be a struct");
};
assert!(!fields[0].is_nullable(), "map key must stay non-nullable");
}
}