// SPDX-License-Identifier: Apache-2.0

// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership

// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements

//! DFSchema is an extended schema struct that DataFusion uses to provide support for

//! fields with optional relation names.

use std::collections::{BTreeSet, HashSet};

use std::fmt::Write as _;

use std::fmt::{Display, Formatter};

use std::hash::{BuildHasher, Hash};

use std::sync::{Arc, LazyLock, OnceLock};

use crate::error::{_plan_err, _schema_err, DataFusionError, Result};

use crate::hash_utils::RandomState;

use crate::{

Column, FunctionalDependencies, SchemaError, TableReference, field_not_found,

unqualified_field_not_found,

};

use arrow::compute::can_cast_types;

use arrow::datatypes::{

DataType, Field, FieldRef, Fields, Metadata, Schema, SchemaBuilder, SchemaRef,

};

use hashbrown::hash_table::{Entry, HashTable};

/// A reference-counted reference to a [DFSchema].

pub type DFSchemaRef = Arc<DFSchema>;

/// DFSchema wraps an Arrow schema and add a relation (table) name.

///

/// The schema may hold the fields across multiple tables. Some fields may be

/// qualified and some unqualified. A qualified field is a field that has a

/// relation name associated with it.

///

/// Unqualified fields must be unique not only amongst themselves, but also must

/// have a distinct name from any qualified field names. This allows finding a

/// qualified field by name to be possible, so long as there aren't multiple

/// qualified fields with the same name.

///]

/// # See Also

/// * [DFSchemaRef], an alias to `Arc<DFSchema>`

/// * [DataTypeExt], common methods for working with Arrow [DataType]s

/// * [FieldExt], extension methods for working with Arrow [Field]s

///

/// [DataTypeExt]: crate::datatype::DataTypeExt

/// [FieldExt]: crate::datatype::FieldExt

///

/// # Creating qualified schemas

///

/// Use [DFSchema::try_from_qualified_schema] to create a qualified schema from

/// an Arrow schema.

///

/// ```rust

/// use arrow::datatypes::{DataType, Field, Schema};

/// use datafusion_common::{Column, DFSchema};

///

/// let arrow_schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);

///

/// let df_schema = DFSchema::try_from_qualified_schema("t1", &arrow_schema).unwrap();

/// let column = Column::from_qualified_name("t1.c1");

/// assert!(df_schema.has_column(&column));

///

/// // Can also access qualified fields with unqualified name, if it's unambiguous

/// let column = Column::from_qualified_name("c1");

/// assert!(df_schema.has_column(&column));

/// ```

///

/// # Creating unqualified schemas

///

/// Create an unqualified schema using TryFrom:

///

/// ```rust

/// use arrow::datatypes::{DataType, Field, Schema};

/// use datafusion_common::{Column, DFSchema};

///

/// let arrow_schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);

///

/// let df_schema = DFSchema::try_from(arrow_schema).unwrap();

/// let column = Column::new_unqualified("c1");

/// assert!(df_schema.has_column(&column));

/// ```

///

/// # Converting back to Arrow schema

///

/// Use the `Into` trait to convert `DFSchema` into an Arrow schema:

///

/// ```rust

/// use arrow::datatypes::{Field, Schema};

/// use datafusion_common::DFSchema;

/// use std::collections::HashMap;

///

/// let df_schema = DFSchema::from_unqualified_fields(

/// vec![Field::new("c1", arrow::datatypes::DataType::Int32, false)].into(),

/// HashMap::new(),

/// )

/// .unwrap();

/// let schema: &Schema = df_schema.as_arrow();

/// assert_eq!(schema.fields().len(), 1);

/// ```

#[derive(Clone)]

pub struct DFSchema {

/// Inner Arrow schema reference.

inner: SchemaRef,

/// Optional qualifiers for each column in this schema. In the same order as

/// the `self.inner.fields()`

field_qualifiers: Vec<Option<TableReference>>,

/// Stores functional dependencies in the schema.

functional_dependencies: FunctionalDependencies,

/// Lazily built lookup from field name to positions in `inner`, only used

/// for schemas with more than [`LINEAR_NAME_SCAN_MAX`] fields.

///

/// Derived from the field names alone, so it is shared by clones and must

/// be reset whenever `inner` changes. It takes no part in equality,

/// hashing or `Debug` output.

name_index: OnceLock<Arc<NameIndex>>,

}

impl DFSchema {

/// Creates an empty `DFSchema`

pub fn empty() -> Self {

Self {

inner: Arc::new(Schema::new([])),

field_qualifiers: vec![],

functional_dependencies: FunctionalDependencies::empty(),

name_index: OnceLock::new(),

}

}

/// Returns a reference to a shared empty [`DFSchema`].

pub fn empty_ref() -> &'static DFSchemaRef {

static EMPTY: LazyLock<DFSchemaRef> =

LazyLock::new(|| Arc::new(DFSchema::empty()));

&EMPTY

}

/// Return a reference to the inner Arrow [`Schema`]

///

/// Note this does not have the qualifier information

pub fn as_arrow(&self) -> &Schema {

self.inner.as_ref()

}

/// Return a reference to the inner Arrow [`SchemaRef`]

///

/// Note this does not have the qualifier information

pub fn inner(&self) -> &SchemaRef {

&self.inner

}

/// Create a `DFSchema` from an Arrow schema where all the fields have a given qualifier

pub fn new_with_metadata(

qualified_fields: Vec<(Option<TableReference>, Arc<Field>)>,

metadata: impl Into<Metadata>,

) -> Result<Self> {

let (qualifiers, fields): (Vec<Option<TableReference>>, Vec<Arc<Field>>) =

qualified_fields.into_iter().unzip();

let schema = Arc::new(Schema::new_with_metadata(fields, metadata));

let dfschema = Self {

inner: schema,

field_qualifiers: qualifiers,

functional_dependencies: FunctionalDependencies::empty(),

name_index: OnceLock::new(),

};

dfschema.check_names()?;

Ok(dfschema)

}

/// Create a new `DFSchema` from a list of Arrow [Field]s

pub fn from_unqualified_fields(

fields: Fields,

metadata: impl Into<Metadata>,

) -> Result<Self> {

let field_count = fields.len();

let schema = Arc::new(Schema::new_with_metadata(fields, metadata));

let dfschema = Self {

inner: schema,

field_qualifiers: vec![None; field_count],

functional_dependencies: FunctionalDependencies::empty(),

name_index: OnceLock::new(),

};

dfschema.check_names()?;

Ok(dfschema)

}

/// Create a `DFSchema` from an Arrow schema and a given qualifier

///

/// To create a schema from an Arrow schema without a qualifier, use

/// `DFSchema::try_from`.

pub fn try_from_qualified_schema(

qualifier: impl Into<TableReference>,

schema: &Schema,

) -> Result<Self> {

let qualifier = qualifier.into();

let schema = DFSchema {

inner: schema.clone().into(),

field_qualifiers: vec![Some(qualifier); schema.fields.len()],

functional_dependencies: FunctionalDependencies::empty(),

name_index: OnceLock::new(),

};

schema.check_names()?;

Ok(schema)

}

/// Create a `DFSchema` from an Arrow schema where all the fields have a given qualifier

pub fn from_field_specific_qualified_schema(

qualifiers: Vec<Option<TableReference>>,

schema: &SchemaRef,

) -> Result<Self> {

let dfschema = Self {

inner: Arc::clone(schema),

field_qualifiers: qualifiers,

functional_dependencies: FunctionalDependencies::empty(),

name_index: OnceLock::new(),

};

dfschema.check_names()?;

Ok(dfschema)

}

/// Return the same schema, where all fields have a given qualifier.

pub fn with_field_specific_qualified_schema(

&self,

qualifiers: Vec<Option<TableReference>>,

) -> Result<Self> {

if qualifiers.len() != self.fields().len() {

return _plan_err!(

"Number of qualifiers must match number of fields. Expected {}, got {}",

self.fields().len(),

qualifiers.len()

);

}

Ok(DFSchema {

inner: Arc::clone(&self.inner),

field_qualifiers: qualifiers,

functional_dependencies: self.functional_dependencies.clone(),

name_index: OnceLock::new(),

})

}

/// Check if the schema have some fields with the same name

pub fn check_names(&self) -> Result<()> {

let mut qualified_names = BTreeSet::new();

let mut unqualified_names = BTreeSet::new();

for (field, qualifier) in self.inner.fields().iter().zip(&self.field_qualifiers) {

if let Some(qualifier) = qualifier {

if !qualified_names.insert((qualifier, field.name())) {

return _schema_err!(SchemaError::DuplicateQualifiedField {

qualifier: Box::new(qualifier.clone()),

name: field.name().to_string(),

});

}

} else if !unqualified_names.insert(field.name()) {

return _schema_err!(SchemaError::DuplicateUnqualifiedField {

name: field.name().to_string()

});

}

}

for (qualifier, name) in qualified_names {

if unqualified_names.contains(name) {

return _schema_err!(SchemaError::AmbiguousReference {

field: Box::new(Column::new(Some(qualifier.clone()), name))

});

}

}

Ok(())

}

/// Assigns functional dependencies.

pub fn with_functional_dependencies(

mut self,

functional_dependencies: FunctionalDependencies,

) -> Result<Self> {

if functional_dependencies.is_valid(self.inner.fields.len()) {

self.functional_dependencies = functional_dependencies;

Ok(self)

} else {

_plan_err!(

"Invalid functional dependency: {:?}",

functional_dependencies

)

}

}

/// Create a new schema that contains the fields from this schema followed by the fields

/// from the supplied schema. An error will be returned if there are duplicate field names.

pub fn join(&self, schema: &DFSchema) -> Result<Self> {

let mut schema_builder = SchemaBuilder::with_capacity(

self.inner.fields().len() + schema.fields().len(),

);

schema_builder.extend(self.inner.fields().iter().cloned());

schema_builder.extend(schema.fields().iter().cloned());

let new_schema = schema_builder.finish();

let mut new_metadata = self.inner.metadata.clone();

new_metadata.extend(schema.inner.metadata.clone());

let new_schema_with_metadata = new_schema.with_metadata(new_metadata);

let mut new_qualifiers = self.field_qualifiers.clone();

new_qualifiers.extend_from_slice(schema.field_qualifiers.as_slice());

let new_self = Self {

inner: Arc::new(new_schema_with_metadata),

field_qualifiers: new_qualifiers,

functional_dependencies: FunctionalDependencies::empty(),

name_index: OnceLock::new(),

};

new_self.check_names()?;

Ok(new_self)

}

/// Modify this schema by appending the fields from the supplied schema, ignoring any

/// duplicate fields.

///

/// ## Merge Precedence

///

/// **Schema-level metadata**: Metadata from both schemas is merged.

/// If both schemas have the same metadata key, the value from the `other_schema` parameter takes precedence.

///

/// **Field-level merging**: Only non-duplicate fields are added. This means that the

/// `self` fields will always take precedence over the `other_schema` fields.

/// Duplicate field detection is based on:

/// - For qualified fields: both qualifier and field name must match

/// - For unqualified fields: only field name needs to match

///

/// Take note how the precedence for fields & metadata merging differs;

/// merging prefers fields from `self` but prefers metadata from `other_schema`.

pub fn merge(&mut self, other_schema: &DFSchema) {

if other_schema.inner.fields.is_empty() {

return;

}

let self_fields: HashSet<(Option<&TableReference>, &FieldRef)> =

self.iter().collect();

let self_unqualified_names: HashSet<&str> = self

.inner

.fields

.iter()

.map(|field| field.name().as_str())

.collect();

let mut schema_builder = SchemaBuilder::from(self.inner.fields.clone());

let mut qualifiers = Vec::with_capacity(other_schema.inner.fields.len());

for (qualifier, field) in other_schema.iter() {

// skip duplicate columns

let duplicated_field = match qualifier {

Some(q) => self_fields.contains(&(Some(q), field)),

// for unqualified columns, check as unqualified name

None => self_unqualified_names.contains(field.name().as_str()),

};

if !duplicated_field {

schema_builder.push(Arc::clone(field));

qualifiers.push(qualifier.cloned());

}

}

let mut metadata = self.inner.metadata.clone();

metadata.extend(other_schema.inner.metadata.clone());

let finished = schema_builder.finish();

let finished_with_metadata = finished.with_metadata(metadata);

self.inner = finished_with_metadata.into();

self.field_qualifiers.extend(qualifiers);

self.name_index = OnceLock::new();

}

/// Get a list of fields for this schema

pub fn fields(&self) -> &Fields {

&self.inner.fields

}

/// Returns a reference to [`FieldRef`] for a column at specific index

/// within the schema.

///

/// See also [Self::qualified_field] to get both qualifier and field

pub fn field(&self, i: usize) -> &FieldRef {

&self.inner.fields[i]

}

/// Returns the qualifier (if any) and [`FieldRef`] for a column at specific

/// index within the schema.

pub fn qualified_field(&self, i: usize) -> (Option<&TableReference>, &FieldRef) {

(self.field_qualifiers[i].as_ref(), self.field(i))

}

/// Returns the indices of the fields named `name`, in ascending order.

///

/// Narrow schemas are scanned; wider ones consult [`Self::name_index`].

fn indices_of_name<'a>(&'a self, name: &'a str) -> NameIndices<'a> {

let fields = self.fields();

match self.name_index() {

Some(index) => NameIndices::Indexed {

next: index.first_index_of(fields, name),

index,

},

None => NameIndices::Scan {

fields: fields.iter().enumerate(),

name,

},

}

}

/// Returns the name index, building it on first use, or `None` if the

/// schema is too narrow for an index to beat a scan.

fn name_index(&self) -> Option<&NameIndex> {

let fields = self.fields();

if fields.len() <= LINEAR_NAME_SCAN_MAX {

return None;

}

Some(

self.name_index

.get_or_init(|| Arc::new(NameIndex::new(fields))),

)

}

pub fn index_of_column_by_name(

&self,

qualifier: Option<&TableReference>,

name: &str,

) -> Option<usize> {

self.indices_of_name(name).find(|&idx| {

match (qualifier, &self.field_qualifiers[idx]) {

// field to lookup is qualified.

// current field is qualified and not shared between relations, compare both

// qualifier and name.

(Some(q), Some(field_q)) => q.resolved_eq(field_q),

// field to lookup is qualified but current field is unqualified.

(Some(_), None) => false,

// field to lookup is unqualified, no need to compare qualifier

(None, Some(_)) | (None, None) => true,

}

})

}

/// Find the index of the column with the given qualifier and name,

/// returning `None` if not found

///

/// See [Self::index_of_column] for a version that returns an error if the

/// column is not found

pub fn maybe_index_of_column(&self, col: &Column) -> Option<usize> {

self.index_of_column_by_name(col.relation.as_ref(), &col.name)

}

/// Find the index of the column with the given qualifier and name,

/// returning `Err` if not found

///

/// See [Self::maybe_index_of_column] for a version that returns `None` if

/// the column is not found

pub fn index_of_column(&self, col: &Column) -> Result<usize> {

self.maybe_index_of_column(col)

.ok_or_else(|| field_not_found(col.relation.clone(), &col.name, self))

}

/// Check if the column is in the current schema

pub fn is_column_from_schema(&self, col: &Column) -> bool {

self.index_of_column_by_name(col.relation.as_ref(), &col.name)

.is_some()

}

/// Find the [`FieldRef`] with the given name and optional qualifier

pub fn field_with_name(

&self,

qualifier: Option<&TableReference>,

name: &str,

) -> Result<&FieldRef> {

if let Some(qualifier) = qualifier {

self.field_with_qualified_name(qualifier, name)

} else {

self.field_with_unqualified_name(name)

}

}

/// Find the qualified field with the given name

pub fn qualified_field_with_name(

&self,

qualifier: Option<&TableReference>,

name: &str,

) -> Result<(Option<&TableReference>, &FieldRef)> {

if let Some(qualifier) = qualifier {

let idx = self

.index_of_column_by_name(Some(qualifier), name)

.ok_or_else(|| field_not_found(Some(qualifier.clone()), name, self))?;

Ok((self.field_qualifiers[idx].as_ref(), self.field(idx)))

} else {

self.qualified_field_with_unqualified_name(name)

}

}

/// Find all fields having the given qualifier

pub fn fields_with_qualified(&self, qualifier: &TableReference) -> Vec<&FieldRef> {

self.iter()

.filter(|(q, _)| q.map(|q| q.eq(qualifier)).unwrap_or(false))

.map(|(_, f)| f)

.collect()

}

/// Find all fields indices having the given qualifier

pub fn fields_indices_with_qualified(

&self,

qualifier: &TableReference,

) -> Vec<usize> {

self.iter()

.enumerate()

.filter_map(|(idx, (q, _))| q.and_then(|q| q.eq(qualifier).then_some(idx)))

.collect()

}

/// Find all fields that match the given name

pub fn fields_with_unqualified_name(&self, name: &str) -> Vec<&FieldRef> {

self.indices_of_name(name)

.map(|idx| self.field(idx))

.collect()

}

/// Find all fields that match the given name and return them with their qualifier

pub fn qualified_fields_with_unqualified_name(

&self,

name: &str,

) -> Vec<(Option<&TableReference>, &FieldRef)> {

self.indices_of_name(name)

.map(|idx| self.qualified_field(idx))

.collect()

}

/// Find all fields that match the given name and convert to column

pub fn columns_with_unqualified_name(&self, name: &str) -> Vec<Column> {

self.indices_of_name(name)

.map(|idx| {

let (qualifier, field) = self.qualified_field(idx);

Column::new(qualifier.cloned(), field.name())

})

.collect()

}

/// Return all `Column`s for the schema

pub fn columns(&self) -> Vec<Column> {

self.iter()

.map(|(qualifier, field)| {

Column::new(qualifier.cloned(), field.name().clone())

})

.collect()

}

/// Find the qualified field with the given unqualified name

pub fn qualified_field_with_unqualified_name(

&self,

name: &str,

) -> Result<(Option<&TableReference>, &FieldRef)> {

let matches = self.qualified_fields_with_unqualified_name(name);

match matches.len() {

0 => Err(unqualified_field_not_found(name, self)),

1 => Ok((matches[0].0, matches[0].1)),

_ => {

// When `matches` size > 1, it doesn't necessarily mean an `ambiguous name` problem.

// Because name may generate from Alias/... . It means that it don't own qualifier.

// For example:

// Join on id = b.id

// Project a.id as id TableScan b id

// In this case, there isn't `ambiguous name` problem. When `matches` just contains

// one field without qualifier, we should return it.

let fields_without_qualifier = matches

.iter()

.filter(|(q, _)| q.is_none())

.collect::<Vec<_>>();

if fields_without_qualifier.len() == 1 {

Ok((fields_without_qualifier[0].0, fields_without_qualifier[0].1))

} else {

_schema_err!(SchemaError::AmbiguousReference {

field: Box::new(Column::new_unqualified(name.to_string()))

})

}

}

}

}

/// Find the field with the given name

pub fn field_with_unqualified_name(&self, name: &str) -> Result<&FieldRef> {

self.qualified_field_with_unqualified_name(name)

.map(|(_, field)| field)

}

/// Find the field with the given qualified name

pub fn field_with_qualified_name(

&self,

qualifier: &TableReference,

name: &str,

) -> Result<&FieldRef> {

let idx = self

.index_of_column_by_name(Some(qualifier), name)

.ok_or_else(|| field_not_found(Some(qualifier.clone()), name, self))?;

Ok(self.field(idx))

}

/// Find the field with the given qualified column

pub fn qualified_field_from_column(

&self,

column: &Column,

) -> Result<(Option<&TableReference>, &FieldRef)> {

self.qualified_field_with_name(column.relation.as_ref(), &column.name)

}

/// Find if the field exists with the given name

pub fn has_column_with_unqualified_name(&self, name: &str) -> bool {

self.indices_of_name(name).next().is_some()

}

/// Find if the field exists with the given qualified name

pub fn has_column_with_qualified_name(

&self,

qualifier: &TableReference,

name: &str,

) -> bool {

self.indices_of_name(name).any(|idx| {

self.field_qualifiers[idx]

.as_ref()

.is_some_and(|q| q.eq(qualifier))

})

}

/// Find if the field exists with the given qualified column

pub fn has_column(&self, column: &Column) -> bool {

match &column.relation {

Some(r) => self.has_column_with_qualified_name(r, &column.name),

None => self.has_column_with_unqualified_name(&column.name),

}

}

/// Check to see if unqualified field names matches field names in Arrow schema

pub fn matches_arrow_schema(&self, arrow_schema: &Schema) -> bool {

self.inner

.fields

.iter()

.zip(arrow_schema.fields().iter())

.all(|(dffield, arrowfield)| dffield.name() == arrowfield.name())

}

/// Returns true if the two schemas have the same qualified named

/// fields with logically equivalent data types. Returns false otherwise.

///

/// Use [DFSchema]::equivalent_names_and_types for stricter semantic type

/// equivalence checking.

pub fn logically_equivalent_names_and_types(&self, other: &Self) -> bool {

if self.fields().len() != other.fields().len() {

return false;

}

let self_fields = self.iter();

let other_fields = other.iter();

self_fields.zip(other_fields).all(|((q1, f1), (q2, f2))| {

q1 == q2

&& f1.name() == f2.name()

&& Self::datatype_is_logically_equal(f1.data_type(), f2.data_type())

})

}

/// Returns Ok if the two schemas have the same qualified named

/// fields with the compatible data types.

///

/// Returns an `Err` with a message otherwise.

///

/// This is a specialized version of Eq that ignores differences in

/// nullability and metadata.

///

/// Use [DFSchema]::logically_equivalent_names_and_types for a weaker

/// logical type checking, which for example would consider a dictionary

/// encoded UTF8 array to be equivalent to a plain UTF8 array.

pub fn has_equivalent_names_and_types(&self, other: &Self) -> Result<()> {

// case 1 : schema length mismatch

if self.fields().len() != other.fields().len() {

_plan_err!(

"Schema mismatch: the schema length are not same \

Expected schema length: {}, got: {}",

self.fields().len(),

other.fields().len()

)

} else {

// case 2 : schema length match, but fields mismatch

// check if the fields name are the same and have the same data types

self.fields()

.iter()

.zip(other.fields().iter())

.try_for_each(|(f1, f2)| {

if f1.name() != f2.name()

|| (!DFSchema::datatype_is_semantically_equal(

f1.data_type(),

f2.data_type(),

))

{

_plan_err!(

"Schema mismatch: Expected field '{}' with type {}, \

but got '{}' with type {}.",

f1.name(),

f1.data_type(),

f2.name(),

f2.data_type()

)

} else {

Ok(())

}

})

}

}

/// Checks if two [`DataType`]s are logically equal. This is a notably weaker constraint

/// than datatype_is_semantically_equal in that different representations of same data can be

/// logically but not semantically equivalent. Semantically equivalent types are always also

/// logically equivalent. For example:

/// - a Dictionary<K,V> type is logically equal to a plain V type

/// - a Dictionary<K1, V1> is also logically equal to Dictionary<K2, V1>

/// - a RunEndEncoded<K,V> type is logically equal to a plain V type

/// - a RunEndEncoded<K1, V1> is also logically equal to RunEndEncoded<K2, V1>

/// - Utf8 and Utf8View are logically equal

pub fn datatype_is_logically_equal(dt1: &DataType, dt2: &DataType) -> bool {

// check nested fields

match (dt1, dt2) {

(DataType::Dictionary(_, v1), DataType::Dictionary(_, v2)) => {

Self::datatype_is_logically_equal(v1.as_ref(), v2.as_ref())

}

(DataType::Dictionary(_, v1), othertype)

| (othertype, DataType::Dictionary(_, v1)) => {

Self::datatype_is_logically_equal(v1.as_ref(), othertype)

}

(DataType::RunEndEncoded(_, v1), DataType::RunEndEncoded(_, v2)) => {

Self::datatype_is_logically_equal(v1.data_type(), v2.data_type())

}

(DataType::RunEndEncoded(_, v1), othertype)

| (othertype, DataType::RunEndEncoded(_, v1)) => {

Self::datatype_is_logically_equal(v1.data_type(), othertype)

}

(DataType::List(f1), DataType::List(f2))

| (DataType::LargeList(f1), DataType::LargeList(f2))

| (DataType::ListView(f1), DataType::ListView(f2))

| (DataType::LargeListView(f1), DataType::LargeListView(f2))

| (DataType::FixedSizeList(f1, _), DataType::FixedSizeList(f2, _)) => {

// Don't compare the names of the technical inner field

// Usually "item" but that's not mandated

Self::datatype_is_logically_equal(f1.data_type(), f2.data_type())

}

(DataType::Map(f1, _), DataType::Map(f2, _)) => {

// Don't compare the names of the technical inner fields

// Usually "entries", "key", "value" but that's not mandated

match (f1.data_type(), f2.data_type()) {

(DataType::Struct(f1_inner), DataType::Struct(f2_inner)) => {

f1_inner.len() == f2_inner.len()

&& f1_inner.iter().zip(f2_inner.iter()).all(|(f1, f2)| {

Self::datatype_is_logically_equal(

f1.data_type(),

f2.data_type(),

)

})

}

_ => panic!("Map type should have an inner struct field"),

}

}

(DataType::Struct(fields1), DataType::Struct(fields2)) => {

let iter1 = fields1.iter();

let iter2 = fields2.iter();

fields1.len() == fields2.len() &&

// all fields have to be the same

iter1

.zip(iter2)

.all(|(f1, f2)| Self::field_is_logically_equal(f1, f2))

}

(DataType::Union(fields1, _), DataType::Union(fields2, _)) => {

let iter1 = fields1.iter();

let iter2 = fields2.iter();

fields1.len() == fields2.len() &&

// all fields have to be the same

iter1

.zip(iter2)

.all(|((t1, f1), (t2, f2))| t1 == t2 && Self::field_is_logically_equal(f1, f2))

}

// Utf8 and Utf8View are logically equivalent

(DataType::Utf8, DataType::Utf8View) => true,

(DataType::Utf8View, DataType::Utf8) => true,

_ => Self::datatype_is_semantically_equal(dt1, dt2),

}

}

/// Returns true of two [`DataType`]s are semantically equal (same

/// name and type), ignoring both metadata and nullability, decimal precision/scale,

/// and timezone time units/timezones.

///

/// request to upstream: <https://github.com/apache/arrow-rs/issues/3199>

pub fn datatype_is_semantically_equal(dt1: &DataType, dt2: &DataType) -> bool {

// check nested fields

match (dt1, dt2) {

(DataType::Dictionary(k1, v1), DataType::Dictionary(k2, v2)) => {

Self::datatype_is_semantically_equal(k1.as_ref(), k2.as_ref())

&& Self::datatype_is_semantically_equal(v1.as_ref(), v2.as_ref())

}

(DataType::RunEndEncoded(k1, v1), DataType::RunEndEncoded(k2, v2)) => {

Self::datatype_is_semantically_equal(k1.data_type(), k2.data_type())

&& Self::datatype_is_semantically_equal(

v1.data_type(),

v2.data_type(),

)

}

(DataType::List(f1), DataType::List(f2))

| (DataType::LargeList(f1), DataType::LargeList(f2))

| (DataType::ListView(f1), DataType::ListView(f2))

| (DataType::LargeListView(f1), DataType::LargeListView(f2))

| (DataType::FixedSizeList(f1, _), DataType::FixedSizeList(f2, _)) => {

// Don't compare the names of the technical inner field

// Usually "item" but that's not mandated

Self::datatype_is_semantically_equal(f1.data_type(), f2.data_type())

}

(DataType::Map(f1, _), DataType::Map(f2, _)) => {

// Don't compare the names of the technical inner fields

// Usually "entries", "key", "value" but that's not mandated

match (f1.data_type(), f2.data_type()) {

(DataType::Struct(f1_inner), DataType::Struct(f2_inner)) => {

f1_inner.len() == f2_inner.len()

&& f1_inner.iter().zip(f2_inner.iter()).all(|(f1, f2)| {

Self::datatype_is_semantically_equal(

f1.data_type(),

f2.data_type(),

)

})

}

_ => panic!("Map type should have an inner struct field"),

}

}

(DataType::Struct(fields1), DataType::Struct(fields2)) => {

let iter1 = fields1.iter();

let iter2 = fields2.iter();

fields1.len() == fields2.len() &&

// all fields have to be the same

iter1

.zip(iter2)

.all(|(f1, f2)| Self::field_is_semantically_equal(f1, f2))

}

(DataType::Union(fields1, _), DataType::Union(fields2, _)) => {

let iter1 = fields1.iter();

let iter2 = fields2.iter();

fields1.len() == fields2.len() &&

// all fields have to be the same

iter1

.zip(iter2)

.all(|((t1, f1), (t2, f2))| t1 == t2 && Self::field_is_semantically_equal(f1, f2))

}

(

DataType::Decimal32(_l_precision, _l_scale),

DataType::Decimal32(_r_precision, _r_scale),

) => true,

(

DataType::Decimal64(_l_precision, _l_scale),

DataType::Decimal64(_r_precision, _r_scale),

) => true,

(

DataType::Decimal128(_l_precision, _l_scale),

DataType::Decimal128(_r_precision, _r_scale),

) => true,

(

DataType::Decimal256(_l_precision, _l_scale),

DataType::Decimal256(_r_precision, _r_scale),

) => true,

(

DataType::Timestamp(_l_time_unit, _l_timezone),

DataType::Timestamp(_r_time_unit, _r_timezone),

) => true,

_ => dt1 == dt2,

}

}

fn field_is_logically_equal(f1: &Field, f2: &Field) -> bool {

f1.name() == f2.name()

&& Self::datatype_is_logically_equal(f1.data_type(), f2.data_type())

}

fn field_is_semantically_equal(f1: &Field, f2: &Field) -> bool {

f1.name() == f2.name()

&& Self::datatype_is_semantically_equal(f1.data_type(), f2.data_type())

}

/// Strip all field qualifier in schema

pub fn strip_qualifiers(self) -> Self {

DFSchema {

field_qualifiers: vec![None; self.inner.fields.len()],

inner: self.inner,

functional_dependencies: self.functional_dependencies,

name_index: OnceLock::new(),

}

}

/// Replace all field qualifier with new value in schema

pub fn replace_qualifier(self, qualifier: impl Into<TableReference>) -> Self {

let qualifier = qualifier.into();

DFSchema {

field_qualifiers: vec![Some(qualifier); self.inner.fields.len()],

inner: self.inner,

functional_dependencies: self.functional_dependencies,

name_index: OnceLock::new(),

}

}

/// Get list of fully-qualified field names in this schema

pub fn field_names(&self) -> Vec<String> {

self.iter()

.map(|(qualifier, field)| qualified_name(qualifier, field.name()))

.collect::<Vec<_>>()

}

/// Get metadata of this schema

pub fn metadata(&self) -> &Metadata {

&self.inner.metadata

}

/// Get functional dependencies

pub fn functional_dependencies(&self) -> &FunctionalDependencies {

&self.functional_dependencies

}

/// Iterate over the qualifiers and fields in the DFSchema

pub fn iter(&self) -> impl Iterator<Item = (Option<&TableReference>, &FieldRef)> {

self.field_qualifiers

.iter()

.zip(self.inner.fields().iter())

.map(|(qualifier, field)| (qualifier.as_ref(), field))

}

/// Returns a tree-like string representation of the schema.

///

/// This method formats the schema

/// with a tree-like structure showing field names, types, and nullability.

///

/// # Example

///

/// ```

/// use arrow::datatypes::{DataType, Field, Schema};

/// use datafusion_common::DFSchema;

/// use std::collections::HashMap;

///

/// let schema = DFSchema::from_unqualified_fields(

/// vec![

/// Field::new("id", DataType::Int32, false),

/// Field::new("name", DataType::Utf8, true),

/// ]

/// .into(),

/// HashMap::new(),

/// )

/// .unwrap();

///

/// assert_eq!(

/// schema.tree_string().to_string(),

/// r#"root

/// |-- id: int32 (nullable = false)

/// |-- name: utf8 (nullable = true)"#

/// );

/// ```

pub fn tree_string(&self) -> impl Display + '_ {

let mut result = String::from("root\n");

for (qualifier, field) in self.iter() {

let field_name = match qualifier {

Some(q) => format!("{}.{}", q, field.name()),

None => field.name().to_string(),

};

format_field_with_indent(

&mut result,

&field_name,

field.data_type(),

field.is_nullable(),

" ",

);

}

// Remove the trailing newline

if result.ends_with('\n') {

result.pop();

}

result

}

}

/// Format field with proper nested indentation for complex types

fn format_field_with_indent(

result: &mut String,

field_name: &str,

data_type: &DataType,

nullable: bool,

indent: &str,

) {

let nullable_str = nullable.to_string().to_lowercase();

let child_indent = format!("{indent}| ");

match data_type {

DataType::List(field) => {

writeln!(

result,

"{indent}|-- {field_name}: list (nullable = {nullable_str})"

)

.ok();

format_field_with_indent(

result,

field.name(),

field.data_type(),