You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "jecsand838 (via GitHub)" <gi...@apache.org> on 2025/08/04 23:16:41 UTC
[PR] Add arrow-avro `SchemaStore` and fingerprinting [arrow-rs]
jecsand838 opened a new pull request, #8039:
URL: https://github.com/apache/arrow-rs/pull/8039
# Which issue does this PR close?
- Part of https://github.com/apache/arrow-rs/issues/4886
- Pre-work for https://github.com/apache/arrow-rs/pull/8006
# Rationale for this change
Apache Avro’s [single object encoding](https://avro.apache.org/docs/1.11.1/specification/#single-object-encoding) prefixes every record with the marker `0xC3 0x01` followed by a `Rabin` [schema fingerprint ](https://avro.apache.org/docs/1.11.1/specification/#schema-fingerprints) so that readers can identify the correct writer schema without carrying the full definition in each message.
While the current `arrow‑avro` implementation can read container files, it cannot ingest these framed messages or handle streams where the writer schema changes over time.
The Avro specification recommends computing a 64‑bit CRC‑64‑AVRO (Rabin) hashed fingerprint of the [parsed canonical form of a schema](https://avro.apache.org/docs/1.11.1/specification/#parsing-canonical-form-for-schemas) to look up the `Schema` from a local schema store or registry.
This PR introduces **`SchemaStore`** and **fingerprinting** to enable:
* **Zero‑copy schema identification** for decoding streaming Avro messages published in single‑object format (i.e. Kafka, Pulsar, etc) into Arrow.
* **Dynamic schema evolution** by laying the foundation to resolve writer reader schema differences on the fly.
**NOTE:** Integration with `Decoder` and `Reader` coming in next PR.
# What changes are included in this PR?
| Area | Highlights |
| ------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **`schema.rs`** | *New* `Fingerprint`, `SchemaStore`, and `SINGLE_OBJECT_MAGIC`; canonical‑form generator; Rabin fingerprint calculator; `compare_schemas` helper. |
| **`lib.rs`** | `mod schema` is now `pub` |
| **Unit tests** | New tests covering fingerprint generation, store registration/lookup, unknown‑fingerprint errors, and interaction with UTF8‑view decoding. |
| **Docs & Examples** | Extensive inline docs with examples on all new public methods / structs. |
# Are these changes tested?
Yes. New tests cover:
1. **Fingerprinting** against the canonical examples from the Avro spec
2. **`SchemaStore` behavior** deduplication, duplicate registration, and lookup.
# Are there any user-facing changes?
N/A
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add arrow-avro `SchemaStore` and fingerprinting [arrow-rs]
Posted by "scovich (via GitHub)" <gi...@apache.org>.
scovich commented on code in PR #8039:
URL: https://github.com/apache/arrow-rs/pull/8039#discussion_r2255018862
##########
arrow-avro/src/schema.rs:
##########
@@ -260,13 +276,388 @@ pub struct Fixed<'a> {
pub attributes: Attributes<'a>,
}
+/// Supported fingerprint algorithms for Avro schema identification.
+/// Currently only `Rabin` is supported, `SHA256` and `MD5` support will come in a future update
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Default)]
+pub enum FingerprintAlgorithm {
+ /// 64‑bit CRC‑64‑AVRO Rabin fingerprint.
+ #[default]
+ Rabin,
+}
+
+/// 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.
+/// Currently only `Rabin` is supported
+///
+/// <https://avro.apache.org/docs/1.11.1/specification/#schema-fingerprints>
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub enum Fingerprint {
+ /// A 64-bit Rabin fingerprint.
+ Rabin(u64),
+}
+
+/// 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,
+ }
+ }
+}
+
+/// Generates a fingerprint for the given `Schema` using the specified `FingerprintAlgorithm`.
+pub(crate) fn generate_fingerprint(
+ schema: &Schema,
+ hash_type: FingerprintAlgorithm,
+) -> Result<Fingerprint, ArrowError> {
+ let canonical = 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)))
+ }
+ }
+}
+
+/// Generates the 64-bit Rabin fingerprint for the given `Schema`.
+///
+/// The fingerprint is computed from the canonical form of the schema.
+/// This is also known as `CRC-64-AVRO`.
+///
+/// # Returns
+/// A `Fingerprint::Rabin` variant containing the 64-bit fingerprint.
+pub fn generate_fingerprint_rabin(schema: &Schema) -> Result<Fingerprint, ArrowError> {
+ generate_fingerprint(schema, FingerprintAlgorithm::Rabin)
+}
+
+/// 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 fn generate_canonical_form(schema: &Schema) -> Result<String, ArrowError> {
+ build_canonical(schema, None)
+}
+
+/// 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.
+///
+/// The lifetime parameter `'a` corresponds to the lifetime of the string slices
+/// contained within the stored [`Schema`] objects. This means the `SchemaStore`
+/// cannot outlive the data referenced by the schemas it contains.
+///
+/// # Examples
+///
+/// ```no_run
+/// // Create a new store with the default Rabin fingerprinting.
+/// use arrow_avro::schema::{PrimitiveType, Schema, SchemaStore, TypeName};
+///
+/// let mut store = SchemaStore::new();
+/// let schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::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<'a> {
+ /// The hashing algorithm used for generating fingerprints.
+ fingerprint_algorithm: FingerprintAlgorithm,
+ /// A map from a schema's fingerprint to the schema itself.
+ schemas: HashMap<Fingerprint, Schema<'a>>,
+}
+
+impl<'a> TryFrom<&'a [Schema<'a>]> for SchemaStore<'a> {
+ type Error = ArrowError;
+
+ /// Creates a `SchemaStore` from a slice of schemas.
+ /// Each schema in the slice is registered with the new store.
+ fn try_from(schemas: &'a [Schema<'a>]) -> Result<Self, Self::Error> {
+ let mut store = SchemaStore::new();
+ for schema in schemas {
+ store.register(schema.clone())?;
+ }
+ Ok(store)
+ }
+}
+
+impl<'a> SchemaStore<'a> {
+ /// Creates an empty `SchemaStore` using the default fingerprinting algorithm (64-bit Rabin).
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// 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.
+ ///
+ /// # Arguments
+ ///
+ /// * `schema` - The schema to register.
+ ///
+ /// # Returns
+ ///
+ /// A `Result` containing the `Fingerprint` of the schema if successful,
+ /// or an `ArrowError` on failure.
+ pub fn register(&mut self, schema: Schema<'a>) -> Result<Fingerprint, ArrowError> {
+ let fingerprint = generate_fingerprint(&schema, self.fingerprint_algorithm)?;
+ 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)
+ }
+
+ /// 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 `Schema` if found, otherwise `None`.
+ pub fn lookup(&self, fingerprint: &Fingerprint) -> Option<&Schema<'a>> {
+ 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`.
+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| {
+ let field_type =
+ build_canonical(&f.r#type, child_ns.as_deref().or(enclosing_ns))?;
+ Ok(format!(
+ r#"{{"name":{},"type":{}}}"#,
Review Comment:
Makes sense, yup!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add arrow-avro `SchemaStore` and fingerprinting [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #8039:
URL: https://github.com/apache/arrow-rs/pull/8039
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add arrow-avro `SchemaStore` and fingerprinting [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #8039:
URL: https://github.com/apache/arrow-rs/pull/8039#issuecomment-3156330004
Let's keep the code flowing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add arrow-avro `SchemaStore` and fingerprinting [arrow-rs]
Posted by "jecsand838 (via GitHub)" <gi...@apache.org>.
jecsand838 commented on PR #8039:
URL: https://github.com/apache/arrow-rs/pull/8039#issuecomment-3152768433
@scovich @alamb Here's that first PR for the `SchemaStore` work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add arrow-avro `SchemaStore` and fingerprinting [arrow-rs]
Posted by "jecsand838 (via GitHub)" <gi...@apache.org>.
jecsand838 commented on code in PR #8039:
URL: https://github.com/apache/arrow-rs/pull/8039#discussion_r2254884015
##########
arrow-avro/src/schema.rs:
##########
@@ -260,13 +276,388 @@ pub struct Fixed<'a> {
pub attributes: Attributes<'a>,
}
+/// Supported fingerprint algorithms for Avro schema identification.
+/// Currently only `Rabin` is supported, `SHA256` and `MD5` support will come in a future update
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Default)]
+pub enum FingerprintAlgorithm {
+ /// 64‑bit CRC‑64‑AVRO Rabin fingerprint.
+ #[default]
+ Rabin,
+}
+
+/// 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.
+/// Currently only `Rabin` is supported
+///
+/// <https://avro.apache.org/docs/1.11.1/specification/#schema-fingerprints>
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub enum Fingerprint {
+ /// A 64-bit Rabin fingerprint.
+ Rabin(u64),
+}
+
+/// 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,
+ }
+ }
+}
+
+/// Generates a fingerprint for the given `Schema` using the specified `FingerprintAlgorithm`.
+pub(crate) fn generate_fingerprint(
+ schema: &Schema,
+ hash_type: FingerprintAlgorithm,
+) -> Result<Fingerprint, ArrowError> {
+ let canonical = 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)))
+ }
+ }
+}
+
+/// Generates the 64-bit Rabin fingerprint for the given `Schema`.
+///
+/// The fingerprint is computed from the canonical form of the schema.
+/// This is also known as `CRC-64-AVRO`.
+///
+/// # Returns
+/// A `Fingerprint::Rabin` variant containing the 64-bit fingerprint.
+pub fn generate_fingerprint_rabin(schema: &Schema) -> Result<Fingerprint, ArrowError> {
+ generate_fingerprint(schema, FingerprintAlgorithm::Rabin)
+}
+
+/// 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 fn generate_canonical_form(schema: &Schema) -> Result<String, ArrowError> {
+ build_canonical(schema, None)
+}
+
+/// 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.
+///
+/// The lifetime parameter `'a` corresponds to the lifetime of the string slices
+/// contained within the stored [`Schema`] objects. This means the `SchemaStore`
+/// cannot outlive the data referenced by the schemas it contains.
+///
+/// # Examples
+///
+/// ```no_run
+/// // Create a new store with the default Rabin fingerprinting.
+/// use arrow_avro::schema::{PrimitiveType, Schema, SchemaStore, TypeName};
+///
+/// let mut store = SchemaStore::new();
+/// let schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::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<'a> {
+ /// The hashing algorithm used for generating fingerprints.
+ fingerprint_algorithm: FingerprintAlgorithm,
+ /// A map from a schema's fingerprint to the schema itself.
+ schemas: HashMap<Fingerprint, Schema<'a>>,
+}
+
+impl<'a> TryFrom<&'a [Schema<'a>]> for SchemaStore<'a> {
+ type Error = ArrowError;
+
+ /// Creates a `SchemaStore` from a slice of schemas.
+ /// Each schema in the slice is registered with the new store.
+ fn try_from(schemas: &'a [Schema<'a>]) -> Result<Self, Self::Error> {
+ let mut store = SchemaStore::new();
+ for schema in schemas {
+ store.register(schema.clone())?;
+ }
+ Ok(store)
+ }
+}
+
+impl<'a> SchemaStore<'a> {
+ /// Creates an empty `SchemaStore` using the default fingerprinting algorithm (64-bit Rabin).
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// 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.
+ ///
+ /// # Arguments
+ ///
+ /// * `schema` - The schema to register.
+ ///
+ /// # Returns
+ ///
+ /// A `Result` containing the `Fingerprint` of the schema if successful,
+ /// or an `ArrowError` on failure.
+ pub fn register(&mut self, schema: Schema<'a>) -> Result<Fingerprint, ArrowError> {
+ let fingerprint = generate_fingerprint(&schema, self.fingerprint_algorithm)?;
+ 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)
+ }
+
+ /// 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 `Schema` if found, otherwise `None`.
+ pub fn lookup(&self, fingerprint: &Fingerprint) -> Option<&Schema<'a>> {
+ 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`.
+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| {
+ let field_type =
+ build_canonical(&f.r#type, child_ns.as_deref().or(enclosing_ns))?;
+ Ok(format!(
+ r#"{{"name":{},"type":{}}}"#,
Review Comment:
I believe the whitespace is handled automatically by Serde, but the canonical form also mandates attribute order, absence of extraneous keys, and deterministic byte output. As I understand it, `json!` produces a `serde_json::Value` whose serialization order depends on map implementation and cargo features, and always allocates owned Strings.
Building the fragment with `format!` avoids those pitfalls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add arrow-avro `SchemaStore` and fingerprinting [arrow-rs]
Posted by "jecsand838 (via GitHub)" <gi...@apache.org>.
jecsand838 commented on PR #8039:
URL: https://github.com/apache/arrow-rs/pull/8039#issuecomment-3155960947
@scovich
> LGTM.
>
> Aside: I get the appeal of zero-copy schemas, but I'm pretty sure this schema store will be very difficult to use in practice unless all possible schemas are known up front. Adding a new schema to the store partway through decoding will be ~impossible. But that's a problem with the existing schema API, not this new schema store.
Avro pretty much requires you to know all possible schemas upfront. The one inconvenience I can foresee is related to developing a `SchemaStore` trait which can stay up to date with an external registry. However in a real world scenario what would likely occur is the current `reader_schema` becoming another `writer_schema` and a new `reader_schema` being assigned. So this would have more complications than just the lifetimes.
I think for this initial implementation it's acceptable to have the caller responsible for making a new `Decoder` upon schema change. Just my 2 cents though of course.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add arrow-avro `SchemaStore` and fingerprinting [arrow-rs]
Posted by "scovich (via GitHub)" <gi...@apache.org>.
scovich commented on code in PR #8039:
URL: https://github.com/apache/arrow-rs/pull/8039#discussion_r2254286137
##########
arrow-avro/src/schema.rs:
##########
@@ -260,13 +276,388 @@ pub struct Fixed<'a> {
pub attributes: Attributes<'a>,
}
+/// Supported fingerprint algorithms for Avro schema identification.
+/// Currently only `Rabin` is supported, `SHA256` and `MD5` support will come in a future update
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Default)]
+pub enum FingerprintAlgorithm {
+ /// 64‑bit CRC‑64‑AVRO Rabin fingerprint.
+ #[default]
+ Rabin,
+}
+
+/// 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.
+/// Currently only `Rabin` is supported
+///
+/// <https://avro.apache.org/docs/1.11.1/specification/#schema-fingerprints>
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub enum Fingerprint {
+ /// A 64-bit Rabin fingerprint.
+ Rabin(u64),
+}
+
+/// 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,
+ }
+ }
+}
+
+/// Generates a fingerprint for the given `Schema` using the specified `FingerprintAlgorithm`.
+pub(crate) fn generate_fingerprint(
+ schema: &Schema,
+ hash_type: FingerprintAlgorithm,
+) -> Result<Fingerprint, ArrowError> {
+ let canonical = 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)))
+ }
+ }
+}
+
+/// Generates the 64-bit Rabin fingerprint for the given `Schema`.
+///
+/// The fingerprint is computed from the canonical form of the schema.
+/// This is also known as `CRC-64-AVRO`.
+///
+/// # Returns
+/// A `Fingerprint::Rabin` variant containing the 64-bit fingerprint.
+pub fn generate_fingerprint_rabin(schema: &Schema) -> Result<Fingerprint, ArrowError> {
+ generate_fingerprint(schema, FingerprintAlgorithm::Rabin)
+}
+
+/// 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 fn generate_canonical_form(schema: &Schema) -> Result<String, ArrowError> {
+ build_canonical(schema, None)
+}
+
+/// 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.
+///
+/// The lifetime parameter `'a` corresponds to the lifetime of the string slices
+/// contained within the stored [`Schema`] objects. This means the `SchemaStore`
+/// cannot outlive the data referenced by the schemas it contains.
+///
+/// # Examples
+///
+/// ```no_run
+/// // Create a new store with the default Rabin fingerprinting.
+/// use arrow_avro::schema::{PrimitiveType, Schema, SchemaStore, TypeName};
+///
+/// let mut store = SchemaStore::new();
+/// let schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::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<'a> {
+ /// The hashing algorithm used for generating fingerprints.
+ fingerprint_algorithm: FingerprintAlgorithm,
+ /// A map from a schema's fingerprint to the schema itself.
+ schemas: HashMap<Fingerprint, Schema<'a>>,
+}
+
+impl<'a> TryFrom<&'a [Schema<'a>]> for SchemaStore<'a> {
+ type Error = ArrowError;
+
+ /// Creates a `SchemaStore` from a slice of schemas.
+ /// Each schema in the slice is registered with the new store.
+ fn try_from(schemas: &'a [Schema<'a>]) -> Result<Self, Self::Error> {
+ let mut store = SchemaStore::new();
+ for schema in schemas {
+ store.register(schema.clone())?;
+ }
+ Ok(store)
+ }
+}
+
+impl<'a> SchemaStore<'a> {
+ /// Creates an empty `SchemaStore` using the default fingerprinting algorithm (64-bit Rabin).
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// 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.
+ ///
+ /// # Arguments
+ ///
+ /// * `schema` - The schema to register.
+ ///
+ /// # Returns
+ ///
+ /// A `Result` containing the `Fingerprint` of the schema if successful,
+ /// or an `ArrowError` on failure.
+ pub fn register(&mut self, schema: Schema<'a>) -> Result<Fingerprint, ArrowError> {
+ let fingerprint = generate_fingerprint(&schema, self.fingerprint_algorithm)?;
+ 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)
+ }
+
+ /// 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 `Schema` if found, otherwise `None`.
+ pub fn lookup(&self, fingerprint: &Fingerprint) -> Option<&Schema<'a>> {
+ 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`.
+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| {
+ let field_type =
+ build_canonical(&f.r#type, child_ns.as_deref().or(enclosing_ns))?;
+ Ok(format!(
+ r#"{{"name":{},"type":{}}}"#,
Review Comment:
What's the difference between this and the `json!` macro (since we anyway have a dependency on serde_json crate)? I guess the macro uses too much whitespace that avro canonical schema forbids?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org