// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
//! Avro Schema representations for Arrow.
#[cfg(feature = "canonical_extension_types")]
use arrow_schema::extension::ExtensionType;
use arrow_schema::{
ArrowError, DataType, Field as ArrowField, IntervalUnit, Schema as ArrowSchema, TimeUnit,
UnionMode,
};
use serde::{Deserialize, Serialize};
use serde_json::{Map as JsonMap, Value, json};
#[cfg(feature = "sha256")]
use sha2::{Digest, Sha256};
use std::borrow::Cow;
use std::cmp::PartialEq;
use std::collections::hash_map::Entry;
use std::collections::{HashMap, HashSet};
use strum_macros::AsRefStr;
/// The Avro single‑object encoding “magic” bytes (`0xC3 0x01`)
pub const SINGLE_OBJECT_MAGIC: [u8; 2] = [0xC3, 0x01];
/// The Confluent "magic" byte (`0x00`)
pub const CONFLUENT_MAGIC: [u8; 1] = [0x00];
/// The maximum possible length of a prefix.
/// SHA256 (32) + single-object magic (2)
pub const MAX_PREFIX_LEN: usize = 34;
/// The metadata key used for storing the JSON encoded `Schema`
pub const SCHEMA_METADATA_KEY: &str = "avro.schema";
/// Metadata key used to represent Avro enum symbols in an Arrow schema.
pub const AVRO_ENUM_SYMBOLS_METADATA_KEY: &str = "avro.enum.symbols";
/// Metadata key used to store the default value of a field in an Avro schema.
pub const AVRO_FIELD_DEFAULT_METADATA_KEY: &str = "avro.field.default";
/// Metadata key used to store the name of a type in an Avro schema.
pub const AVRO_NAME_METADATA_KEY: &str = "avro.name";
/// Metadata key used to store the name of a type in an Avro schema.
pub const AVRO_NAMESPACE_METADATA_KEY: &str = "avro.namespace";
/// Metadata key used to store the documentation for a type in an Avro schema.
pub const AVRO_DOC_METADATA_KEY: &str = "avro.doc";
/// Default name for the root record in an Avro schema.
pub const AVRO_ROOT_RECORD_DEFAULT_NAME: &str = "topLevelRecord";
/// Avro types are not nullable, with nullability instead encoded as a union
/// where one of the variants is the null type.
///
/// To accommodate this, we specially case two-variant unions where one of the
/// variants is the null type, and use this to derive arrow's notion of nullability
#[derive(Debug, Copy, Clone, PartialEq, Default)]
pub(crate) enum Nullability {
/// The nulls are encoded as the first union variant
#[default]
NullFirst,
/// The nulls are encoded as the second union variant
NullSecond,
}
impl Nullability {
/// Returns the index of the non-null variant in the union.
pub(crate) fn non_null_index(&self) -> usize {
match self {
Nullability::NullFirst => 1,
Nullability::NullSecond => 0,
}
}
}
/// Either a [`PrimitiveType`] or a reference to a previously defined named type
///
/// <https://avro.apache.org/docs/1.11.1/specification/#names>
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(untagged)]
/// A type name in an Avro schema
///
/// This represents the different ways a type can be referenced in an Avro schema.
pub(crate) enum TypeName<'a> {
/// A primitive type like null, boolean, int, etc.
Primitive(PrimitiveType),
/// A reference to another named type
Ref(&'a str),
}
/// A primitive type
///
/// <https://avro.apache.org/docs/1.11.1/specification/#primitive-types>
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, AsRefStr)]
#[serde(rename_all = "camelCase")]
#[strum(serialize_all = "lowercase")]
pub(crate) enum PrimitiveType {
/// null: no value
Null,
/// boolean: a binary value
Boolean,
/// int: 32-bit signed integer
Int,
/// long: 64-bit signed integer
Long,
/// float: single precision (32-bit) IEEE 754 floating-point number
Float,
/// double: double precision (64-bit) IEEE 754 floating-point number
Double,
/// bytes: sequence of 8-bit unsigned bytes
Bytes,
/// string: Unicode character sequence
String,
}
/// Additional attributes within a `Schema`
///
/// <https://avro.apache.org/docs/1.11.1/specification/#schema-declaration>
#[derive(Debug, Clone, PartialEq, Eq, Default, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Attributes<'a> {
/// A logical type name
///
/// <https://avro.apache.org/docs/1.11.1/specification/#logical-types>
#[serde(default)]
pub(crate) logical_type: Option<&'a str>,
/// Additional JSON attributes
#[serde(flatten)]
pub(crate) additional: HashMap<&'a str, Value>,
}
impl Attributes<'_> {
/// Returns the field metadata for this [`Attributes`]
pub(crate) fn field_metadata(&self) -> HashMap<String, String> {
self.additional
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect()
}
}
/// A type definition that is not a variant of [`ComplexType`]
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Type<'a> {
/// The type of this Avro data structure
#[serde(borrow)]
pub(crate) r#type: TypeName<'a>,
/// Additional attributes associated with this type
#[serde(flatten)]
pub(crate) attributes: Attributes<'a>,
}
/// An Avro schema
///
/// This represents the different shapes of Avro schemas as defined in the specification.
/// See <https://avro.apache.org/docs/1.11.1/specification/#schemas> for more details.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(untagged)]
pub(crate) enum Schema<'a> {
/// A direct type name (primitive or reference)
#[serde(borrow)]
TypeName(TypeName<'a>),
/// A union of multiple schemas (e.g., ["null", "string"])
#[serde(borrow)]
Union(Vec<Schema<'a>>),
/// A complex type such as record, array, map, etc.
#[serde(borrow)]
Complex(ComplexType<'a>),
/// A type with attributes
#[serde(borrow)]
Type(Type<'a>),
}
/// A complex type
///
/// <https://avro.apache.org/docs/1.11.1/specification/#complex-types>
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub(crate) enum ComplexType<'a> {
/// Record type: a sequence of fields with names and types
#[serde(borrow)]
Record(Record<'a>),
/// Enum type: a set of named values
#[serde(borrow)]
Enum(Enum<'a>),
/// Array type: a sequence of values of the same type
#[serde(borrow)]
Array(Array<'a>),
/// Map type: a mapping from strings to values of the same type
#[serde(borrow)]
Map(Map<'a>),
/// Fixed type: a fixed-size byte array
#[serde(borrow)]
Fixed(Fixed<'a>),
}
/// A record
///
/// <https://avro.apache.org/docs/1.11.1/specification/#schema-record>
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct Record<'a> {
/// Name of the record
#[serde(borrow)]
pub(crate) name: &'a str,
/// Optional namespace for the record, provides a way to organize names
#[serde(borrow, default)]
pub(crate) namespace: Option<&'a str>,
/// Optional documentation string for the record
#[serde(borrow, default)]
pub(crate) doc: Option<Cow<'a, str>>,
/// Alternative names for this record
#[serde(borrow, default)]
pub(crate) aliases: Vec<&'a str>,
/// The fields contained in this record
#[serde(borrow)]
pub(crate) fields: Vec<Field<'a>>,
/// Additional attributes for this record
#[serde(flatten)]
pub(crate) attributes: Attributes<'a>,
}
fn deserialize_default<'de, D>(deserializer: D) -> Result<Option<Value>, D::Error>
where
D: serde::Deserializer<'de>,
{
Value::deserialize(deserializer).map(Some)
}
/// A field within a [`Record`]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct Field<'a> {
/// Name of the field within the record
#[serde(borrow)]
pub(crate) name: &'a str,
/// Optional documentation for this field
#[serde(borrow, default)]
pub(crate) doc: Option<Cow<'a, str>>,
/// The field's type definition
#[serde(borrow)]
pub(crate) r#type: Schema<'a>,
/// Optional default value for this field
#[serde(deserialize_with = "deserialize_default", default)]
pub(crate) default: Option<Value>,
/// Alternative names (aliases) for this field (Avro spec: field-level aliases).
/// Borrowed from input JSON where possible.
#[serde(borrow, default)]
pub(crate) aliases: Vec<&'a str>,
}
/// An enumeration
///
/// <https://avro.apache.org/docs/1.11.1/specification/#enums>
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct Enum<'a> {
/// Name of the enum
#[serde(borrow)]
pub(crate) name: &'a str,
/// Optional namespace for the enum, provides organizational structure
#[serde(borrow, default)]
pub(crate) namespace: Option<&'a str>,
/// Optional documentation string describing the enum
#[serde(borrow, default)]
pub(crate) doc: Option<Cow<'a, str>>,
/// Alternative names for this enum
#[serde(borrow, default)]
pub(crate) aliases: Vec<&'a str>,
/// The symbols (values) that this enum can have
#[serde(borrow)]
pub(crate) symbols: Vec<&'a str>,
/// Optional default value for this enum
#[serde(borrow, default)]
pub(crate) default: Option<&'a str>,
/// Additional attributes for this enum
#[serde(flatten)]
pub(crate) attributes: Attributes<'a>,
}
/// An array
///
/// <https://avro.apache.org/docs/1.11.1/specification/#arrays>
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct Array<'a> {
/// The schema for items in this array
#[serde(borrow)]
pub(crate) items: Box<Schema<'a>>,
/// Additional attributes for this array
#[serde(flatten)]
pub(crate) attributes: Attributes<'a>,
}
/// A map
///
/// <https://avro.apache.org/docs/1.11.1/specification/#maps>
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct Map<'a> {
/// The schema for values in this map
#[serde(borrow)]
pub(crate) values: Box<Schema<'a>>,
/// Additional attributes for this map
#[serde(flatten)]
pub(crate) attributes: Attributes<'a>,
}
/// A fixed length binary array
///
/// <https://avro.apache.org/docs/1.11.1/specification/#fixed>
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct Fixed<'a> {
/// Name of the fixed type
#[serde(borrow)]
pub(crate) name: &'a str,
/// Optional namespace for the fixed type
#[serde(borrow, default)]
pub(crate) namespace: Option<&'a str>,
/// Alternative names for this fixed type
#[serde(borrow, default)]
pub(crate) aliases: Vec<&'a str>,
/// The number of bytes in this fixed type
pub(crate) size: usize,
/// Additional attributes for this fixed type
#[serde(flatten)]
pub(crate) attributes: Attributes<'a>,
}
#[derive(Debug, Copy, Clone, PartialEq, Default)]
pub(crate) struct AvroSchemaOptions {
pub(crate) null_order: Option<Nullability>,
pub(crate) strip_metadata: bool,
}
/// A wrapper for an Avro schema in its JSON string representation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AvroSchema {
/// The Avro schema as a JSON string.
pub json_string: String,
}
impl TryFrom<&ArrowSchema> for AvroSchema {
type Error = ArrowError;
/// Converts an `ArrowSchema` to `AvroSchema`, delegating to
/// `AvroSchema::from_arrow_with_options` with `None` so that the
/// union null ordering is decided by `Nullability::default()`.
fn try_from(schema: &ArrowSchema) -> Result<Self, Self::Error> {
AvroSchema::from_arrow_with_options(schema, None)
}
}
impl AvroSchema {
/// Creates a new `AvroSchema` from a JSON string.
pub fn new(json_string: String) -> Self {
Self { json_string }
}
pub(crate) fn schema(&self) -> Result<Schema<'_>, ArrowError> {
serde_json::from_str(self.json_string.as_str())
.map_err(|e| ArrowError::ParseError(format!("Invalid Avro schema JSON: {e}")))
}
/// Returns the fingerprint of the schema, computed using the specified [`FingerprintAlgorithm`].
///
/// The fingerprint is computed over the schema's Parsed Canonical Form
/// as defined by the Avro specification. Depending on `hash_type`, this
/// will return one of the supported [`Fingerprint`] variants:
/// - [`Fingerprint::Rabin`] for [`FingerprintAlgorithm::Rabin`]
/// - `Fingerprint::MD5` for `FingerprintAlgorithm::MD5`
/// - `Fingerprint::SHA256` for `FingerprintAlgorithm::SHA256`
///
/// Note: [`FingerprintAlgorithm::Id`] or [`FingerprintAlgorithm::Id64`] cannot be used to generate a fingerprint
/// and will result in an error. If you intend to use a Schema Registry ID-based
/// wire format, either use [`SchemaStore::set`] or load the [`Fingerprint::Id`] directly via [`Fingerprint::load_fingerprint_id`] or for
/// [`Fingerprint::Id64`] via [`Fingerprint::load_fingerprint_id64`].
///
/// See also: <https://avro.apache.org/docs/1.11.1/specification/#schema-fingerprints>
///
/// # Errors
/// Returns an error if deserializing the schema fails, if generating the
/// canonical form of the schema fails, or if `hash_type` is [`FingerprintAlgorithm::Id`].
///
/// # Examples
/// ```
/// use arrow_avro::schema::{AvroSchema, FingerprintAlgorithm};
///
/// let avro = AvroSchema::new("\"string\"".to_string());
/// let fp = avro.fingerprint(FingerprintAlgorithm::Rabin).unwrap();
/// ```
pub fn fingerprint(&self, hash_type: FingerprintAlgorithm) -> Result<Fingerprint, ArrowError> {
Self::generate_fingerprint(&self.schema()?, hash_type)
}
pub(crate) fn project(&self, projection: &[usize]) -> Result<Self, ArrowError> {
let mut value: Value = serde_json::from_str(&self.json_string)
.map_err(|e| ArrowError::AvroError(format!("Invalid Avro schema JSON: {e}")))?;
let obj = value.as_object_mut().ok_or_else(|| {
ArrowError::AvroError(
"Projected schema must be a JSON object Avro record schema".to_string(),
)
})?;
match obj.get("type").and_then(|v| v.as_str()) {
Some("record") => {}
Some(other) => {
return Err(ArrowError::AvroError(format!(
"Projected schema must be an Avro record, found type '{other}'"
)));
}
None => {
return Err(ArrowError::AvroError(
"Projected schema missing required 'type' field".to_string(),
));
}
}
let fields_val = obj.get_mut("fields").ok_or_else(|| {
ArrowError::AvroError("Avro record schema missing required 'fields'".to_string())
})?;
let projected_fields = {
let mut original_fields = match fields_val {
Value::Array(arr) => std::mem::take(arr),
_ => {
return Err(ArrowError::AvroError(
"Avro record schema 'fields' must be an array".to_string(),
));
}
};
let len = original_fields.len();
let mut seen: HashSet<usize> = HashSet::with_capacity(projection.len());
let mut out: Vec<Value> = Vec::with_capacity(projection.len());
for &i in projection {
if i >= len {
return Err(ArrowError::AvroError(format!(
"Projection index {i} out of bounds for record with {len} fields"
)));
}
if !seen.insert(i) {
return Err(ArrowError::AvroError(format!(
"Duplicate projection index {i}"
)));
}
out.push(std::mem::replace(&mut original_fields[i], Value::Null));
}
out
};
*fields_val = Value::Array(projected_fields);
let json_string = serde_json::to_string(&value).map_err(|e| {
ArrowError::AvroError(format!(
"Failed to serialize projected Avro schema JSON: {e}"
))
})?;
Ok(Self::new(json_string))
}
pub(crate) fn generate_fingerprint(
schema: &Schema,
hash_type: FingerprintAlgorithm,
) -> Result<Fingerprint, ArrowError> {
let canonical = Self::generate_canonical_form(schema).map_err(|e| {
ArrowError::ComputeError(format!("Failed to generate canonical form for schema: {e}"))
})?;
match hash_type {
FingerprintAlgorithm::Rabin => {
Ok(Fingerprint::Rabin(compute_fingerprint_rabin(&canonical)))
}
FingerprintAlgorithm::Id | FingerprintAlgorithm::Id64 => Err(ArrowError::SchemaError(
"FingerprintAlgorithm of Id or Id64 cannot be used to generate a fingerprint; \
if using Fingerprint::Id, pass the registry ID in instead using the set method."
.to_string(),
)),
#[cfg(feature = "md5")]
FingerprintAlgorithm::MD5 => Ok(Fingerprint::MD5(compute_fingerprint_md5(&canonical))),
#[cfg(feature = "sha256")]
FingerprintAlgorithm::SHA256 => {
Ok(Fingerprint::SHA256(compute_fingerprint_sha256(&canonical)))
}
}
}
/// Generates the Parsed Canonical Form for the given `Schema`.
///
/// The canonical form is a standardized JSON representation of the schema,
/// primarily used for generating a schema fingerprint for equality checking.
///
/// This form strips attributes that do not affect the schema's identity,
/// such as `doc` fields, `aliases`, and any properties not defined in the
/// Avro specification.
///
/// <https://avro.apache.org/docs/1.11.1/specification/#parsing-canonical-form-for-schemas>
pub(crate) fn generate_canonical_form(schema: &Schema) -> Result<String, ArrowError> {
build_canonical(schema, None)
}
/// Build Avro JSON from an Arrow [`ArrowSchema`], applying the given null‑union order and optionally stripping internal Arrow metadata.
///
/// If the input Arrow schema already contains Avro JSON in
/// [`SCHEMA_METADATA_KEY`], that JSON is returned verbatim to preserve
/// the exact header encoding alignment; otherwise, a new JSON is generated
/// honoring `null_union_order` at **all nullable sites**.
pub(crate) fn from_arrow_with_options(
schema: &ArrowSchema,
options: Option<AvroSchemaOptions>,
) -> Result<AvroSchema, ArrowError> {
let opts = options.unwrap_or_default();
let order = opts.null_order.unwrap_or_default();
let strip = opts.strip_metadata;
if !strip {
if let Some(json) = schema.metadata.get(SCHEMA_METADATA_KEY) {
return Ok(AvroSchema::new(json.clone()));
}
}
let mut name_gen = NameGenerator::default();
let fields_json = schema
.fields()
.iter()
.map(|f| arrow_field_to_avro(f, &mut name_gen, order, strip))
.collect::<Result<Vec<_>, _>>()?;
let record_name = schema
.metadata
.get(AVRO_NAME_METADATA_KEY)
.map_or(AVRO_ROOT_RECORD_DEFAULT_NAME, |s| s.as_str());
let mut record = JsonMap::with_capacity(schema.metadata.len() + 4);
record.insert("type".into(), Value::String("record".into()));
record.insert(
"name".into(),
Value::String(sanitise_avro_name(record_name)),
);
if let Some(ns) = schema.metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
record.insert("namespace".into(), Value::String(ns.clone()));
}
if let Some(doc) = schema.metadata.get(AVRO_DOC_METADATA_KEY) {
record.insert("doc".into(), Value::String(doc.clone()));
}
record.insert("fields".into(), Value::Array(fields_json));
extend_with_passthrough_metadata(&mut record, &schema.metadata);
let json_string = serde_json::to_string(&Value::Object(record))
.map_err(|e| ArrowError::SchemaError(format!("Serializing Avro JSON failed: {e}")))?;
Ok(AvroSchema::new(json_string))
}
}
/// A stack-allocated, fixed-size buffer for the prefix.
#[derive(Debug, Copy, Clone)]
pub(crate) struct Prefix {
buf: [u8; MAX_PREFIX_LEN],
len: u8,
}
impl Prefix {
#[inline]
pub(crate) fn as_slice(&self) -> &[u8] {
&self.buf[..self.len as usize]
}
}
/// Defines the strategy for generating the per-record prefix for an Avro binary stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum FingerprintStrategy {
/// Use the 64-bit Rabin fingerprint (default for single-object encoding).
#[default]
Rabin,
/// Use a Confluent Schema Registry 32-bit ID.
Id(u32),
/// Use an Apicurio Schema Registry 64-bit ID.
Id64(u64),
#[cfg(feature = "md5")]
/// Use the 128-bit MD5 fingerprint.
MD5,
#[cfg(feature = "sha256")]
/// Use the 256-bit SHA-256 fingerprint.
SHA256,
}
impl From<Fingerprint> for FingerprintStrategy {
fn from(f: Fingerprint) -> Self {
Self::from(&f)
}
}
impl From<FingerprintAlgorithm> for FingerprintStrategy {
fn from(f: FingerprintAlgorithm) -> Self {
match f {
FingerprintAlgorithm::Rabin => FingerprintStrategy::Rabin,
FingerprintAlgorithm::Id => FingerprintStrategy::Id(0),
FingerprintAlgorithm::Id64 => FingerprintStrategy::Id64(0),
#[cfg(feature = "md5")]
FingerprintAlgorithm::MD5 => FingerprintStrategy::MD5,
#[cfg(feature = "sha256")]
FingerprintAlgorithm::SHA256 => FingerprintStrategy::SHA256,
}
}
}
impl From<&Fingerprint> for FingerprintStrategy {
fn from(f: &Fingerprint) -> Self {
match f {
Fingerprint::Rabin(_) => FingerprintStrategy::Rabin,
Fingerprint::Id(_) => FingerprintStrategy::Id(0),
Fingerprint::Id64(_) => FingerprintStrategy::Id64(0),
#[cfg(feature = "md5")]
Fingerprint::MD5(_) => FingerprintStrategy::MD5,
#[cfg(feature = "sha256")]
Fingerprint::SHA256(_) => FingerprintStrategy::SHA256,
}
}
}
/// Supported fingerprint algorithms for Avro schema identification.
/// For use with Confluent Schema Registry IDs, set to None.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Default)]
pub enum FingerprintAlgorithm {
/// 64‑bit CRC‑64‑AVRO Rabin fingerprint.
#[default]
Rabin,
/// Represents a 32 bit fingerprint not based on a hash algorithm, (e.g., a 32-bit Schema Registry ID.)
Id,
/// Represents a 64 bit fingerprint not based on a hash algorithm, (e.g., a 64-bit Schema Registry ID.)
Id64,
#[cfg(feature = "md5")]
/// 128-bit MD5 message digest.
MD5,
#[cfg(feature = "sha256")]
/// 256-bit SHA-256 digest.
SHA256,
}
/// Allow easy extraction of the algorithm used to create a fingerprint.
impl From<&Fingerprint> for FingerprintAlgorithm {
fn from(fp: &Fingerprint) -> Self {
match fp {
Fingerprint::Rabin(_) => FingerprintAlgorithm::Rabin,
Fingerprint::Id(_) => FingerprintAlgorithm::Id,
Fingerprint::Id64(_) => FingerprintAlgorithm::Id64,
#[cfg(feature = "md5")]
Fingerprint::MD5(_) => FingerprintAlgorithm::MD5,
#[cfg(feature = "sha256")]
Fingerprint::SHA256(_) => FingerprintAlgorithm::SHA256,
}
}
}
impl From<FingerprintStrategy> for FingerprintAlgorithm {
fn from(s: FingerprintStrategy) -> Self {
Self::from(&s)
}
}
impl From<&FingerprintStrategy> for FingerprintAlgorithm {
fn from(s: &FingerprintStrategy) -> Self {
match s {
FingerprintStrategy::Rabin => FingerprintAlgorithm::Rabin,
FingerprintStrategy::Id(_) => FingerprintAlgorithm::Id,
FingerprintStrategy::Id64(_) => FingerprintAlgorithm::Id64,
#[cfg(feature = "md5")]
FingerprintStrategy::MD5 => FingerprintAlgorithm::MD5,
#[cfg(feature = "sha256")]
FingerprintStrategy::SHA256 => FingerprintAlgorithm::SHA256,
}
}
}
/// A schema fingerprint in one of the supported formats.
///
/// This is used as the key inside `SchemaStore` `HashMap`. Each `SchemaStore`
/// instance always stores only one variant, matching its configured
/// `FingerprintAlgorithm`, but the enum makes the API uniform.
///
/// <https://avro.apache.org/docs/1.11.1/specification/#schema-fingerprints>
/// <https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format>
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum Fingerprint {
/// A 64-bit Rabin fingerprint.
Rabin(u64),
/// A 32-bit Schema Registry ID.
Id(u32),
/// A 64-bit Schema Registry ID.
Id64(u64),
#[cfg(feature = "md5")]
/// A 128-bit MD5 fingerprint.
MD5([u8; 16]),
#[cfg(feature = "sha256")]
/// A 256-bit SHA-256 fingerprint.
SHA256([u8; 32]),
}
impl From<FingerprintStrategy> for Fingerprint {
fn from(s: FingerprintStrategy) -> Self {
Self::from(&s)
}
}
impl From<&FingerprintStrategy> for Fingerprint {
fn from(s: &FingerprintStrategy) -> Self {
match s {
FingerprintStrategy::Rabin => Fingerprint::Rabin(0),
FingerprintStrategy::Id(id) => Fingerprint::Id(*id),
FingerprintStrategy::Id64(id) => Fingerprint::Id64(*id),
#[cfg(feature = "md5")]
FingerprintStrategy::MD5 => Fingerprint::MD5([0; 16]),
#[cfg(feature = "sha256")]
FingerprintStrategy::SHA256 => Fingerprint::SHA256([0; 32]),
}
}
}
impl From<FingerprintAlgorithm> for Fingerprint {
fn from(s: FingerprintAlgorithm) -> Self {
match s {
FingerprintAlgorithm::Rabin => Fingerprint::Rabin(0),
FingerprintAlgorithm::Id => Fingerprint::Id(0),
FingerprintAlgorithm::Id64 => Fingerprint::Id64(0),
#[cfg(feature = "md5")]
FingerprintAlgorithm::MD5 => Fingerprint::MD5([0; 16]),
#[cfg(feature = "sha256")]
FingerprintAlgorithm::SHA256 => Fingerprint::SHA256([0; 32]),
}
}
}
impl Fingerprint {
/// Loads the 32-bit Schema Registry fingerprint (Confluent Schema Registry ID).
///
/// The provided `id` is in big-endian wire order; this converts it to host order
/// and returns `Fingerprint::Id`.
///
/// # Returns
/// A `Fingerprint::Id` variant containing the 32-bit fingerprint.
pub fn load_fingerprint_id(id: u32) -> Self {
Fingerprint::Id(u32::from_be(id))
}
/// Loads the 64-bit Schema Registry fingerprint (Apicurio Schema Registry ID).
///
/// The provided `id` is in big-endian wire order; this converts it to host order
/// and returns `Fingerprint::Id64`.
///
/// # Returns
/// A `Fingerprint::Id64` variant containing the 64-bit fingerprint.
pub fn load_fingerprint_id64(id: u64) -> Self {
Fingerprint::Id64(u64::from_be(id))
}
/// Constructs a serialized prefix represented as a `Vec<u8>` based on the variant of the enum.
///
/// This method serializes data in different formats depending on the variant of `self`:
/// - **`Id(id)`**: Uses the Confluent wire format, which includes a predefined magic header (`CONFLUENT_MAGIC`)
/// followed by the big-endian byte representation of the `id`.
/// - **`Id64(id)`**: Uses the Apicurio wire format, which includes a predefined magic header (`CONFLUENT_MAGIC`)
/// followed by the big-endian 8-byte representation of the `id`.
/// - **`Rabin(val)`**: Uses the Avro single-object specification format. This includes a different magic header
/// (`SINGLE_OBJECT_MAGIC`) followed by the little-endian byte representation of the `val`.
/// - **`MD5(bytes)`** (optional, `md5` feature enabled): A non-standard extension that adds the
/// `SINGLE_OBJECT_MAGIC` header followed by the provided `bytes`.
/// - **`SHA256(bytes)`** (optional, `sha256` feature enabled): Similar to the `MD5` variant, this is
/// a non-standard extension that attaches the `SINGLE_OBJECT_MAGIC` header followed by the given `bytes`.
///
/// # Returns
///
/// A `Prefix` containing the serialized prefix data.
///
/// # Features
///
/// - You can optionally enable the `md5` feature to include the `MD5` variant.
/// - You can optionally enable the `sha256` feature to include the `SHA256` variant.
///
pub(crate) fn make_prefix(&self) -> Prefix {
let mut buf = [0u8; MAX_PREFIX_LEN];
let len = match self {
Self::Id(val) => write_prefix(&mut buf, &CONFLUENT_MAGIC, &val.to_be_bytes()),
Self::Id64(val) => write_prefix(&mut buf, &CONFLUENT_MAGIC, &val.to_be_bytes()),
Self::Rabin(val) => write_prefix(&mut buf, &SINGLE_OBJECT_MAGIC, &val.to_le_bytes()),
#[cfg(feature = "md5")]
Self::MD5(val) => write_prefix(&mut buf, &SINGLE_OBJECT_MAGIC, val),
#[cfg(feature = "sha256")]
Self::SHA256(val) => write_prefix(&mut buf, &SINGLE_OBJECT_MAGIC, val),
};
Prefix { buf, len }
}
}
fn write_prefix<const MAGIC_LEN: usize, const PAYLOAD_LEN: usize>(
buf: &mut [u8; MAX_PREFIX_LEN],
magic: &[u8; MAGIC_LEN],
payload: &[u8; PAYLOAD_LEN],
) -> u8 {
debug_assert!(MAGIC_LEN + PAYLOAD_LEN <= MAX_PREFIX_LEN);
let total = MAGIC_LEN + PAYLOAD_LEN;
let prefix_slice = &mut buf[..total];
prefix_slice[..MAGIC_LEN].copy_from_slice(magic);
prefix_slice[MAGIC_LEN..total].copy_from_slice(payload);
total as u8
}
/// An in-memory cache of Avro schemas, indexed by their fingerprint.
///
/// `SchemaStore` provides a mechanism to store and retrieve Avro schemas efficiently.
/// Each schema is associated with a unique [`Fingerprint`], which is generated based
/// on the schema's canonical form and a specific hashing algorithm.
///
/// A `SchemaStore` instance is configured to use a single [`FingerprintAlgorithm`] such as Rabin,
/// MD5 (not yet supported), or SHA256 (not yet supported) for all its operations.
/// This ensures consistency when generating fingerprints and looking up schemas.
/// All schemas registered will have their fingerprint computed with this algorithm, and
/// lookups must use a matching fingerprint.
///
/// # Examples
///
/// ```no_run
/// // Create a new store with the default Rabin fingerprinting.
/// use arrow_avro::schema::{AvroSchema, SchemaStore};
///
/// let mut store = SchemaStore::new();
/// let schema = AvroSchema::new("\"string\"".to_string());
/// // Register the schema to get its fingerprint.
/// let fingerprint = store.register(schema.clone()).unwrap();
/// // Use the fingerprint to look up the schema.
/// let retrieved_schema = store.lookup(&fingerprint).cloned();
/// assert_eq!(retrieved_schema, Some(schema));
/// ```
#[derive(Debug, Clone, Default)]
pub struct SchemaStore {
/// The hashing algorithm used for generating fingerprints.
fingerprint_algorithm: FingerprintAlgorithm,
/// A map from a schema's fingerprint to the schema itself.
schemas: HashMap<Fingerprint, AvroSchema>,
}
impl TryFrom<HashMap<Fingerprint, AvroSchema>> for SchemaStore {
type Error = ArrowError;
/// Creates a `SchemaStore` from a HashMap of schemas.
/// Each schema in the HashMap is registered with the new store.
fn try_from(schemas: HashMap<Fingerprint, AvroSchema>) -> Result<Self, Self::Error> {
Ok(Self {
schemas,
..Self::default()
})
}
}
impl SchemaStore {
/// Creates an empty `SchemaStore` using the default fingerprinting algorithm (64-bit Rabin).
pub fn new() -> Self {
Self::default()
}
/// Creates an empty `SchemaStore` using the default fingerprinting algorithm (64-bit Rabin).
pub fn new_with_type(fingerprint_algorithm: FingerprintAlgorithm) -> Self {
Self {
fingerprint_algorithm,
..Self::default()
}
}
/// Registers a schema with the store and the provided fingerprint.
/// Note: Confluent wire format implementations should leverage this method.
///
/// A schema is set in the store, using the provided fingerprint. If a schema
/// with the same fingerprint does not already exist in the store, the new schema
/// is inserted. If the fingerprint already exists, the existing schema is not overwritten.
///
/// # Arguments
///
/// * `fingerprint` - A reference to the `Fingerprint` of the schema to register.
/// * `schema` - The `AvroSchema` to register.
///
/// # Returns
///
/// A `Result` returning the provided `Fingerprint` of the schema if successful,
/// or an `ArrowError` on failure.
pub fn set(
&mut self,
fingerprint: Fingerprint,
schema: AvroSchema,
) -> Result<Fingerprint, ArrowError> {
match self.schemas.entry(fingerprint) {
Entry::Occupied(entry) => {
if entry.get() != &schema {
return Err(ArrowError::ComputeError(format!(
"Schema fingerprint collision detected for fingerprint {fingerprint:?}"
)));
}
}
Entry::Vacant(entry) => {
entry.insert(schema);
}
}
Ok(fingerprint)
}
/// Registers a schema with the store and returns its fingerprint.
///
/// A fingerprint is calculated for the given schema using the store's configured
/// hash type. If a schema with the same fingerprint does not already exist in the
/// store, the new schema is inserted. If the fingerprint already exists, the
/// existing schema is not overwritten. If FingerprintAlgorithm is set to Id or Id64, this
/// method will return an error. Confluent wire format implementations should leverage the
/// set method instead.
///
/// # Arguments
///
/// * `schema` - The `AvroSchema` to register.
///
/// # Returns
///
/// A `Result` containing the `Fingerprint` of the schema if successful,
/// or an `ArrowError` on failure.
pub fn register(&mut self, schema: AvroSchema) -> Result<Fingerprint, ArrowError> {
if self.fingerprint_algorithm == FingerprintAlgorithm::Id
|| self.fingerprint_algorithm == FingerprintAlgorithm::Id64
{
return Err(ArrowError::SchemaError(
"Invalid FingerprintAlgorithm; unable to generate fingerprint. \
Use the set method directly instead, providing a valid fingerprint"
.to_string(),
));
}
let fingerprint =
AvroSchema::generate_fingerprint(&schema.schema()?, self.fingerprint_algorithm)?;
self.set(fingerprint, schema)?;
Ok(fingerprint)
}
/// Looks up a schema by its `Fingerprint`.
///
/// # Arguments
///
/// * `fingerprint` - A reference to the `Fingerprint` of the schema to look up.
///
/// # Returns
///
/// An `Option` containing a clone of the `AvroSchema` if found, otherwise `None`.
pub fn lookup(&self, fingerprint: &Fingerprint) -> Option<&AvroSchema> {
self.schemas.get(fingerprint)
}
/// Returns a `Vec` containing **all unique [`Fingerprint`]s** currently
/// held by this [`SchemaStore`].
///
/// The order of the returned fingerprints is unspecified and should not be
/// relied upon.
pub fn fingerprints(&self) -> Vec<Fingerprint> {
self.schemas.keys().copied().collect()
}
/// Returns the `FingerprintAlgorithm` used by the `SchemaStore` for fingerprinting.
pub(crate) fn fingerprint_algorithm(&self) -> FingerprintAlgorithm {
self.fingerprint_algorithm
}
}
fn quote(s: &str) -> Result<String, ArrowError> {
serde_json::to_string(s)
.map_err(|e| ArrowError::ComputeError(format!("Failed to quote string: {e}")))
}
// Avro names are defined by a `name` and an optional `namespace`.
// The full name is composed of the namespace and the name, separated by a dot.
//
// Avro specification defines two ways to specify a full name:
// 1. The `name` attribute contains the full name (e.g., "a.b.c.d").
// In this case, the `namespace` attribute is ignored.
// 2. The `name` attribute contains the simple name (e.g., "d") and the
// `namespace` attribute contains the namespace (e.g., "a.b.c").
//
// Each part of the name must match the regex `^[A-Za-z_][A-Za-z0-9_]*$`.
// Complex paths with quotes or backticks like `a."hi".b` are not supported.
//
// This function constructs the full name and extracts the namespace,
// handling both ways of specifying the name. It prioritizes a namespace
// defined within the `name` attribute itself, then the explicit `namespace_attr`,
// and finally the `enclosing_ns`.
pub(crate) fn make_full_name(
name: &str,
namespace_attr: Option<&str>,
enclosing_ns: Option<&str>,
) -> (String, Option<String>) {
// `name` already contains a dot then treat as full-name, ignore namespace.
if let Some((ns, _)) = name.rsplit_once('.') {
return (name.to_string(), Some(ns.to_string()));
}
match namespace_attr.or(enclosing_ns) {
Some(ns) => (format!("{ns}.{name}"), Some(ns.to_string())),
None => (name.to_string(), None),
}
}
fn build_canonical(schema: &Schema, enclosing_ns: Option<&str>) -> Result<String, ArrowError> {
Ok(match schema {
Schema::TypeName(tn) | Schema::Type(Type { r#type: tn, .. }) => match tn {
TypeName::Primitive(pt) => quote(pt.as_ref())?,
TypeName::Ref(name) => {
let (full_name, _) = make_full_name(name, None, enclosing_ns);
quote(&full_name)?
}
},
Schema::Union(branches) => format!(
"[{}]",
branches
.iter()
.map(|b| build_canonical(b, enclosing_ns))
.collect::<Result<Vec<_>, _>>()?
.join(",")
),
Schema::Complex(ct) => match ct {
ComplexType::Record(r) => {
let (full_name, child_ns) = make_full_name(r.name, r.namespace, enclosing_ns);
let fields = r
.fields
.iter()
.map(|f| {
// PCF [STRIP] per Avro spec: keep only attributes relevant to parsing
// ("name" and "type" for fields) and **strip others** such as doc,
// default, order, and **aliases**. This preserves canonicalization. See:
// https://avro.apache.org/docs/1.11.1/specification/#parsing-canonical-form-for-schemas
let field_type =
build_canonical(&f.r#type, child_ns.as_deref().or(enclosing_ns))?;
Ok(format!(
r#"{{"name":{},"type":{}}}"#,
quote(f.name)?,
field_type
))
})
.collect::<Result<Vec<_>, ArrowError>>()?
.join(",");
format!(
r#"{{"name":{},"type":"record","fields":[{fields}]}}"#,
quote(&full_name)?,
)
}
ComplexType::Enum(e) => {
let (full_name, _) = make_full_name(e.name, e.namespace, enclosing_ns);
let symbols = e
.symbols
.iter()
.map(|s| quote(s))
.collect::<Result<Vec<_>, _>>()?
.join(",");
format!(
r#"{{"name":{},"type":"enum","symbols":[{symbols}]}}"#,
quote(&full_name)?
)
}
ComplexType::Array(arr) => format!(
r#"{{"type":"array","items":{}}}"#,
build_canonical(&arr.items, enclosing_ns)?
),
ComplexType::Map(map) => format!(
r#"{{"type":"map","values":{}}}"#,
build_canonical(&map.values, enclosing_ns)?
),
ComplexType::Fixed(f) => {
let (full_name, _) = make_full_name(f.name, f.namespace, enclosing_ns);
format!(
r#"{{"name":{},"type":"fixed","size":{}}}"#,
quote(&full_name)?,
f.size
)
}
},
})
}
/// 64‑bit Rabin fingerprint as described in the Avro spec.
const EMPTY: u64 = 0xc15d_213a_a4d7_a795;
/// Build one entry of the polynomial‑division table.
///
/// We cannot yet write `for _ in 0..8` here: `for` loops rely on
/// `Iterator::next`, which is not `const` on stable Rust. Until the
/// `const_for` feature (tracking issue #87575) is stabilized, a `while`
/// loop is the only option in a `const fn`
const fn one_entry(i: usize) -> u64 {
let mut fp = i as u64;
let mut j = 0;
while j < 8 {
fp = (fp >> 1) ^ (EMPTY & (0u64.wrapping_sub(fp & 1)));
j += 1;
}
fp
}
/// Build the full 256‑entry table at compile time.
///
/// We cannot yet write `for _ in 0..256` here: `for` loops rely on
/// `Iterator::next`, which is not `const` on stable Rust. Until the
/// `const_for` feature (tracking issue #87575) is stabilized, a `while`
/// loop is the only option in a `const fn`
const fn build_table() -> [u64; 256] {
let mut table = [0u64; 256];
let mut i = 0;
while i < 256 {
table[i] = one_entry(i);
i += 1;
}
table
}
/// The pre‑computed table.
static FINGERPRINT_TABLE: [u64; 256] = build_table();
/// Computes the 64-bit Rabin fingerprint for a given canonical schema string.
/// This implementation is based on the Avro specification for schema fingerprinting.
pub(crate) fn compute_fingerprint_rabin(canonical_form: &str) -> u64 {
let mut fp = EMPTY;
for &byte in canonical_form.as_bytes() {
let idx = ((fp as u8) ^ byte) as usize;
fp = (fp >> 8) ^ FINGERPRINT_TABLE[idx];
}
fp
}
#[cfg(feature = "md5")]
/// Compute the **128‑bit MD5** fingerprint of the canonical form.
///
/// Returns a 16‑byte array (`[u8; 16]`) containing the full MD5 digest,
/// exactly as required by the Avro specification.
#[inline]
pub(crate) fn compute_fingerprint_md5(canonical_form: &str) -> [u8; 16] {
let digest = md5::compute(canonical_form.as_bytes());
digest.0
}
#[cfg(feature = "sha256")]
/// Compute the **256‑bit SHA‑256** fingerprint of the canonical form.
///
/// Returns a 32‑byte array (`[u8; 32]`) containing the full SHA‑256 digest.
#[inline]
pub(crate) fn compute_fingerprint_sha256(canonical_form: &str) -> [u8; 32] {
let mut hasher = Sha256::new();
hasher.update(canonical_form.as_bytes());
let digest = hasher.finalize();
digest.into()
}
#[inline]
fn is_internal_arrow_key(key: &str) -> bool {
key.starts_with("ARROW:") || key == SCHEMA_METADATA_KEY
}
/// Copies Arrow schema metadata entries to the provided JSON map,
/// skipping keys that are Avro-reserved, internal Arrow keys, or
/// nested under the `avro.schema.` namespace. Values that parse as
/// JSON are inserted as JSON; otherwise the raw string is preserved.
fn extend_with_passthrough_metadata(
target: &mut JsonMap<String, Value>,
metadata: &HashMap<String, String>,
) {
for (meta_key, meta_val) in metadata {
if meta_key.starts_with("avro.") || is_internal_arrow_key(meta_key) {
continue;
}
let json_val =
serde_json::from_str(meta_val).unwrap_or_else(|_| Value::String(meta_val.clone()));
target.insert(meta_key.clone(), json_val);
}
}
// Sanitize an arbitrary string so it is a valid Avro field or type name
fn sanitise_avro_name(base_name: &str) -> String {
if base_name.is_empty() {
return "_".to_owned();
}
let mut out: String = base_name
.chars()
.map(|char| {
if char.is_ascii_alphanumeric() || char == '_' {
char
} else {
'_'
}
})
.collect();
if out.as_bytes()[0].is_ascii_digit() {
out.insert(0, '_');
}
out
}
#[derive(Default)]
struct NameGenerator {
used: HashSet<String>,
counters: HashMap<String, usize>,
}
impl NameGenerator {
fn make_unique(&mut self, field_name: &str) -> String {
let field_name = sanitise_avro_name(field_name);
if self.used.insert(field_name.clone()) {
self.counters.insert(field_name.clone(), 1);
return field_name;
}
let counter = self.counters.entry(field_name.clone()).or_insert(1);
loop {
let candidate = format!("{field_name}_{}", *counter);
if self.used.insert(candidate.clone()) {
return candidate;
}
*counter += 1;
}
}
}
fn merge_extras(schema: Value, extras: JsonMap<String, Value>) -> Value {
if extras.is_empty() {
return schema;
}
match schema {
Value::Object(mut map) => {
map.extend(extras);
Value::Object(map)
}
Value::Array(mut union) => {
// For unions, we cannot attach attributes to the array itself (per Avro spec).
// As a fallback for extension metadata, attach extras to the first non-null branch object.
if let Some(non_null) = union.iter_mut().find(|val| val.as_str() != Some("null")) {
let original = std::mem::take(non_null);
*non_null = merge_extras(original, extras);
}
Value::Array(union)
}
primitive => {
let mut map = JsonMap::with_capacity(extras.len() + 1);
map.insert("type".into(), primitive);
map.extend(extras);
Value::Object(map)
}
}
}
#[inline]
fn is_avro_json_null(v: &Value) -> bool {
matches!(v, Value::String(s) if s == "null")
}
fn wrap_nullable(inner: Value, null_order: Nullability) -> Value {
let null = Value::String("null".into());
match inner {
Value::Array(mut union) => {
// If this site is already a union and already contains "null",
// preserve the branch order exactly. Reordering "null" breaks
// the correspondence between Arrow union child order (type_ids)
// and the Avro branch index written on the wire.
if union.iter().any(is_avro_json_null) {
return Value::Array(union);
}
// Otherwise, inject "null" without reordering existing branches.
match null_order {
Nullability::NullFirst => union.insert(0, null),
Nullability::NullSecond => union.push(null),
}
Value::Array(union)
}
other => match null_order {
Nullability::NullFirst => Value::Array(vec![null, other]),
Nullability::NullSecond => Value::Array(vec![other, null]),
},
}
}
fn min_fixed_bytes_for_precision(p: usize) -> usize {
// From the spec: max precision for n=1..=32 bytes:
// [2,4,6,9,11,14,16,18,21,23,26,28,31,33,35,38,40,43,45,47,50,52,55,57,59,62,64,67,69,71,74,76]
const MAX_P: [usize; 32] = [
2, 4, 6, 9, 11, 14, 16, 18, 21, 23, 26, 28, 31, 33, 35, 38, 40, 43, 45, 47, 50, 52, 55, 57,
59, 62, 64, 67, 69, 71, 74, 76,
];
for (i, &max_p) in MAX_P.iter().enumerate() {
if p <= max_p {
return i + 1;
}
}
32 // saturate at Decimal256
}
fn union_branch_signature(branch: &Value) -> Result<String, ArrowError> {
match branch {
Value::String(t) => Ok(format!("P:{t}")),
Value::Object(map) => {
let t = map.get("type").and_then(|v| v.as_str()).ok_or_else(|| {
ArrowError::SchemaError("Union branch object missing string 'type'".into())
})?;
match t {
"record" | "enum" | "fixed" => {
let name = map.get("name").and_then(|v| v.as_str()).ok_or_else(|| {
ArrowError::SchemaError(format!(
"Union branch '{t}' missing required 'name'"
))
})?;
Ok(format!("N:{t}:{name}"))
}
"array" | "map" => Ok(format!("C:{t}")),
other => Ok(format!("P:{other}")),
}
}
Value::Array(_) => Err(ArrowError::SchemaError(
"Avro union may not immediately contain another union".into(),
)),
_ => Err(ArrowError::SchemaError(
"Invalid JSON for Avro union branch".into(),
)),
}
}
fn datatype_to_avro(
dt: &DataType,
field_name: &str,
metadata: &HashMap<String, String>,
name_gen: &mut NameGenerator,
null_order: Nullability,
strip: bool,
) -> Result<(Value, JsonMap<String, Value>), ArrowError> {
let mut extras = JsonMap::new();
let mut handle_decimal = |precision: &u8, scale: &i8| -> Result<Value, ArrowError> {
if *scale < 0 {
return Err(ArrowError::SchemaError(format!(
"Invalid Avro decimal for field '{field_name}': scale ({scale}) must be >= 0"
)));
}
if (*scale as usize) > (*precision as usize) {
return Err(ArrowError::SchemaError(format!(
"Invalid Avro decimal for field '{field_name}': scale ({scale}) \
must be <= precision ({precision})"
)));
}
let mut meta = JsonMap::from_iter([
("logicalType".into(), json!("decimal")),
("precision".into(), json!(*precision)),
("scale".into(), json!(*scale)),
]);
let mut fixed_size = metadata.get("size").and_then(|v| v.parse::<usize>().ok());
let carries_name = metadata.contains_key(AVRO_NAME_METADATA_KEY)
|| metadata.contains_key(AVRO_NAMESPACE_METADATA_KEY);
if fixed_size.is_none() && carries_name {
fixed_size = Some(min_fixed_bytes_for_precision(*precision as usize));
}
if let Some(size) = fixed_size {
meta.insert("type".into(), json!("fixed"));
meta.insert("size".into(), json!(size));
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
meta.insert("name".into(), json!(chosen_name));
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
meta.insert("namespace".into(), json!(ns));
}
} else {
// default to bytes-backed decimal
meta.insert("type".into(), json!("bytes"));
}
Ok(Value::Object(meta))
};
let val = match dt {
DataType::Null => Value::String("null".into()),
DataType::Boolean => Value::String("boolean".into()),
#[cfg(not(feature = "avro_custom_types"))]
DataType::Int8 | DataType::Int16 | DataType::UInt8 | DataType::UInt16 => {
Value::String("int".into())
}
DataType::Int32 => Value::String("int".into()),
#[cfg(feature = "avro_custom_types")]
DataType::Int8 => json!({ "type": "int", "logicalType": "arrow.int8" }),
#[cfg(feature = "avro_custom_types")]
DataType::Int16 => json!({ "type": "int", "logicalType": "arrow.int16" }),
#[cfg(feature = "avro_custom_types")]
DataType::UInt8 => json!({ "type": "int", "logicalType": "arrow.uint8" }),
#[cfg(feature = "avro_custom_types")]
DataType::UInt16 => json!({ "type": "int", "logicalType": "arrow.uint16" }),
#[cfg(not(feature = "avro_custom_types"))]
DataType::UInt32 => Value::String("long".into()),
#[cfg(feature = "avro_custom_types")]
DataType::UInt32 => json!({ "type": "long", "logicalType": "arrow.uint32" }),
DataType::Int64 => Value::String("long".into()),
#[cfg(not(feature = "avro_custom_types"))]
DataType::UInt64 => Value::String("long".into()),
#[cfg(feature = "avro_custom_types")]
DataType::UInt64 => {
// UInt64 must use fixed(8) to avoid overflow
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut obj = JsonMap::from_iter([
("type".into(), json!("fixed")),
("name".into(), json!(chosen_name)),
("size".into(), json!(8)),
("logicalType".into(), json!("arrow.uint64")),
]);
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
obj.insert("namespace".into(), json!(ns));
}
json!(obj)
}
#[cfg(not(feature = "avro_custom_types"))]
DataType::Float16 => Value::String("float".into()),
#[cfg(feature = "avro_custom_types")]
DataType::Float16 => {
// Float16 uses fixed(2) for IEEE-754 bits
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut obj = JsonMap::from_iter([
("type".into(), json!("fixed")),
("name".into(), json!(chosen_name)),
("size".into(), json!(2)),
("logicalType".into(), json!("arrow.float16")),
]);
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
obj.insert("namespace".into(), json!(ns));
}
json!(obj)
}
DataType::Float32 => Value::String("float".into()),
DataType::Float64 => Value::String("double".into()),
DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => Value::String("string".into()),
DataType::Binary | DataType::LargeBinary => Value::String("bytes".into()),
DataType::BinaryView => {
if !strip {
extras.insert("arrowBinaryView".into(), Value::Bool(true));
}
Value::String("bytes".into())
}
DataType::FixedSizeBinary(len) => {
let md_is_uuid = metadata
.get("logicalType")
.map(|s| s.trim_matches('"') == "uuid")
.unwrap_or(false);
#[cfg(feature = "canonical_extension_types")]
let ext_is_uuid = metadata
.get(arrow_schema::extension::EXTENSION_TYPE_NAME_KEY)
.map(|v| v == arrow_schema::extension::Uuid::NAME || v == "uuid")
.unwrap_or(false);
#[cfg(not(feature = "canonical_extension_types"))]
let ext_is_uuid = false;
let is_uuid = (*len == 16) && (md_is_uuid || ext_is_uuid);
if is_uuid {
json!({ "type": "string", "logicalType": "uuid" })
} else {
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut obj = JsonMap::from_iter([
("type".into(), json!("fixed")),
("name".into(), json!(chosen_name)),
("size".into(), json!(len)),
]);
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
obj.insert("namespace".into(), json!(ns));
}
Value::Object(obj)
}
}
#[cfg(feature = "small_decimals")]
DataType::Decimal32(precision, scale) | DataType::Decimal64(precision, scale) => {
handle_decimal(precision, scale)?
}
DataType::Decimal128(precision, scale) | DataType::Decimal256(precision, scale) => {
handle_decimal(precision, scale)?
}
DataType::Date32 => json!({ "type": "int", "logicalType": "date" }),
#[cfg(not(feature = "avro_custom_types"))]
DataType::Date64 => json!({ "type": "long", "logicalType": "local-timestamp-millis" }),
#[cfg(feature = "avro_custom_types")]
DataType::Date64 => json!({ "type": "long", "logicalType": "arrow.date64" }),
DataType::Time32(unit) => match unit {
TimeUnit::Millisecond => json!({ "type": "int", "logicalType": "time-millis" }),
#[cfg(not(feature = "avro_custom_types"))]
TimeUnit::Second => {
// Encoder converts seconds to milliseconds, so use time-millis
if !strip {
extras.insert("arrowTimeUnit".into(), Value::String("second".into()));
}
json!({ "type": "int", "logicalType": "time-millis" })
}
#[cfg(feature = "avro_custom_types")]
TimeUnit::Second => {
json!({ "type": "int", "logicalType": "arrow.time32-second" })
}
_ => Value::String("int".into()),
},
DataType::Time64(unit) => match unit {
TimeUnit::Microsecond => json!({ "type": "long", "logicalType": "time-micros" }),
#[cfg(not(feature = "avro_custom_types"))]
TimeUnit::Nanosecond => {
// Encoder truncates nanoseconds to microseconds, so use time-micros
if !strip {
extras.insert("arrowTimeUnit".into(), Value::String("nanosecond".into()));
}
json!({ "type": "long", "logicalType": "time-micros" })
}
#[cfg(feature = "avro_custom_types")]
TimeUnit::Nanosecond => {
json!({ "type": "long", "logicalType": "arrow.time64-nanosecond" })
}
_ => Value::String("long".into()),
},
DataType::Timestamp(unit, tz) => {
#[cfg(feature = "avro_custom_types")]
if matches!(unit, TimeUnit::Second) {
let logical_type = if tz.is_some() {
"arrow.timestamp-second"
} else {
"arrow.local-timestamp-second"
};
return Ok((
json!({ "type": "long", "logicalType": logical_type }),
extras,
));
}
let logical_type = match (unit, tz.is_some()) {
(TimeUnit::Millisecond, true) => "timestamp-millis",
(TimeUnit::Millisecond, false) => "local-timestamp-millis",
(TimeUnit::Microsecond, true) => "timestamp-micros",
(TimeUnit::Microsecond, false) => "local-timestamp-micros",
(TimeUnit::Nanosecond, true) => "timestamp-nanos",
(TimeUnit::Nanosecond, false) => "local-timestamp-nanos",
(TimeUnit::Second, has_tz) => {
// Encoder converts seconds to milliseconds, so use timestamp-millis
if !strip {
extras.insert("arrowTimeUnit".into(), Value::String("second".into()));
}
let ts_logical_type = if has_tz {
"timestamp-millis"
} else {
"local-timestamp-millis"
};
return Ok((
json!({ "type": "long", "logicalType": ts_logical_type }),
extras,
));
}
};
if !strip && matches!(unit, TimeUnit::Nanosecond) {
extras.insert("arrowTimeUnit".into(), Value::String("nanosecond".into()));
}
json!({ "type": "long", "logicalType": logical_type })
}
#[cfg(not(feature = "avro_custom_types"))]
DataType::Duration(_unit) => Value::String("long".into()),
#[cfg(feature = "avro_custom_types")]
DataType::Duration(unit) => {
// When the feature is enabled, create an Avro schema object
// with the correct `logicalType` annotation.
let logical_type = match unit {
TimeUnit::Second => "arrow.duration-seconds",
TimeUnit::Millisecond => "arrow.duration-millis",
TimeUnit::Microsecond => "arrow.duration-micros",
TimeUnit::Nanosecond => "arrow.duration-nanos",
};
json!({ "type": "long", "logicalType": logical_type })
}
#[cfg(not(feature = "avro_custom_types"))]
DataType::Interval(IntervalUnit::MonthDayNano) => {
// Avro duration logical type: fixed(12) with months/days/millis per spec.
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut obj = JsonMap::from_iter([
("type".into(), json!("fixed")),
("name".into(), json!(chosen_name)),
("size".into(), json!(12)),
("logicalType".into(), json!("duration")),
]);
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
obj.insert("namespace".into(), json!(ns));
}
json!(obj)
}
#[cfg(feature = "avro_custom_types")]
DataType::Interval(IntervalUnit::MonthDayNano) => {
// Arrow MonthDayNano interval: i32 months + i32 days + i64 nanos (16 bytes).
// We preserve the Arrow native representation via a custom logical type.
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut obj = JsonMap::from_iter([
("type".into(), json!("fixed")),
("name".into(), json!(chosen_name)),
("size".into(), json!(16)),
("logicalType".into(), json!("arrow.interval-month-day-nano")),
]);
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
obj.insert("namespace".into(), json!(ns));
}
json!(obj)
}
#[cfg(not(feature = "avro_custom_types"))]
DataType::Interval(IntervalUnit::YearMonth) => {
// Encode as Avro `duration` (fixed(12)) like MonthDayNano
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut extras = JsonMap::from_iter([
("type".into(), json!("fixed")),
("name".into(), json!(chosen_name)),
("size".into(), json!(12)),
("logicalType".into(), json!("duration")),
]);
if !strip {
extras.insert(
"arrowIntervalUnit".into(),
Value::String("yearmonth".into()),
);
}
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
extras.insert("namespace".into(), json!(ns));
}
json!(extras)
}
#[cfg(feature = "avro_custom_types")]
DataType::Interval(IntervalUnit::YearMonth) => {
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut obj = JsonMap::from_iter([
("type".into(), json!("fixed")),
("name".into(), json!(chosen_name)),
("size".into(), json!(4)),
("logicalType".into(), json!("arrow.interval-year-month")),
]);
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
obj.insert("namespace".into(), json!(ns));
}
json!(obj)
}
#[cfg(not(feature = "avro_custom_types"))]
DataType::Interval(IntervalUnit::DayTime) => {
// Encode as Avro `duration` (fixed(12)) like MonthDayNano
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut obj = JsonMap::from_iter([
("type".into(), json!("fixed")),
("name".into(), json!(chosen_name)),
("size".into(), json!(12)),
("logicalType".into(), json!("duration")),
]);
if !strip {
obj.insert("arrowIntervalUnit".into(), Value::String("daytime".into()));
}
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
obj.insert("namespace".into(), json!(ns));
}
json!(obj)
}
#[cfg(feature = "avro_custom_types")]
DataType::Interval(IntervalUnit::DayTime) => {
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut obj = JsonMap::from_iter([
("type".into(), json!("fixed")),
("name".into(), json!(chosen_name)),
("size".into(), json!(8)),
("logicalType".into(), json!("arrow.interval-day-time")),
]);
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
obj.insert("namespace".into(), json!(ns));
}
json!(obj)
}
DataType::List(child) | DataType::LargeList(child) => {
if matches!(dt, DataType::LargeList(_)) && !strip {
extras.insert("arrowLargeList".into(), Value::Bool(true));
}
let items_schema = process_datatype(
child.data_type(),
child.name(),
child.metadata(),
name_gen,
null_order,
child.is_nullable(),
strip,
)?;
json!({
"type": "array",
"items": items_schema
})
}
DataType::ListView(child) | DataType::LargeListView(child) => {
if matches!(dt, DataType::LargeListView(_)) && !strip {
extras.insert("arrowLargeList".into(), Value::Bool(true));
}
if !strip {
extras.insert("arrowListView".into(), Value::Bool(true));
}
let items_schema = process_datatype(
child.data_type(),
child.name(),
child.metadata(),
name_gen,
null_order,
child.is_nullable(),
strip,
)?;
json!({
"type": "array",
"items": items_schema
})
}
DataType::FixedSizeList(child, len) => {
if !strip {
extras.insert("arrowFixedSize".into(), json!(len));
}
let items_schema = process_datatype(
child.data_type(),
child.name(),
child.metadata(),
name_gen,
null_order,
child.is_nullable(),
strip,
)?;
json!({
"type": "array",
"items": items_schema
})
}
DataType::Map(entries, _) => {
let value_field = match entries.data_type() {
DataType::Struct(fs) => &fs[1],
_ => {
return Err(ArrowError::SchemaError(
"Map 'entries' field must be Struct(key,value)".into(),
));
}
};
let values_schema = process_datatype(
value_field.data_type(),
value_field.name(),
value_field.metadata(),
name_gen,
null_order,
value_field.is_nullable(),
strip,
)?;
json!({
"type": "map",
"values": values_schema
})
}
DataType::Struct(fields) => {
let avro_fields = fields
.iter()
.map(|field| arrow_field_to_avro(field, name_gen, null_order, strip))
.collect::<Result<Vec<_>, _>>()?;
// Prefer avro.name/avro.namespace when provided on the struct field metadata
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut obj = JsonMap::from_iter([
("type".into(), json!("record")),
("name".into(), json!(chosen_name)),
("fields".into(), Value::Array(avro_fields)),
]);
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
obj.insert("namespace".into(), json!(ns));
}
Value::Object(obj)
}
DataType::Dictionary(_, value) => {
if let Some(j) = metadata.get(AVRO_ENUM_SYMBOLS_METADATA_KEY) {
let symbols: Vec<&str> =
serde_json::from_str(j).map_err(|e| ArrowError::ParseError(e.to_string()))?;
// Prefer avro.name/namespace when provided for enums
let chosen_name = metadata
.get(AVRO_NAME_METADATA_KEY)
.map(|s| sanitise_avro_name(s))
.unwrap_or_else(|| name_gen.make_unique(field_name));
let mut obj = JsonMap::from_iter([
("type".into(), json!("enum")),
("name".into(), json!(chosen_name)),
("symbols".into(), json!(symbols)),
]);
if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
obj.insert("namespace".into(), json!(ns));
}
Value::Object(obj)
} else {
process_datatype(
value.as_ref(),
field_name,
metadata,
name_gen,
null_order,
false,
strip,
)?
}
}
#[cfg(feature = "avro_custom_types")]
DataType::RunEndEncoded(run_ends, values) => {
let bits = match run_ends.data_type() {
DataType::Int16 => 16,
DataType::Int32 => 32,
DataType::Int64 => 64,
other => {
return Err(ArrowError::SchemaError(format!(
"RunEndEncoded requires Int16/Int32/Int64 for run_ends, found: {other:?}"
)));
}
};
// Build the value site schema, preserving its own nullability
let (value_schema, value_extras) = datatype_to_avro(
values.data_type(),
values.name(),
values.metadata(),
name_gen,
null_order,
strip,
)?;
let mut merged = merge_extras(value_schema, value_extras);
if values.is_nullable() {
merged = wrap_nullable(merged, null_order);
}
let mut extras = JsonMap::new();
extras.insert("logicalType".into(), json!("arrow.run-end-encoded"));
extras.insert("arrow.runEndIndexBits".into(), json!(bits));
return Ok((merged, extras));
}
#[cfg(not(feature = "avro_custom_types"))]
DataType::RunEndEncoded(_run_ends, values) => {
let (value_schema, _extras) = datatype_to_avro(
values.data_type(),
values.name(),
values.metadata(),
name_gen,
null_order,
strip,
)?;
return Ok((value_schema, JsonMap::new()));
}
DataType::Union(fields, mode) => {
let mut branches: Vec<Value> = Vec::with_capacity(fields.len());
let mut type_ids: Vec<i32> = Vec::with_capacity(fields.len());
for (type_id, field_ref) in fields.iter() {
// NOTE: `process_datatype` would wrap nullability; force is_nullable=false here.
let (branch_schema, _branch_extras) = datatype_to_avro(
field_ref.data_type(),
field_ref.name(),
field_ref.metadata(),
name_gen,
null_order,
strip,
)?;
// Avro unions cannot immediately contain another union
if matches!(branch_schema, Value::Array(_)) {
return Err(ArrowError::SchemaError(
"Avro union may not immediately contain another union".into(),
));
}
branches.push(branch_schema);
type_ids.push(type_id as i32);
}
let mut seen: HashSet<String> = HashSet::with_capacity(branches.len());
for b in &branches {
let sig = union_branch_signature(b)?;
if !seen.insert(sig) {
return Err(ArrowError::SchemaError(
"Avro union contains duplicate branch types (disallowed by spec)".into(),
));
}
}
if !strip {
extras.insert(
"arrowUnionMode".into(),
Value::String(
match mode {
UnionMode::Sparse => "sparse",
UnionMode::Dense => "dense",
}
.to_string(),
),
);
extras.insert(
"arrowUnionTypeIds".into(),
Value::Array(type_ids.into_iter().map(|id| json!(id)).collect()),
);
}
Value::Array(branches)
}
#[cfg(not(feature = "small_decimals"))]
other => {
return Err(ArrowError::NotYetImplemented(format!(
"Arrow type {other:?} has no Avro representation"
)));
}
};
Ok((val, extras))
}
fn process_datatype(
dt: &DataType,
field_name: &str,
metadata: &HashMap<String, String>,
name_gen: &mut NameGenerator,
null_order: Nullability,
is_nullable: bool,
strip: bool,
) -> Result<Value, ArrowError> {
let (schema, extras) = datatype_to_avro(dt, field_name, metadata, name_gen, null_order, strip)?;
let mut merged = merge_extras(schema, extras);
if is_nullable {
merged = wrap_nullable(merged, null_order)
}
Ok(merged)
}
fn arrow_field_to_avro(
field: &ArrowField,
name_gen: &mut NameGenerator,
null_order: Nullability,
strip: bool,
) -> Result<Value, ArrowError> {
let avro_name = sanitise_avro_name(field.name());
let schema_value = process_datatype(
field.data_type(),
&avro_name,
field.metadata(),
name_gen,
null_order,
field.is_nullable(),
strip,
)?;
// Build the field map
let mut map = JsonMap::with_capacity(field.metadata().len() + 3);
map.insert("name".into(), Value::String(avro_name));
map.insert("type".into(), schema_value);
// Transfer selected metadata
for (meta_key, meta_val) in field.metadata() {
if is_internal_arrow_key(meta_key) {
continue;
}
match meta_key.as_str() {
AVRO_DOC_METADATA_KEY => {
map.insert("doc".into(), Value::String(meta_val.clone()));
}
AVRO_FIELD_DEFAULT_METADATA_KEY => {
let default_value = serde_json::from_str(meta_val)
.unwrap_or_else(|_| Value::String(meta_val.clone()));
map.insert("default".into(), default_value);
}
_ => {
let json_val = serde_json::from_str(meta_val)
.unwrap_or_else(|_| Value::String(meta_val.clone()));
map.insert(meta_key.clone(), json_val);
}
}
}
Ok(Value::Object(map))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::codec::{AvroField, AvroFieldBuilder};
use arrow_schema::{DataType, Fields, SchemaBuilder, TimeUnit, UnionFields};
use serde_json::json;
use std::sync::Arc;
fn int_schema() -> Schema<'static> {
Schema::TypeName(TypeName::Primitive(PrimitiveType::Int))
}
fn record_schema() -> Schema<'static> {
Schema::Complex(ComplexType::Record(Record {
name: "record1",
namespace: Some("test.namespace"),
doc: Some(Cow::from("A test record")),
aliases: vec![],
fields: vec![
Field {
name: "field1",
doc: Some(Cow::from("An integer field")),
r#type: int_schema(),
default: None,
aliases: vec![],
},
Field {
name: "field2",
doc: None,
r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
default: None,
aliases: vec![],
},
],
attributes: Attributes::default(),
}))
}
fn single_field_schema(field: ArrowField) -> arrow_schema::Schema {
let mut sb = SchemaBuilder::new();
sb.push(field);
sb.finish()
}
fn assert_json_contains(avro_json: &str, needle: &str) {
assert!(
avro_json.contains(needle),
"JSON did not contain `{needle}` : {avro_json}"
)
}
#[test]
fn test_deserialize() {
let t: Schema = serde_json::from_str("\"string\"").unwrap();
assert_eq!(
t,
Schema::TypeName(TypeName::Primitive(PrimitiveType::String))
);
let t: Schema = serde_json::from_str("[\"int\", \"null\"]").unwrap();
assert_eq!(
t,
Schema::Union(vec![
Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
])
);
let t: Type = serde_json::from_str(
r#"{
"type":"long",
"logicalType":"timestamp-micros"
}"#,
)
.unwrap();
let timestamp = Type {
r#type: TypeName::Primitive(PrimitiveType::Long),
attributes: Attributes {
logical_type: Some("timestamp-micros"),
additional: Default::default(),
},
};
assert_eq!(t, timestamp);
let t: ComplexType = serde_json::from_str(
r#"{
"type":"fixed",
"name":"fixed",
"namespace":"topLevelRecord.value",
"size":11,
"logicalType":"decimal",
"precision":25,
"scale":2
}"#,
)
.unwrap();
let decimal = ComplexType::Fixed(Fixed {
name: "fixed",
namespace: Some("topLevelRecord.value"),
aliases: vec![],
size: 11,
attributes: Attributes {
logical_type: Some("decimal"),
additional: vec![("precision", json!(25)), ("scale", json!(2))]
.into_iter()
.collect(),
},
});
assert_eq!(t, decimal);
let schema: Schema = serde_json::from_str(
r#"{
"type":"record",
"name":"topLevelRecord",
"fields":[
{
"name":"value",
"type":[
{
"type":"fixed",
"name":"fixed",
"namespace":"topLevelRecord.value",
"size":11,
"logicalType":"decimal",
"precision":25,
"scale":2
},
"null"
]
}
]
}"#,
)
.unwrap();
assert_eq!(
schema,
Schema::Complex(ComplexType::Record(Record {
name: "topLevelRecord",
namespace: None,
doc: None,
aliases: vec![],
fields: vec![Field {
name: "value",
doc: None,
r#type: Schema::Union(vec![
Schema::Complex(decimal),
Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
]),
default: None,
aliases: vec![],
},],
attributes: Default::default(),
}))
);
let schema: Schema = serde_json::from_str(
r#"{
"type": "record",
"name": "LongList",
"aliases": ["LinkedLongs"],
"fields" : [
{"name": "value", "type": "long"},
{"name": "next", "type": ["null", "LongList"]}
]
}"#,
)
.unwrap();
assert_eq!(
schema,
Schema::Complex(ComplexType::Record(Record {
name: "LongList",
namespace: None,
doc: None,
aliases: vec!["LinkedLongs"],
fields: vec![
Field {
name: "value",
doc: None,
r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
default: None,
aliases: vec![],
},
Field {
name: "next",
doc: None,
r#type: Schema::Union(vec![
Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
Schema::TypeName(TypeName::Ref("LongList")),
]),
default: None,
aliases: vec![],
}
],
attributes: Attributes::default(),
}))
);
// Recursive schema are not supported
let err = AvroField::try_from(&schema).unwrap_err().to_string();
assert_eq!(err, "Parser error: Failed to resolve .LongList");
let schema: Schema = serde_json::from_str(
r#"{
"type":"record",
"name":"topLevelRecord",
"fields":[
{
"name":"id",
"type":[
"int",
"null"
]
},
{
"name":"timestamp_col",
"type":[
{
"type":"long",
"logicalType":"timestamp-micros"
},
"null"
]
}
]
}"#,
)
.unwrap();
assert_eq!(
schema,
Schema::Complex(ComplexType::Record(Record {
name: "topLevelRecord",
namespace: None,
doc: None,
aliases: vec![],
fields: vec![
Field {
name: "id",
doc: None,
r#type: Schema::Union(vec![
Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
]),
default: None,
aliases: vec![],
},
Field {
name: "timestamp_col",
doc: None,
r#type: Schema::Union(vec![
Schema::Type(timestamp),
Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
]),
default: None,
aliases: vec![],
}
],
attributes: Default::default(),
}))
);
let codec = AvroField::try_from(&schema).unwrap();
let expected_arrow_field = arrow_schema::Field::new(
"topLevelRecord",
DataType::Struct(Fields::from(vec![
arrow_schema::Field::new("id", DataType::Int32, true),
arrow_schema::Field::new(
"timestamp_col",
DataType::Timestamp(TimeUnit::Microsecond, Some("+00:00".into())),
true,
),
])),
false,
)
.with_metadata(std::collections::HashMap::from([(
AVRO_NAME_METADATA_KEY.to_string(),
"topLevelRecord".to_string(),
)]));
assert_eq!(codec.field(), expected_arrow_field);
let schema: Schema = serde_json::from_str(
r#"{
"type": "record",
"name": "HandshakeRequest", "namespace":"org.apache.avro.ipc",
"fields": [
{"name": "clientHash", "type": {"type": "fixed", "name": "MD5", "size": 16}},
{"name": "clientProtocol", "type": ["null", "string"]},
{"name": "serverHash", "type": "MD5"},
{"name": "meta", "type": ["null", {"type": "map", "values": "bytes"}]}
]
}"#,
)
.unwrap();
assert_eq!(
schema,
Schema::Complex(ComplexType::Record(Record {
name: "HandshakeRequest",
namespace: Some("org.apache.avro.ipc"),
doc: None,
aliases: vec![],
fields: vec![
Field {
name: "clientHash",
doc: None,
r#type: Schema::Complex(ComplexType::Fixed(Fixed {
name: "MD5",
namespace: None,
aliases: vec![],
size: 16,
attributes: Default::default(),
})),
default: None,
aliases: vec![],
},
Field {
name: "clientProtocol",
doc: None,
r#type: Schema::Union(vec![
Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
]),
default: None,
aliases: vec![],
},
Field {
name: "serverHash",
doc: None,
r#type: Schema::TypeName(TypeName::Ref("MD5")),
default: None,
aliases: vec![],
},
Field {
name: "meta",
doc: None,
r#type: Schema::Union(vec![
Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
Schema::Complex(ComplexType::Map(Map {
values: Box::new(Schema::TypeName(TypeName::Primitive(
PrimitiveType::Bytes
))),
attributes: Default::default(),
})),
]),
default: None,
aliases: vec![],
}
],
attributes: Default::default(),
}))
);
}
#[test]
fn test_canonical_form_generation_comprehensive_record() {
// NOTE: This schema is identical to the one used in test_deserialize_comprehensive.
let json_str = r#"{
"type": "record",
"name": "E2eComprehensive",
"namespace": "org.apache.arrow.avrotests.v1",
"doc": "Comprehensive Avro writer schema to exercise arrow-avro Reader/Decoder paths.",
"fields": [
{"name": "id", "type": "long", "doc": "Primary row id", "aliases": ["identifier"]},
{"name": "flag", "type": "boolean", "default": true, "doc": "A sample boolean with default true"},
{"name": "ratio_f32", "type": "float", "default": 0.0, "doc": "Float32 example"},
{"name": "ratio_f64", "type": "double", "default": 0.0, "doc": "Float64 example"},
{"name": "count_i32", "type": "int", "default": 0, "doc": "Int32 example"},
{"name": "count_i64", "type": "long", "default": 0, "doc": "Int64 example"},
{"name": "opt_i32_nullfirst", "type": ["null", "int"], "default": null, "doc": "Nullable int (null-first)"},
{"name": "opt_str_nullsecond", "type": ["string", "null"], "default": "", "aliases": ["old_opt_str"], "doc": "Nullable string (null-second). Default is empty string."},
{"name": "tri_union_prim", "type": ["int", "string", "boolean"], "default": 0, "doc": "Union[int, string, boolean] with default on first branch (int=0)."},
{"name": "str_utf8", "type": "string", "default": "default", "doc": "Plain Utf8 string (Reader may use Utf8View)."},
{"name": "raw_bytes", "type": "bytes", "default": "", "doc": "Raw bytes field"},
{"name": "fx16_plain", "type": {"type": "fixed", "name": "Fx16", "namespace": "org.apache.arrow.avrotests.v1.types", "aliases": ["Fixed16Old"], "size": 16}, "doc": "Plain fixed(16)"},
{"name": "dec_bytes_s10_2", "type": {"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2}, "doc": "Decimal encoded on bytes, precision 10, scale 2"},
{"name": "dec_fix_s20_4", "type": {"type": "fixed", "name": "DecFix20", "namespace": "org.apache.arrow.avrotests.v1.types", "size": 20, "logicalType": "decimal", "precision": 20, "scale": 4}, "doc": "Decimal encoded on fixed(20), precision 20, scale 4"},
{"name": "uuid_str", "type": {"type": "string", "logicalType": "uuid"}, "doc": "UUID logical type on string"},
{"name": "d_date", "type": {"type": "int", "logicalType": "date"}, "doc": "Date32: days since 1970-01-01"},
{"name": "t_millis", "type": {"type": "int", "logicalType": "time-millis"}, "doc": "Time32-millis"},
{"name": "t_micros", "type": {"type": "long", "logicalType": "time-micros"}, "doc": "Time64-micros"},
{"name": "ts_millis_utc", "type": {"type": "long", "logicalType": "timestamp-millis"}, "doc": "Timestamp ms (UTC)"},
{"name": "ts_micros_utc", "type": {"type": "long", "logicalType": "timestamp-micros"}, "doc": "Timestamp µs (UTC)"},
{"name": "ts_millis_local", "type": {"type": "long", "logicalType": "local-timestamp-millis"}, "doc": "Local timestamp ms"},
{"name": "ts_micros_local", "type": {"type": "long", "logicalType": "local-timestamp-micros"}, "doc": "Local timestamp µs"},
{"name": "interval_mdn", "type": {"type": "fixed", "name": "Dur12", "namespace": "org.apache.arrow.avrotests.v1.types", "size": 12, "logicalType": "duration"}, "doc": "Duration: fixed(12) little-endian (months, days, millis)"},
{"name": "status", "type": {"type": "enum", "name": "Status", "namespace": "org.apache.arrow.avrotests.v1.types", "symbols": ["UNKNOWN", "NEW", "PROCESSING", "DONE"], "aliases": ["State"], "doc": "Processing status enum with default"}, "default": "UNKNOWN", "doc": "Enum field using default when resolving"},
{"name": "arr_union", "type": {"type": "array", "items": ["long", "string", "null"]}, "default": [], "doc": "Array whose items are a union[long,string,null]"},
{"name": "map_union", "type": {"type": "map", "values": ["null", "double", "string"]}, "default": {}, "doc": "Map whose values are a union[null,double,string]"},
{"name": "address", "type": {"type": "record", "name": "Address", "namespace": "org.apache.arrow.avrotests.v1.types", "doc": "Postal address with defaults and field alias", "fields": [
{"name": "street", "type": "string", "default": "", "aliases": ["street_name"], "doc": "Street (field alias = street_name)"},
{"name": "zip", "type": "int", "default": 0, "doc": "ZIP/postal code"},
{"name": "country", "type": "string", "default": "US", "doc": "Country code"}
]}, "doc": "Embedded Address record"},
{"name": "maybe_auth", "type": {"type": "record", "name": "MaybeAuth", "namespace": "org.apache.arrow.avrotests.v1.types", "doc": "Optional auth token model", "fields": [
{"name": "user", "type": "string", "doc": "Username"},
{"name": "token", "type": ["null", "bytes"], "default": null, "doc": "Nullable auth token"}
]}},
{"name": "union_enum_record_array_map", "type": [
{"type": "enum", "name": "Color", "namespace": "org.apache.arrow.avrotests.v1.types", "symbols": ["RED", "GREEN", "BLUE"], "doc": "Color enum"},
{"type": "record", "name": "RecA", "namespace": "org.apache.arrow.avrotests.v1.types", "fields": [{"name": "a", "type": "int"}, {"name": "b", "type": "string"}]},
{"type": "record", "name": "RecB", "namespace": "org.apache.arrow.avrotests.v1.types", "fields": [{"name": "x", "type": "long"}, {"name": "y", "type": "bytes"}]},
{"type": "array", "items": "long"},
{"type": "map", "values": "string"}
], "doc": "Union of enum, two records, array, and map"},
{"name": "union_date_or_fixed4", "type": [
{"type": "int", "logicalType": "date"},
{"type": "fixed", "name": "Fx4", "size": 4}
], "doc": "Union of date(int) or fixed(4)"},
{"name": "union_interval_or_string", "type": [
{"type": "fixed", "name": "Dur12U", "size": 12, "logicalType": "duration"},
"string"
], "doc": "Union of duration(fixed12) or string"},
{"name": "union_uuid_or_fixed10", "type": [
{"type": "string", "logicalType": "uuid"},
{"type": "fixed", "name": "Fx10", "size": 10}
], "doc": "Union of UUID string or fixed(10)"},
{"name": "array_records_with_union", "type": {"type": "array", "items": {
"type": "record", "name": "KV", "namespace": "org.apache.arrow.avrotests.v1.types",
"fields": [
{"name": "key", "type": "string"},
{"name": "val", "type": ["null", "int", "long"], "default": null}
]
}}, "doc": "Array<record{key, val: union[null,int,long]}>", "default": []},
{"name": "union_map_or_array_int", "type": [
{"type": "map", "values": "int"},
{"type": "array", "items": "int"}
], "doc": "Union[map<string,int>, array<int>]"},
{"name": "renamed_with_default", "type": "int", "default": 42, "aliases": ["old_count"], "doc": "Field with alias and default"},
{"name": "person", "type": {"type": "record", "name": "PersonV2", "namespace": "com.example.v2", "aliases": ["com.example.Person"], "doc": "Person record with alias pointing to previous namespace/name", "fields": [
{"name": "name", "type": "string"},
{"name": "age", "type": "int", "default": 0}
]}, "doc": "Record using type alias for schema evolution tests"}
]
}"#;
let avro = AvroSchema::new(json_str.to_string());
let parsed = avro.schema().expect("schema should deserialize");
let expected_canonical_form = r#"{"name":"org.apache.arrow.avrotests.v1.E2eComprehensive","type":"record","fields":[{"name":"id","type":"long"},{"name":"flag","type":"boolean"},{"name":"ratio_f32","type":"float"},{"name":"ratio_f64","type":"double"},{"name":"count_i32","type":"int"},{"name":"count_i64","type":"long"},{"name":"opt_i32_nullfirst","type":["null","int"]},{"name":"opt_str_nullsecond","type":["string","null"]},{"name":"tri_union_prim","type":["int","string","boolean"]},{"name":"str_utf8","type":"string"},{"name":"raw_bytes","type":"bytes"},{"name":"fx16_plain","type":{"name":"org.apache.arrow.avrotests.v1.types.Fx16","type":"fixed","size":16}},{"name":"dec_bytes_s10_2","type":"bytes"},{"name":"dec_fix_s20_4","type":{"name":"org.apache.arrow.avrotests.v1.types.DecFix20","type":"fixed","size":20}},{"name":"uuid_str","type":"string"},{"name":"d_date","type":"int"},{"name":"t_millis","type":"int"},{"name":"t_micros","type":"long"},{"name":"ts_millis_utc","type":"long"},{"name":"ts_micros_utc","type":"long"},{"name":"ts_millis_local","type":"long"},{"name":"ts_micros_local","type":"long"},{"name":"interval_mdn","type":{"name":"org.apache.arrow.avrotests.v1.types.Dur12","type":"fixed","size":12}},{"name":"status","type":{"name":"org.apache.arrow.avrotests.v1.types.Status","type":"enum","symbols":["UNKNOWN","NEW","PROCESSING","DONE"]}},{"name":"arr_union","type":{"type":"array","items":["long","string","null"]}},{"name":"map_union","type":{"type":"map","values":["null","double","string"]}},{"name":"address","type":{"name":"org.apache.arrow.avrotests.v1.types.Address","type":"record","fields":[{"name":"street","type":"string"},{"name":"zip","type":"int"},{"name":"country","type":"string"}]}},{"name":"maybe_auth","type":{"name":"org.apache.arrow.avrotests.v1.types.MaybeAuth","type":"record","fields":[{"name":"user","type":"string"},{"name":"token","type":["null","bytes"]}]}},{"name":"union_enum_record_array_map","type":[{"name":"org.apache.arrow.avrotests.v1.types.Color","type":"enum","symbols":["RED","GREEN","BLUE"]},{"name":"org.apache.arrow.avrotests.v1.types.RecA","type":"record","fields":[{"name":"a","type":"int"},{"name":"b","type":"string"}]},{"name":"org.apache.arrow.avrotests.v1.types.RecB","type":"record","fields":[{"name":"x","type":"long"},{"name":"y","type":"bytes"}]},{"type":"array","items":"long"},{"type":"map","values":"string"}]},{"name":"union_date_or_fixed4","type":["int",{"name":"org.apache.arrow.avrotests.v1.Fx4","type":"fixed","size":4}]},{"name":"union_interval_or_string","type":[{"name":"org.apache.arrow.avrotests.v1.Dur12U","type":"fixed","size":12},"string"]},{"name":"union_uuid_or_fixed10","type":["string",{"name":"org.apache.arrow.avrotests.v1.Fx10","type":"fixed","size":10}]},{"name":"array_records_with_union","type":{"type":"array","items":{"name":"org.apache.arrow.avrotests.v1.types.KV","type":"record","fields":[{"name":"key","type":"string"},{"name":"val","type":["null","int","long"]}]}}},{"name":"union_map_or_array_int","type":[{"type":"map","values":"int"},{"type":"array","items":"int"}]},{"name":"renamed_with_default","type":"int"},{"name":"person","type":{"name":"com.example.v2.PersonV2","type":"record","fields":[{"name":"name","type":"string"},{"name":"age","type":"int"}]}}]}"#;
let canonical_form =
AvroSchema::generate_canonical_form(&parsed).expect("canonical form should be built");
assert_eq!(
canonical_form, expected_canonical_form,
"Canonical form must match Avro spec PCF exactly"
);
}
#[test]
fn test_new_schema_store() {
let store = SchemaStore::new();
assert!(store.schemas.is_empty());
}
#[test]
fn test_try_from_schemas_rabin() {
let int_avro_schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
let record_avro_schema = AvroSchema::new(serde_json::to_string(&record_schema()).unwrap());
let mut schemas: HashMap<Fingerprint, AvroSchema> = HashMap::new();
schemas.insert(
int_avro_schema
.fingerprint(FingerprintAlgorithm::Rabin)
.unwrap(),
int_avro_schema.clone(),
);
schemas.insert(
record_avro_schema
.fingerprint(FingerprintAlgorithm::Rabin)
.unwrap(),
record_avro_schema.clone(),
);
let store = SchemaStore::try_from(schemas).unwrap();
let int_fp = int_avro_schema
.fingerprint(FingerprintAlgorithm::Rabin)
.unwrap();
assert_eq!(store.lookup(&int_fp).cloned(), Some(int_avro_schema));
let rec_fp = record_avro_schema
.fingerprint(FingerprintAlgorithm::Rabin)
.unwrap();
assert_eq!(store.lookup(&rec_fp).cloned(), Some(record_avro_schema));
}
#[test]
fn test_try_from_with_duplicates() {
let int_avro_schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
let record_avro_schema = AvroSchema::new(serde_json::to_string(&record_schema()).unwrap());
let mut schemas: HashMap<Fingerprint, AvroSchema> = HashMap::new();
schemas.insert(
int_avro_schema
.fingerprint(FingerprintAlgorithm::Rabin)
.unwrap(),
int_avro_schema.clone(),
);
schemas.insert(
record_avro_schema
.fingerprint(FingerprintAlgorithm::Rabin)
.unwrap(),
record_avro_schema.clone(),
);
// Insert duplicate of int schema
schemas.insert(
int_avro_schema
.fingerprint(FingerprintAlgorithm::Rabin)
.unwrap(),
int_avro_schema.clone(),
);
let store = SchemaStore::try_from(schemas).unwrap();
assert_eq!(store.schemas.len(), 2);
let int_fp = int_avro_schema
.fingerprint(FingerprintAlgorithm::Rabin)
.unwrap();
assert_eq!(store.lookup(&int_fp).cloned(), Some(int_avro_schema));
}
#[test]
fn test_register_and_lookup_rabin() {
let mut store = SchemaStore::new();
let schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
let fp_enum = store.register(schema.clone()).unwrap();
match fp_enum {
Fingerprint::Rabin(fp_val) => {
assert_eq!(
store.lookup(&Fingerprint::Rabin(fp_val)).cloned(),
Some(schema.clone())
);
assert!(
store
.lookup(&Fingerprint::Rabin(fp_val.wrapping_add(1)))
.is_none()
);
}
Fingerprint::Id(_id) => {
unreachable!("This test should only generate Rabin fingerprints")
}
Fingerprint::Id64(_id) => {
unreachable!("This test should only generate Rabin fingerprints")
}
#[cfg(feature = "md5")]
Fingerprint::MD5(_id) => {
unreachable!("This test should only generate Rabin fingerprints")
}
#[cfg(feature = "sha256")]
Fingerprint::SHA256(_id) => {
unreachable!("This test should only generate Rabin fingerprints")
}
}
}
#[test]
fn test_set_and_lookup_id() {
let mut store = SchemaStore::new();
let schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
let id = 42u32;
let fp = Fingerprint::Id(id);
let out_fp = store.set(fp, schema.clone()).unwrap();
assert_eq!(out_fp, fp);
assert_eq!(store.lookup(&fp).cloned(), Some(schema.clone()));
assert!(store.lookup(&Fingerprint::Id(id.wrapping_add(1))).is_none());
}
#[test]
fn test_set_and_lookup_id64() {
let mut store = SchemaStore::new();
let schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
let id64: u64 = 0xDEAD_BEEF_DEAD_BEEF;
let fp = Fingerprint::Id64(id64);
let out_fp = store.set(fp, schema.clone()).unwrap();
assert_eq!(out_fp, fp, "set should return the same Id64 fingerprint");
assert_eq!(
store.lookup(&fp).cloned(),
Some(schema.clone()),
"lookup should find the schema by Id64"
);
assert!(
store
.lookup(&Fingerprint::Id64(id64.wrapping_add(1)))
.is_none(),
"lookup with a different Id64 must return None"
);
}
#[test]
fn test_fingerprint_id64_conversions() {
let algo_from_fp = FingerprintAlgorithm::from(&Fingerprint::Id64(123));
assert_eq!(algo_from_fp, FingerprintAlgorithm::Id64);
let fp_from_algo = Fingerprint::from(FingerprintAlgorithm::Id64);
assert!(matches!(fp_from_algo, Fingerprint::Id64(0)));
let strategy_from_fp = FingerprintStrategy::from(Fingerprint::Id64(5));
assert!(matches!(strategy_from_fp, FingerprintStrategy::Id64(0)));
let algo_from_strategy = FingerprintAlgorithm::from(strategy_from_fp);
assert_eq!(algo_from_strategy, FingerprintAlgorithm::Id64);
}
#[test]
fn test_register_duplicate_schema() {
let mut store = SchemaStore::new();
let schema1 = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
let schema2 = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
let fingerprint1 = store.register(schema1).unwrap();
let fingerprint2 = store.register(schema2).unwrap();
assert_eq!(fingerprint1, fingerprint2);
assert_eq!(store.schemas.len(), 1);
}
#[test]
fn test_set_and_lookup_with_provided_fingerprint() {
let mut store = SchemaStore::new();
let schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
let fp = schema.fingerprint(FingerprintAlgorithm::Rabin).unwrap();
let out_fp = store.set(fp, schema.clone()).unwrap();
assert_eq!(out_fp, fp);
assert_eq!(store.lookup(&fp).cloned(), Some(schema));
}
#[test]
fn test_set_duplicate_same_schema_ok() {
let mut store = SchemaStore::new();
let schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
let fp = schema.fingerprint(FingerprintAlgorithm::Rabin).unwrap();
let _ = store.set(fp, schema.clone()).unwrap();
let _ = store.set(fp, schema.clone()).unwrap();
assert_eq!(store.schemas.len(), 1);
}
#[test]
fn test_set_duplicate_different_schema_collision_error() {
let mut store = SchemaStore::new();
let schema1 = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
let schema2 = AvroSchema::new(serde_json::to_string(&record_schema()).unwrap());
// Use the same Fingerprint::Id to simulate a collision across different schemas
let fp = Fingerprint::Id(123);
let _ = store.set(fp, schema1).unwrap();
let err = store.set(fp, schema2).unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("Schema fingerprint collision"));
}
#[test]
fn test_canonical_form_generation_primitive() {
let schema = int_schema();
let canonical_form = AvroSchema::generate_canonical_form(&schema).unwrap();
assert_eq!(canonical_form, r#""int""#);
}
#[test]
fn test_canonical_form_generation_record() {
let schema = record_schema();
let expected_canonical_form = r#"{"name":"test.namespace.record1","type":"record","fields":[{"name":"field1","type":"int"},{"name":"field2","type":"string"}]}"#;
let canonical_form = AvroSchema::generate_canonical_form(&schema).unwrap();
assert_eq!(canonical_form, expected_canonical_form);
}
#[test]
fn test_fingerprint_calculation() {
let canonical_form = r#"{"fields":[{"name":"a","type":"long"},{"name":"b","type":"string"}],"name":"test","type":"record"}"#;
let expected_fingerprint = 10505236152925314060;
let fingerprint = compute_fingerprint_rabin(canonical_form);
assert_eq!(fingerprint, expected_fingerprint);
}
#[test]
fn test_register_and_lookup_complex_schema() {
let mut store = SchemaStore::new();
let schema = AvroSchema::new(serde_json::to_string(&record_schema()).unwrap());
let canonical_form = r#"{"name":"test.namespace.record1","type":"record","fields":[{"name":"field1","type":"int"},{"name":"field2","type":"string"}]}"#;
let expected_fingerprint = Fingerprint::Rabin(compute_fingerprint_rabin(canonical_form));
let fingerprint = store.register(schema.clone()).unwrap();
assert_eq!(fingerprint, expected_fingerprint);
let looked_up = store.lookup(&fingerprint).cloned();
assert_eq!(looked_up, Some(schema));
}
#[test]
fn test_fingerprints_returns_all_keys() {
let mut store = SchemaStore::new();
let fp_int = store
.register(AvroSchema::new(
serde_json::to_string(&int_schema()).unwrap(),
))
.unwrap();
let fp_record = store
.register(AvroSchema::new(
serde_json::to_string(&record_schema()).unwrap(),
))
.unwrap();
let fps = store.fingerprints();
assert_eq!(fps.len(), 2);
assert!(fps.contains(&fp_int));
assert!(fps.contains(&fp_record));
}
#[test]
fn test_canonical_form_strips_attributes() {
let schema_with_attrs = Schema::Complex(ComplexType::Record(Record {
name: "record_with_attrs",
namespace: None,
doc: Some(Cow::from("This doc should be stripped")),
aliases: vec!["alias1", "alias2"],
fields: vec![Field {
name: "f1",
doc: Some(Cow::from("field doc")),
r#type: Schema::Type(Type {
r#type: TypeName::Primitive(PrimitiveType::Bytes),
attributes: Attributes {
logical_type: None,
additional: HashMap::from([("precision", json!(4))]),
},
}),
default: None,
aliases: vec![],
}],
attributes: Attributes {
logical_type: None,
additional: HashMap::from([("custom_attr", json!("value"))]),
},
}));
let expected_canonical_form = r#"{"name":"record_with_attrs","type":"record","fields":[{"name":"f1","type":"bytes"}]}"#;
let canonical_form = AvroSchema::generate_canonical_form(&schema_with_attrs).unwrap();
assert_eq!(canonical_form, expected_canonical_form);
}
#[cfg(not(feature = "avro_custom_types"))]
#[test]
fn test_primitive_mappings() {
let cases = vec![
(DataType::Boolean, "\"boolean\""),
(DataType::Int8, "\"int\""),
(DataType::Int16, "\"int\""),
(DataType::Int32, "\"int\""),
(DataType::Int64, "\"long\""),
(DataType::UInt8, "\"int\""),
(DataType::UInt16, "\"int\""),
(DataType::UInt32, "\"long\""),
(DataType::UInt64, "\"long\""),
(DataType::Float16, "\"float\""),
(DataType::Float32, "\"float\""),
(DataType::Float64, "\"double\""),
(DataType::Utf8, "\"string\""),
(DataType::Binary, "\"bytes\""),
];
for (dt, avro_token) in cases {
let field = ArrowField::new("col", dt.clone(), false);
let arrow_schema = single_field_schema(field);
let avro = AvroSchema::try_from(&arrow_schema).unwrap();
assert_json_contains(&avro.json_string, avro_token);
}
}
#[cfg(feature = "avro_custom_types")]
#[test]
fn test_primitive_mappings() {
let cases = vec![
(DataType::Boolean, "\"boolean\""),
(DataType::Int8, "\"logicalType\":\"arrow.int8\""),
(DataType::Int16, "\"logicalType\":\"arrow.int16\""),
(DataType::Int32, "\"int\""),
(DataType::Int64, "\"long\""),
(DataType::UInt8, "\"logicalType\":\"arrow.uint8\""),
(DataType::UInt16, "\"logicalType\":\"arrow.uint16\""),
(DataType::UInt32, "\"logicalType\":\"arrow.uint32\""),
(DataType::UInt64, "\"logicalType\":\"arrow.uint64\""),
(DataType::Float16, "\"logicalType\":\"arrow.float16\""),
(DataType::Float32, "\"float\""),
(DataType::Float64, "\"double\""),
(DataType::Utf8, "\"string\""),
(DataType::Binary, "\"bytes\""),
];
for (dt, avro_token) in cases {
let field = ArrowField::new("col", dt.clone(), false);
let arrow_schema = single_field_schema(field);
let avro = AvroSchema::try_from(&arrow_schema).unwrap();
assert_json_contains(&avro.json_string, avro_token);
}
}
#[cfg(feature = "avro_custom_types")]
#[test]
fn test_custom_fixed_logical_types_preserve_namespace_metadata() {
let namespace = "com.example.types";
let mut md_u64 = HashMap::new();
md_u64.insert(AVRO_NAME_METADATA_KEY.to_string(), "U64Type".to_string());
md_u64.insert(
AVRO_NAMESPACE_METADATA_KEY.to_string(),
namespace.to_string(),
);
let mut md_f16 = HashMap::new();
md_f16.insert(AVRO_NAME_METADATA_KEY.to_string(), "F16Type".to_string());
md_f16.insert(
AVRO_NAMESPACE_METADATA_KEY.to_string(),
namespace.to_string(),
);
let mut md_iv_ym = HashMap::new();
md_iv_ym.insert(AVRO_NAME_METADATA_KEY.to_string(), "IvYmType".to_string());
md_iv_ym.insert(
AVRO_NAMESPACE_METADATA_KEY.to_string(),
namespace.to_string(),
);
let mut md_iv_dt = HashMap::new();
md_iv_dt.insert(AVRO_NAME_METADATA_KEY.to_string(), "IvDtType".to_string());
md_iv_dt.insert(
AVRO_NAMESPACE_METADATA_KEY.to_string(),
namespace.to_string(),
);
let arrow_schema = ArrowSchema::new(vec![
ArrowField::new("u64_col", DataType::UInt64, false).with_metadata(md_u64),
ArrowField::new("f16_col", DataType::Float16, false).with_metadata(md_f16),
ArrowField::new(
"iv_ym_col",
DataType::Interval(IntervalUnit::YearMonth),
false,
)
.with_metadata(md_iv_ym),
ArrowField::new(
"iv_dt_col",
DataType::Interval(IntervalUnit::DayTime),
false,
)
.with_metadata(md_iv_dt),
]);
let avro = AvroSchema::try_from(&arrow_schema).unwrap();
let root: Value = serde_json::from_str(&avro.json_string).unwrap();
let fields = root
.get("fields")
.and_then(|f| f.as_array())
.expect("record fields array");
let expected = [
("u64_col", "arrow.uint64"),
("f16_col", "arrow.float16"),
("iv_ym_col", "arrow.interval-year-month"),
("iv_dt_col", "arrow.interval-day-time"),
];
for (field_name, logical_type) in expected {
let field = fields
.iter()
.find(|f| f.get("name").and_then(Value::as_str) == Some(field_name))
.unwrap_or_else(|| panic!("missing field {field_name}"));
let ty = field
.get("type")
.and_then(Value::as_object)
.unwrap_or_else(|| panic!("field {field_name} type must be object"));
assert_eq!(ty.get("type").and_then(Value::as_str), Some("fixed"));
assert_eq!(
ty.get("logicalType").and_then(Value::as_str),
Some(logical_type)
);
assert_eq!(
ty.get("namespace").and_then(Value::as_str),
Some(namespace),
"field {field_name} must preserve avro.namespace metadata"
);
}
}
#[cfg(feature = "avro_custom_types")]
#[test]
fn test_custom_fixed_logical_types_omit_namespace_without_metadata() {
let mut md_u64 = HashMap::new();
md_u64.insert(AVRO_NAME_METADATA_KEY.to_string(), "U64Type".to_string());
let mut md_f16 = HashMap::new();
md_f16.insert(AVRO_NAME_METADATA_KEY.to_string(), "F16Type".to_string());
let mut md_iv_ym = HashMap::new();
md_iv_ym.insert(AVRO_NAME_METADATA_KEY.to_string(), "IvYmType".to_string());
let mut md_iv_dt = HashMap::new();
md_iv_dt.insert(AVRO_NAME_METADATA_KEY.to_string(), "IvDtType".to_string());
let arrow_schema = ArrowSchema::new(vec![
ArrowField::new("u64_col", DataType::UInt64, false).with_metadata(md_u64),
ArrowField::new("f16_col", DataType::Float16, false).with_metadata(md_f16),
ArrowField::new(
"iv_ym_col",
DataType::Interval(IntervalUnit::YearMonth),
false,
)
.with_metadata(md_iv_ym),
ArrowField::new(
"iv_dt_col",
DataType::Interval(IntervalUnit::DayTime),
false,
)
.with_metadata(md_iv_dt),
]);
let avro = AvroSchema::try_from(&arrow_schema).unwrap();
let root: Value = serde_json::from_str(&avro.json_string).unwrap();
let fields = root
.get("fields")
.and_then(|f| f.as_array())
.expect("record fields array");
for field_name in ["u64_col", "f16_col", "iv_ym_col", "iv_dt_col"] {
let field = fields
.iter()
.find(|f| f.get("name").and_then(Value::as_str) == Some(field_name))
.unwrap_or_else(|| panic!("missing field {field_name}"));
let ty = field
.get("type")
.and_then(Value::as_object)
.unwrap_or_else(|| panic!("field {field_name} type must be object"));
assert_eq!(ty.get("type").and_then(Value::as_str), Some("fixed"));
assert!(
!ty.contains_key("namespace"),
"field {field_name} should not include namespace when metadata lacks avro.namespace"
);
}
}
#[test]
fn test_temporal_mappings() {
let cases = vec![
(DataType::Date32, "\"logicalType\":\"date\""),
(
DataType::Time32(TimeUnit::Millisecond),
"\"logicalType\":\"time-millis\"",
),
(
DataType::Time64(TimeUnit::Microsecond),
"\"logicalType\":\"time-micros\"",
),
(
DataType::Timestamp(TimeUnit::Millisecond, None),
"\"logicalType\":\"local-timestamp-millis\"",
),
(
DataType::Timestamp(TimeUnit::Microsecond, Some("+00:00".into())),
"\"logicalType\":\"timestamp-micros\"",
),
];
for (dt, needle) in cases {
let field = ArrowField::new("ts", dt.clone(), true);
let arrow_schema = single_field_schema(field);
let avro = AvroSchema::try_from(&arrow_schema).unwrap();
assert_json_contains(&avro.json_string, needle);
}
}
#[test]
fn test_decimal_and_uuid() {
let decimal_field = ArrowField::new("amount", DataType::Decimal128(25, 2), false);
let dec_schema = single_field_schema(decimal_field);
let avro_dec = AvroSchema::try_from(&dec_schema).unwrap();
assert_json_contains(&avro_dec.json_string, "\"logicalType\":\"decimal\"");
assert_json_contains(&avro_dec.json_string, "\"precision\":25");
assert_json_contains(&avro_dec.json_string, "\"scale\":2");
let mut md = HashMap::new();
md.insert("logicalType".into(), "uuid".into());
let uuid_field =
ArrowField::new("id", DataType::FixedSizeBinary(16), false).with_metadata(md);
let uuid_schema = single_field_schema(uuid_field);
let avro_uuid = AvroSchema::try_from(&uuid_schema).unwrap();
assert_json_contains(&avro_uuid.json_string, "\"logicalType\":\"uuid\"");
}
#[cfg(not(feature = "avro_custom_types"))]
#[test]
fn test_interval_month_day_nano_duration_schema() {
let interval_field = ArrowField::new(
"span",
DataType::Interval(IntervalUnit::MonthDayNano),
false,
);
let s = single_field_schema(interval_field);
let avro = AvroSchema::try_from(&s).unwrap();
assert_json_contains(&avro.json_string, "\"logicalType\":\"duration\"");
assert_json_contains(&avro.json_string, "\"size\":12");
}
#[cfg(feature = "avro_custom_types")]
#[test]
fn test_interval_month_day_nano_custom_schema() {
let interval_field = ArrowField::new(
"span",
DataType::Interval(IntervalUnit::MonthDayNano),
false,
);
let s = single_field_schema(interval_field);
let avro = AvroSchema::try_from(&s).unwrap();
assert_json_contains(
&avro.json_string,
"\"logicalType\":\"arrow.interval-month-day-nano\"",
);
assert_json_contains(&avro.json_string, "\"size\":16");
}
#[cfg(feature = "avro_custom_types")]
#[test]
fn test_duration_custom_logical_type() {
let dur_field = ArrowField::new("latency", DataType::Duration(TimeUnit::Nanosecond), false);
let s2 = single_field_schema(dur_field);
let avro2 = AvroSchema::try_from(&s2).unwrap();
assert_json_contains(
&avro2.json_string,
"\"logicalType\":\"arrow.duration-nanos\"",
);
}
#[test]
fn test_complex_types() {
let list_dt = DataType::List(Arc::new(ArrowField::new("item", DataType::Int32, true)));
let list_schema = single_field_schema(ArrowField::new("numbers", list_dt, false));
let avro_list = AvroSchema::try_from(&list_schema).unwrap();
assert_json_contains(&avro_list.json_string, "\"type\":\"array\"");
assert_json_contains(&avro_list.json_string, "\"items\"");
let value_field = ArrowField::new("value", DataType::Boolean, true);
let entries_struct = ArrowField::new(
"entries",
DataType::Struct(Fields::from(vec![
ArrowField::new("key", DataType::Utf8, false),
value_field.clone(),
])),
false,
);
let map_dt = DataType::Map(Arc::new(entries_struct), false);
let map_schema = single_field_schema(ArrowField::new("props", map_dt, false));
let avro_map = AvroSchema::try_from(&map_schema).unwrap();
assert_json_contains(&avro_map.json_string, "\"type\":\"map\"");
assert_json_contains(&avro_map.json_string, "\"values\"");
let struct_dt = DataType::Struct(Fields::from(vec![
ArrowField::new("f1", DataType::Int64, false),
ArrowField::new("f2", DataType::Utf8, true),
]));
let struct_schema = single_field_schema(ArrowField::new("person", struct_dt, true));
let avro_struct = AvroSchema::try_from(&struct_schema).unwrap();
assert_json_contains(&avro_struct.json_string, "\"type\":\"record\"");
assert_json_contains(&avro_struct.json_string, "\"null\"");
}
#[test]
fn test_enum_dictionary() {
let mut md = HashMap::new();
md.insert(
AVRO_ENUM_SYMBOLS_METADATA_KEY.into(),
"[\"OPEN\",\"CLOSED\"]".into(),
);
let enum_dt = DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
let field = ArrowField::new("status", enum_dt, false).with_metadata(md);
let schema = single_field_schema(field);
let avro = AvroSchema::try_from(&schema).unwrap();
assert_json_contains(&avro.json_string, "\"type\":\"enum\"");
assert_json_contains(&avro.json_string, "\"symbols\":[\"OPEN\",\"CLOSED\"]");
}
#[test]
fn test_run_end_encoded() {
let ree_dt = DataType::RunEndEncoded(
Arc::new(ArrowField::new("run_ends", DataType::Int32, false)),
Arc::new(ArrowField::new("values", DataType::Utf8, false)),
);
let s = single_field_schema(ArrowField::new("text", ree_dt, false));
let avro = AvroSchema::try_from(&s).unwrap();
assert_json_contains(&avro.json_string, "\"string\"");
}
#[test]
fn test_dense_union() {
let uf: UnionFields = vec![
(2i8, Arc::new(ArrowField::new("a", DataType::Int32, false))),
(7i8, Arc::new(ArrowField::new("b", DataType::Utf8, true))),
]
.into_iter()
.collect();
let union_dt = DataType::Union(uf, UnionMode::Dense);
let s = single_field_schema(ArrowField::new("u", union_dt, false));
let avro =
AvroSchema::try_from(&s).expect("Arrow Union -> Avro union conversion should succeed");
let v: serde_json::Value = serde_json::from_str(&avro.json_string).unwrap();
let fields = v
.get("fields")
.and_then(|x| x.as_array())
.expect("fields array");
let u_field = fields
.iter()
.find(|f| f.get("name").and_then(|n| n.as_str()) == Some("u"))
.expect("field 'u'");
let union = u_field.get("type").expect("u.type");
let arr = union.as_array().expect("u.type must be Avro union array");
assert_eq!(arr.len(), 2, "expected two union branches");
let first = &arr[0];
let obj = first
.as_object()
.expect("first branch should be an object with metadata");
assert_eq!(obj.get("type").and_then(|t| t.as_str()), Some("int"));
assert_eq!(
obj.get("arrowUnionMode").and_then(|m| m.as_str()),
Some("dense")
);
let type_ids: Vec<i64> = obj
.get("arrowUnionTypeIds")
.and_then(|a| a.as_array())
.expect("arrowUnionTypeIds array")
.iter()
.map(|n| n.as_i64().expect("i64"))
.collect();
assert_eq!(type_ids, vec![2, 7], "type id ordering should be preserved");
assert_eq!(arr[1], Value::String("string".into()));
}
#[test]
fn round_trip_primitive() {
let arrow_schema = ArrowSchema::new(vec![ArrowField::new("f1", DataType::Int32, false)]);
let avro_schema = AvroSchema::try_from(&arrow_schema).unwrap();
let decoded = avro_schema.schema().unwrap();
assert!(matches!(decoded, Schema::Complex(_)));
}
#[test]
fn test_name_generator_sanitization_and_uniqueness() {
let f1 = ArrowField::new("weird-name", DataType::FixedSizeBinary(8), false);
let f2 = ArrowField::new("weird name", DataType::FixedSizeBinary(8), false);
let f3 = ArrowField::new("123bad", DataType::FixedSizeBinary(8), false);
let arrow_schema = ArrowSchema::new(vec![f1, f2, f3]);
let avro = AvroSchema::try_from(&arrow_schema).unwrap();
assert_json_contains(&avro.json_string, "\"name\":\"weird_name\"");
assert_json_contains(&avro.json_string, "\"name\":\"weird_name_1\"");
assert_json_contains(&avro.json_string, "\"name\":\"_123bad\"");
}
#[cfg(not(feature = "avro_custom_types"))]
#[test]
fn test_date64_logical_type_mapping() {
let field = ArrowField::new("d", DataType::Date64, true);
let schema = single_field_schema(field);
let avro = AvroSchema::try_from(&schema).unwrap();
assert_json_contains(
&avro.json_string,
"\"logicalType\":\"local-timestamp-millis\"",
);
}
#[cfg(feature = "avro_custom_types")]
#[test]
fn test_date64_logical_type_mapping_custom() {
let field = ArrowField::new("d", DataType::Date64, true);
let schema = single_field_schema(field);
let avro = AvroSchema::try_from(&schema).unwrap();
assert_json_contains(&avro.json_string, "\"logicalType\":\"arrow.date64\"");
}
#[cfg(feature = "avro_custom_types")]
#[test]
fn test_duration_list_extras_propagated() {
let child = ArrowField::new("lat", DataType::Duration(TimeUnit::Microsecond), false);
let list_dt = DataType::List(Arc::new(child));
let arrow_schema = single_field_schema(ArrowField::new("durations", list_dt, false));
let avro = AvroSchema::try_from(&arrow_schema).unwrap();
assert_json_contains(
&avro.json_string,
"\"logicalType\":\"arrow.duration-micros\"",
);
}
#[cfg(not(feature = "avro_custom_types"))]
#[test]
fn test_interval_yearmonth_extra() {
let field = ArrowField::new("iv", DataType::Interval(IntervalUnit::YearMonth), false);
let schema = single_field_schema(field);
let avro = AvroSchema::try_from(&schema).unwrap();
assert_json_contains(&avro.json_string, "\"arrowIntervalUnit\":\"yearmonth\"");
}
#[cfg(not(feature = "avro_custom_types"))]
#[test]
fn test_interval_daytime_extra() {
let field = ArrowField::new("iv_dt", DataType::Interval(IntervalUnit::DayTime), false);
let schema = single_field_schema(field);
let avro = AvroSchema::try_from(&schema).unwrap();
assert_json_contains(&avro.json_string, "\"arrowIntervalUnit\":\"daytime\"");
}
#[cfg(feature = "avro_custom_types")]
#[test]
fn test_interval_yearmonth_custom() {
let field = ArrowField::new("iv", DataType::Interval(IntervalUnit::YearMonth), false);
let schema = single_field_schema(field);
let avro = AvroSchema::try_from(&schema).unwrap();
assert_json_contains(
&avro.json_string,
"\"logicalType\":\"arrow.interval-year-month\"",
);
}
#[cfg(feature = "avro_custom_types")]
#[test]
fn test_interval_daytime_custom() {
let field = ArrowField::new("iv_dt", DataType::Interval(IntervalUnit::DayTime), false);
let schema = single_field_schema(field);
let avro = AvroSchema::try_from(&schema).unwrap();
assert_json_contains(
&avro.json_string,
"\"logicalType\":\"arrow.interval-day-time\"",
);
}
#[test]
fn test_fixed_size_list_extra() {
let child = ArrowField::new("item", DataType::Int32, false);
let dt = DataType::FixedSizeList(Arc::new(child), 3);
let schema = single_field_schema(ArrowField::new("triples", dt, false));
let avro = AvroSchema::try_from(&schema).unwrap();
assert_json_contains(&avro.json_string, "\"arrowFixedSize\":3");
}
#[cfg(feature = "avro_custom_types")]
#[test]
fn test_map_duration_value_extra() {
let val_field = ArrowField::new("value", DataType::Duration(TimeUnit::Second), true);
let entries_struct = ArrowField::new(
"entries",
DataType::Struct(Fields::from(vec![
ArrowField::new("key", DataType::Utf8, false),
val_field,
])),
false,
);
let map_dt = DataType::Map(Arc::new(entries_struct), false);
let schema = single_field_schema(ArrowField::new("metrics", map_dt, false));
let avro = AvroSchema::try_from(&schema).unwrap();
assert_json_contains(
&avro.json_string,
"\"logicalType\":\"arrow.duration-seconds\"",
);
}
#[test]
fn test_schema_with_non_string_defaults_decodes_successfully() {
let schema_json = r#"{
"type": "record",
"name": "R",
"fields": [
{"name": "a", "type": "int", "default": 0},
{"name": "b", "type": {"type": "array", "items": "long"}, "default": [1, 2, 3]},
{"name": "c", "type": {"type": "map", "values": "double"}, "default": {"x": 1.5, "y": 2.5}},
{"name": "inner", "type": {"type": "record", "name": "Inner", "fields": [
{"name": "flag", "type": "boolean", "default": true},
{"name": "name", "type": "string", "default": "hi"}
]}, "default": {"flag": false, "name": "d"}},
{"name": "u", "type": ["int", "null"], "default": 42}
]
}"#;
let schema: Schema = serde_json::from_str(schema_json).expect("schema should parse");
match &schema {
Schema::Complex(ComplexType::Record(_)) => {}
other => panic!("expected record schema, got: {:?}", other),
}
// Avro to Arrow conversion
let field = crate::codec::AvroField::try_from(&schema)
.expect("Avro->Arrow conversion should succeed");
let arrow_field = field.field();
// Build expected Arrow field
let expected_list_item = ArrowField::new(
arrow_schema::Field::LIST_FIELD_DEFAULT_NAME,
DataType::Int64,
false,
);
let expected_b = ArrowField::new("b", DataType::List(Arc::new(expected_list_item)), false);
let expected_map_value = ArrowField::new("value", DataType::Float64, false);
let expected_entries = ArrowField::new(
"entries",
DataType::Struct(Fields::from(vec![
ArrowField::new("key", DataType::Utf8, false),
expected_map_value,
])),
false,
);
let expected_c =
ArrowField::new("c", DataType::Map(Arc::new(expected_entries), false), false);
let mut inner_md = std::collections::HashMap::new();
inner_md.insert(AVRO_NAME_METADATA_KEY.to_string(), "Inner".to_string());
let expected_inner = ArrowField::new(
"inner",
DataType::Struct(Fields::from(vec![
ArrowField::new("flag", DataType::Boolean, false),
ArrowField::new("name", DataType::Utf8, false),
])),
false,
)
.with_metadata(inner_md);
let mut root_md = std::collections::HashMap::new();
root_md.insert(AVRO_NAME_METADATA_KEY.to_string(), "R".to_string());
let expected = ArrowField::new(
"R",
DataType::Struct(Fields::from(vec![
ArrowField::new("a", DataType::Int32, false),
expected_b,
expected_c,
expected_inner,
ArrowField::new("u", DataType::Int32, true),
])),
false,
)
.with_metadata(root_md);
assert_eq!(arrow_field, expected);
}
#[test]
fn default_order_is_consistent() {
let arrow_schema = ArrowSchema::new(vec![ArrowField::new("s", DataType::Utf8, true)]);
let a = AvroSchema::try_from(&arrow_schema).unwrap().json_string;
let b = AvroSchema::from_arrow_with_options(&arrow_schema, None);
assert_eq!(a, b.unwrap().json_string);
}
#[test]
fn test_union_branch_missing_name_errors() {
for t in ["record", "enum", "fixed"] {
let branch = json!({ "type": t });
let err = union_branch_signature(&branch).unwrap_err().to_string();
assert!(
err.contains(&format!("Union branch '{t}' missing required 'name'")),
"expected missing-name error for {t}, got: {err}"
);
}
}
#[test]
fn test_union_branch_named_type_signature_includes_name() {
let rec = json!({ "type": "record", "name": "Foo" });
assert_eq!(union_branch_signature(&rec).unwrap(), "N:record:Foo");
let en = json!({ "type": "enum", "name": "Color", "symbols": ["R", "G", "B"] });
assert_eq!(union_branch_signature(&en).unwrap(), "N:enum:Color");
let fx = json!({ "type": "fixed", "name": "Bytes16", "size": 16 });
assert_eq!(union_branch_signature(&fx).unwrap(), "N:fixed:Bytes16");
}
#[test]
fn test_record_field_alias_resolution_without_default() {
let writer_json = r#"{
"type":"record",
"name":"R",
"fields":[{"name":"old","type":"int"}]
}"#;
let reader_json = r#"{
"type":"record",
"name":"R",
"fields":[{"name":"new","aliases":["old"],"type":"int"}]
}"#;
let writer: Schema = serde_json::from_str(writer_json).unwrap();
let reader: Schema = serde_json::from_str(reader_json).unwrap();
let resolved = AvroFieldBuilder::new(&writer)
.with_reader_schema(&reader)
.with_utf8view(false)
.with_strict_mode(false)
.build()
.unwrap();
let expected = ArrowField::new(
"R",
DataType::Struct(Fields::from(vec![ArrowField::new(
"new",
DataType::Int32,
false,
)])),
false,
)
.with_metadata(HashMap::from_iter([(
"avro.name".to_owned(),
"R".to_owned(),
)]));
assert_eq!(resolved.field(), expected);
}
#[test]
fn test_record_field_alias_ambiguous_in_strict_mode_errors() {
let writer_json = r#"{
"type":"record",
"name":"R",
"fields":[
{"name":"a","type":"int","aliases":["old"]},
{"name":"b","type":"int","aliases":["old"]}
]
}"#;
let reader_json = r#"{
"type":"record",
"name":"R",
"fields":[{"name":"target","type":"int","aliases":["old"]}]
}"#;
let writer: Schema = serde_json::from_str(writer_json).unwrap();
let reader: Schema = serde_json::from_str(reader_json).unwrap();
let err = AvroFieldBuilder::new(&writer)
.with_reader_schema(&reader)
.with_utf8view(false)
.with_strict_mode(true)
.build()
.unwrap_err()
.to_string();
assert!(
err.contains("Ambiguous alias 'old'"),
"expected ambiguous-alias error, got: {err}"
);
}
#[test]
fn test_pragmatic_writer_field_alias_mapping_non_strict() {
let writer_json = r#"{
"type":"record",
"name":"R",
"fields":[{"name":"before","type":"int","aliases":["now"]}]
}"#;
let reader_json = r#"{
"type":"record",
"name":"R",
"fields":[{"name":"now","type":"int"}]
}"#;
let writer: Schema = serde_json::from_str(writer_json).unwrap();
let reader: Schema = serde_json::from_str(reader_json).unwrap();
let resolved = AvroFieldBuilder::new(&writer)
.with_reader_schema(&reader)
.with_utf8view(false)
.with_strict_mode(false)
.build()
.unwrap();
let expected = ArrowField::new(
"R",
DataType::Struct(Fields::from(vec![ArrowField::new(
"now",
DataType::Int32,
false,
)])),
false,
)
.with_metadata(HashMap::from_iter([(
"avro.name".to_owned(),
"R".to_owned(),
)]));
assert_eq!(resolved.field(), expected);
}
#[test]
fn test_missing_reader_field_null_first_no_default_is_ok() {
let writer_json = r#"{
"type":"record",
"name":"R",
"fields":[{"name":"a","type":"int"}]
}"#;
let reader_json = r#"{
"type":"record",
"name":"R",
"fields":[
{"name":"a","type":"int"},
{"name":"b","type":["null","int"]}
]
}"#;
let writer: Schema = serde_json::from_str(writer_json).unwrap();
let reader: Schema = serde_json::from_str(reader_json).unwrap();
let resolved = AvroFieldBuilder::new(&writer)
.with_reader_schema(&reader)
.with_utf8view(false)
.with_strict_mode(false)
.build()
.unwrap();
let expected = ArrowField::new(
"R",
DataType::Struct(Fields::from(vec![
ArrowField::new("a", DataType::Int32, false),
ArrowField::new("b", DataType::Int32, true).with_metadata(HashMap::from([(
AVRO_FIELD_DEFAULT_METADATA_KEY.to_string(),
"null".to_string(),
)])),
])),
false,
)
.with_metadata(HashMap::from_iter([(
"avro.name".to_owned(),
"R".to_owned(),
)]));
assert_eq!(resolved.field(), expected);
}
#[test]
fn test_missing_reader_field_null_second_without_default_errors() {
let writer_json = r#"{
"type":"record",
"name":"R",
"fields":[{"name":"a","type":"int"}]
}"#;
let reader_json = r#"{
"type":"record",
"name":"R",
"fields":[
{"name":"a","type":"int"},
{"name":"b","type":["int","null"]}
]
}"#;
let writer: Schema = serde_json::from_str(writer_json).unwrap();
let reader: Schema = serde_json::from_str(reader_json).unwrap();
let err = AvroFieldBuilder::new(&writer)
.with_reader_schema(&reader)
.with_utf8view(false)
.with_strict_mode(false)
.build()
.unwrap_err()
.to_string();
assert!(
err.contains("must have a default value"),
"expected missing-default error, got: {err}"
);
}
#[test]
fn test_from_arrow_with_options_respects_schema_metadata_when_not_stripping() {
let field = ArrowField::new("x", DataType::Int32, true);
let injected_json =
r#"{"type":"record","name":"Injected","fields":[{"name":"ignored","type":"int"}]}"#
.to_string();
let mut md = HashMap::new();
md.insert(SCHEMA_METADATA_KEY.to_string(), injected_json.clone());
md.insert("custom".to_string(), "123".to_string());
let arrow_schema = ArrowSchema::new_with_metadata(vec![field], md);
let opts = AvroSchemaOptions {
null_order: Some(Nullability::NullSecond),
strip_metadata: false,
};
let out = AvroSchema::from_arrow_with_options(&arrow_schema, Some(opts)).unwrap();
assert_eq!(
out.json_string, injected_json,
"When strip_metadata=false and avro.schema is present, return the embedded JSON verbatim"
);
let v: Value = serde_json::from_str(&out.json_string).unwrap();
assert_eq!(v.get("type").and_then(|t| t.as_str()), Some("record"));
assert_eq!(v.get("name").and_then(|n| n.as_str()), Some("Injected"));
}
#[test]
fn test_from_arrow_with_options_ignores_schema_metadata_when_stripping_and_keeps_passthrough() {
let field = ArrowField::new("x", DataType::Int32, true);
let injected_json =
r#"{"type":"record","name":"Injected","fields":[{"name":"ignored","type":"int"}]}"#
.to_string();
let mut md = HashMap::new();
md.insert(SCHEMA_METADATA_KEY.to_string(), injected_json);
md.insert("custom_meta".to_string(), "7".to_string());
let arrow_schema = ArrowSchema::new_with_metadata(vec![field], md);
let opts = AvroSchemaOptions {
null_order: Some(Nullability::NullFirst),
strip_metadata: true,
};
let out = AvroSchema::from_arrow_with_options(&arrow_schema, Some(opts)).unwrap();
assert_json_contains(&out.json_string, "\"type\":\"record\"");
assert_json_contains(&out.json_string, "\"name\":\"topLevelRecord\"");
assert_json_contains(&out.json_string, "\"custom_meta\":7");
}
#[test]
fn test_from_arrow_with_options_null_first_for_nullable_primitive() {
let field = ArrowField::new("s", DataType::Utf8, true);
let arrow_schema = single_field_schema(field);
let opts = AvroSchemaOptions {
null_order: Some(Nullability::NullFirst),
strip_metadata: true,
};
let out = AvroSchema::from_arrow_with_options(&arrow_schema, Some(opts)).unwrap();
let v: Value = serde_json::from_str(&out.json_string).unwrap();
let arr = v["fields"][0]["type"]
.as_array()
.expect("nullable primitive should be Avro union array");
assert_eq!(arr[0], Value::String("null".into()));
assert_eq!(arr[1], Value::String("string".into()));
}
#[test]
fn test_from_arrow_with_options_null_second_for_nullable_primitive() {
let field = ArrowField::new("s", DataType::Utf8, true);
let arrow_schema = single_field_schema(field);
let opts = AvroSchemaOptions {
null_order: Some(Nullability::NullSecond),
strip_metadata: true,
};
let out = AvroSchema::from_arrow_with_options(&arrow_schema, Some(opts)).unwrap();
let v: Value = serde_json::from_str(&out.json_string).unwrap();
let arr = v["fields"][0]["type"]
.as_array()
.expect("nullable primitive should be Avro union array");
assert_eq!(arr[0], Value::String("string".into()));
assert_eq!(arr[1], Value::String("null".into()));
}
#[test]
fn test_from_arrow_with_options_union_extras_respected_by_strip_metadata() {
let uf: UnionFields = vec![
(2i8, Arc::new(ArrowField::new("a", DataType::Int32, false))),
(7i8, Arc::new(ArrowField::new("b", DataType::Utf8, false))),
]
.into_iter()
.collect();
let union_dt = DataType::Union(uf, UnionMode::Dense);
let arrow_schema = single_field_schema(ArrowField::new("u", union_dt, true));
let with_extras = AvroSchema::from_arrow_with_options(
&arrow_schema,
Some(AvroSchemaOptions {
null_order: Some(Nullability::NullFirst),
strip_metadata: false,
}),
)
.unwrap();
let v_with: Value = serde_json::from_str(&with_extras.json_string).unwrap();
let union_arr = v_with["fields"][0]["type"].as_array().expect("union array");
let first_obj = union_arr
.iter()
.find(|b| b.is_object())
.expect("expected an object branch with extras");
let obj = first_obj.as_object().unwrap();
assert_eq!(obj.get("type").and_then(|t| t.as_str()), Some("int"));
assert_eq!(
obj.get("arrowUnionMode").and_then(|m| m.as_str()),
Some("dense")
);
let type_ids: Vec<i64> = obj["arrowUnionTypeIds"]
.as_array()
.expect("arrowUnionTypeIds array")
.iter()
.map(|n| n.as_i64().expect("i64"))
.collect();
assert_eq!(type_ids, vec![2, 7]);
let stripped = AvroSchema::from_arrow_with_options(
&arrow_schema,
Some(AvroSchemaOptions {
null_order: Some(Nullability::NullFirst),
strip_metadata: true,
}),
)
.unwrap();
let v_stripped: Value = serde_json::from_str(&stripped.json_string).unwrap();
let union_arr2 = v_stripped["fields"][0]["type"]
.as_array()
.expect("union array");
assert!(
!union_arr2.iter().any(|b| b
.as_object()
.is_some_and(|m| m.contains_key("arrowUnionMode"))),
"extras must be removed when strip_metadata=true"
);
assert_eq!(union_arr2[0], Value::String("null".into()));
assert_eq!(union_arr2[1], Value::String("int".into()));
assert_eq!(union_arr2[2], Value::String("string".into()));
}
#[test]
fn test_project_empty_projection() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "a", "type": "int"},
{"name": "b", "type": "string"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let projected = schema.project(&[]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert!(
fields.is_empty(),
"Empty projection should yield empty fields"
);
}
#[test]
fn test_project_single_field() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "a", "type": "int"},
{"name": "b", "type": "string"},
{"name": "c", "type": "long"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let projected = schema.project(&[1]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert_eq!(fields.len(), 1);
assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("b"));
}
#[test]
fn test_project_multiple_fields() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "a", "type": "int"},
{"name": "b", "type": "string"},
{"name": "c", "type": "long"},
{"name": "d", "type": "boolean"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let projected = schema.project(&[0, 2, 3]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert_eq!(fields.len(), 3);
assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("a"));
assert_eq!(fields[1].get("name").and_then(|n| n.as_str()), Some("c"));
assert_eq!(fields[2].get("name").and_then(|n| n.as_str()), Some("d"));
}
#[test]
fn test_project_all_fields() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "a", "type": "int"},
{"name": "b", "type": "string"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let projected = schema.project(&[0, 1]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert_eq!(fields.len(), 2);
assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("a"));
assert_eq!(fields[1].get("name").and_then(|n| n.as_str()), Some("b"));
}
#[test]
fn test_project_reorder_fields() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "a", "type": "int"},
{"name": "b", "type": "string"},
{"name": "c", "type": "long"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
// Project in reverse order
let projected = schema.project(&[2, 0, 1]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert_eq!(fields.len(), 3);
assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("c"));
assert_eq!(fields[1].get("name").and_then(|n| n.as_str()), Some("a"));
assert_eq!(fields[2].get("name").and_then(|n| n.as_str()), Some("b"));
}
#[test]
fn test_project_preserves_record_metadata() {
let schema_json = r#"{
"type": "record",
"name": "MyRecord",
"namespace": "com.example",
"doc": "A test record",
"aliases": ["OldRecord"],
"fields": [
{"name": "a", "type": "int"},
{"name": "b", "type": "string"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let projected = schema.project(&[0]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
assert_eq!(v.get("name").and_then(|n| n.as_str()), Some("MyRecord"));
assert_eq!(
v.get("namespace").and_then(|n| n.as_str()),
Some("com.example")
);
assert_eq!(v.get("doc").and_then(|n| n.as_str()), Some("A test record"));
assert!(v.get("aliases").is_some());
}
#[test]
fn test_project_preserves_field_metadata() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "a", "type": "int", "doc": "Field A", "default": 0},
{"name": "b", "type": "string"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let projected = schema.project(&[0]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert_eq!(
fields[0].get("doc").and_then(|d| d.as_str()),
Some("Field A")
);
assert_eq!(fields[0].get("default").and_then(|d| d.as_i64()), Some(0));
}
#[test]
fn test_project_with_nested_record() {
let schema_json = r#"{
"type": "record",
"name": "Outer",
"fields": [
{"name": "id", "type": "int"},
{"name": "inner", "type": {
"type": "record",
"name": "Inner",
"fields": [
{"name": "x", "type": "int"},
{"name": "y", "type": "string"}
]
}},
{"name": "value", "type": "double"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let projected = schema.project(&[1]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert_eq!(fields.len(), 1);
assert_eq!(
fields[0].get("name").and_then(|n| n.as_str()),
Some("inner")
);
// Verify nested record structure is preserved
let inner_type = fields[0].get("type").unwrap();
assert_eq!(
inner_type.get("type").and_then(|t| t.as_str()),
Some("record")
);
assert_eq!(
inner_type.get("name").and_then(|n| n.as_str()),
Some("Inner")
);
}
#[test]
fn test_project_with_complex_field_types() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "arr", "type": {"type": "array", "items": "int"}},
{"name": "map", "type": {"type": "map", "values": "string"}},
{"name": "union", "type": ["null", "int"]}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let projected = schema.project(&[0, 2]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert_eq!(fields.len(), 2);
// Verify array type is preserved
let arr_type = fields[0].get("type").unwrap();
assert_eq!(arr_type.get("type").and_then(|t| t.as_str()), Some("array"));
// Verify union type is preserved
let union_type = fields[1].get("type").unwrap();
assert!(union_type.is_array());
}
#[test]
fn test_project_error_invalid_json() {
let schema = AvroSchema::new("not valid json".to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("Invalid Avro schema JSON"),
"Expected parse error, got: {msg}"
);
}
#[test]
fn test_project_error_not_object() {
// Primitive type schema (not a JSON object)
let schema = AvroSchema::new(r#""string""#.to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("must be a JSON object"),
"Expected object error, got: {msg}"
);
}
#[test]
fn test_project_error_array_schema() {
// Array (list) is a valid JSON but not a record
let schema = AvroSchema::new(r#"["null", "int"]"#.to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("must be a JSON object"),
"Expected object error for array schema, got: {msg}"
);
}
#[test]
fn test_project_error_type_not_record() {
let schema_json = r#"{
"type": "enum",
"name": "Color",
"symbols": ["RED", "GREEN", "BLUE"]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("must be an Avro record") && msg.contains("'enum'"),
"Expected type mismatch error, got: {msg}"
);
}
#[test]
fn test_project_error_type_array() {
let schema_json = r#"{
"type": "array",
"items": "int"
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("must be an Avro record") && msg.contains("'array'"),
"Expected type mismatch error for array type, got: {msg}"
);
}
#[test]
fn test_project_error_type_fixed() {
let schema_json = r#"{
"type": "fixed",
"name": "MD5",
"size": 16
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("must be an Avro record") && msg.contains("'fixed'"),
"Expected type mismatch error for fixed type, got: {msg}"
);
}
#[test]
fn test_project_error_type_map() {
let schema_json = r#"{
"type": "map",
"values": "string"
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("must be an Avro record") && msg.contains("'map'"),
"Expected type mismatch error for map type, got: {msg}"
);
}
#[test]
fn test_project_error_missing_type_field() {
let schema_json = r#"{
"name": "Test",
"fields": [{"name": "a", "type": "int"}]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("missing required 'type' field"),
"Expected missing type error, got: {msg}"
);
}
#[test]
fn test_project_error_missing_fields() {
let schema_json = r#"{
"type": "record",
"name": "Test"
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("missing required 'fields'"),
"Expected missing fields error, got: {msg}"
);
}
#[test]
fn test_project_error_fields_not_array() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": "not an array"
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("'fields' must be an array"),
"Expected fields array error, got: {msg}"
);
}
#[test]
fn test_project_error_index_out_of_bounds() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "a", "type": "int"},
{"name": "b", "type": "string"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[5]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("out of bounds") && msg.contains("5") && msg.contains("2"),
"Expected out of bounds error, got: {msg}"
);
}
#[test]
fn test_project_error_index_out_of_bounds_edge() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "a", "type": "int"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
// Index 1 is just out of bounds for a 1-element array
let err = schema.project(&[1]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("out of bounds") && msg.contains("1"),
"Expected out of bounds error for edge case, got: {msg}"
);
}
#[test]
fn test_project_error_duplicate_index() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "a", "type": "int"},
{"name": "b", "type": "string"},
{"name": "c", "type": "long"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[0, 1, 0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("Duplicate projection index") && msg.contains("0"),
"Expected duplicate index error, got: {msg}"
);
}
#[test]
fn test_project_error_duplicate_index_consecutive() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "a", "type": "int"},
{"name": "b", "type": "string"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[1, 1]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("Duplicate projection index") && msg.contains("1"),
"Expected duplicate index error for consecutive duplicates, got: {msg}"
);
}
#[test]
fn test_project_with_empty_fields() {
let schema_json = r#"{
"type": "record",
"name": "EmptyRecord",
"fields": []
}"#;
let schema = AvroSchema::new(schema_json.to_string());
// Projecting empty from empty should succeed
let projected = schema.project(&[]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert!(fields.is_empty());
}
#[test]
fn test_project_empty_fields_index_out_of_bounds() {
let schema_json = r#"{
"type": "record",
"name": "EmptyRecord",
"fields": []
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let err = schema.project(&[0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("out of bounds") && msg.contains("0 fields"),
"Expected out of bounds error for empty record, got: {msg}"
);
}
#[test]
fn test_project_result_is_valid_avro_schema() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"namespace": "com.example",
"fields": [
{"name": "id", "type": "long"},
{"name": "name", "type": "string"},
{"name": "active", "type": "boolean"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
let projected = schema.project(&[0, 2]).unwrap();
// Verify the projected schema can be parsed as a valid Avro schema
let parsed = projected.schema();
assert!(parsed.is_ok(), "Projected schema should be valid Avro");
match parsed.unwrap() {
Schema::Complex(ComplexType::Record(r)) => {
assert_eq!(r.name, "Test");
assert_eq!(r.namespace, Some("com.example"));
assert_eq!(r.fields.len(), 2);
assert_eq!(r.fields[0].name, "id");
assert_eq!(r.fields[1].name, "active");
}
_ => panic!("Expected Record schema"),
}
}
#[test]
fn test_project_non_contiguous_indices() {
let schema_json = r#"{
"type": "record",
"name": "Test",
"fields": [
{"name": "f0", "type": "int"},
{"name": "f1", "type": "int"},
{"name": "f2", "type": "int"},
{"name": "f3", "type": "int"},
{"name": "f4", "type": "int"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
// Select every other field
let projected = schema.project(&[0, 2, 4]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert_eq!(fields.len(), 3);
assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("f0"));
assert_eq!(fields[1].get("name").and_then(|n| n.as_str()), Some("f2"));
assert_eq!(fields[2].get("name").and_then(|n| n.as_str()), Some("f4"));
}
#[test]
fn test_project_single_field_from_many() {
let schema_json = r#"{
"type": "record",
"name": "BigRecord",
"fields": [
{"name": "f0", "type": "int"},
{"name": "f1", "type": "int"},
{"name": "f2", "type": "int"},
{"name": "f3", "type": "int"},
{"name": "f4", "type": "int"},
{"name": "f5", "type": "int"},
{"name": "f6", "type": "int"},
{"name": "f7", "type": "int"},
{"name": "f8", "type": "int"},
{"name": "f9", "type": "int"}
]
}"#;
let schema = AvroSchema::new(schema_json.to_string());
// Select only the last field
let projected = schema.project(&[9]).unwrap();
let v: Value = serde_json::from_str(&projected.json_string).unwrap();
let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
assert_eq!(fields.len(), 1);
assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("f9"));
}
}