Merge branch 'main' into issue-8670-decimal-cast-refactor

Author: liamzwbao

CHANGELOG-old.md

@@ -68,7 +68,7 @@ exclude = [
 ]
 
 [workspace.package]
-version = "56.2.0"
+version = "57.0.0"
 homepage = "https://github.com/apache/arrow-rs"
 repository = "https://github.com/apache/arrow-rs"
 authors = ["Apache Arrow <[email protected]>"]
@@ -85,22 +85,22 @@ edition = "2024"
 rust-version = "1.85"
 
 [workspace.dependencies]
-arrow = { version = "56.2.0", path = "./arrow", default-features = false }
-arrow-arith = { version = "56.2.0", path = "./arrow-arith" }
-arrow-array = { version = "56.2.0", path = "./arrow-array" }
-arrow-buffer = { version = "56.2.0", path = "./arrow-buffer" }
-arrow-cast = { version = "56.2.0", path = "./arrow-cast" }
-arrow-csv = { version = "56.2.0", path = "./arrow-csv" }
-arrow-data = { version = "56.2.0", path = "./arrow-data" }
-arrow-ipc = { version = "56.2.0", path = "./arrow-ipc" }
-arrow-json = { version = "56.2.0", path = "./arrow-json" }
-arrow-ord = { version = "56.2.0", path = "./arrow-ord" }
-arrow-pyarrow = { version = "56.2.0", path = "./arrow-pyarrow" }
-arrow-row = { version = "56.2.0", path = "./arrow-row" }
-arrow-schema = { version = "56.2.0", path = "./arrow-schema" }
-arrow-select = { version = "56.2.0", path = "./arrow-select" }
-arrow-string = { version = "56.2.0", path = "./arrow-string" }
-parquet = { version = "56.2.0", path = "./parquet", default-features = false }
+arrow = { version = "57.0.0", path = "./arrow", default-features = false }
+arrow-arith = { version = "57.0.0", path = "./arrow-arith" }
+arrow-array = { version = "57.0.0", path = "./arrow-array" }
+arrow-buffer = { version = "57.0.0", path = "./arrow-buffer" }
+arrow-cast = { version = "57.0.0", path = "./arrow-cast" }
+arrow-csv = { version = "57.0.0", path = "./arrow-csv" }
+arrow-data = { version = "57.0.0", path = "./arrow-data" }
+arrow-ipc = { version = "57.0.0", path = "./arrow-ipc" }
+arrow-json = { version = "57.0.0", path = "./arrow-json" }
+arrow-ord = { version = "57.0.0", path = "./arrow-ord" }
+arrow-pyarrow = { version = "57.0.0", path = "./arrow-pyarrow" }
+arrow-row = { version = "57.0.0", path = "./arrow-row" }
+arrow-schema = { version = "57.0.0", path = "./arrow-schema" }
+arrow-select = { version = "57.0.0", path = "./arrow-select" }
+arrow-string = { version = "57.0.0", path = "./arrow-string" }
+parquet = { version = "57.0.0", path = "./parquet", default-features = false }
 
 # These crates have not yet been released and thus do not use the workspace version
 parquet-geospatial = { version = "0.1.0", path = "./parquet-geospatial" }

CHANGELOG.md

@@ -466,7 +466,7 @@ impl<OffsetSize: OffsetSizeTrait> From<FixedSizeListArray> for GenericListArray<
             _ => unreachable!(),
         };
 
-        let offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(size, value.len()));
+        let offsets = OffsetBuffer::from_repeated_length(size, value.len());
 
         Self {
             data_type: Self::DATA_TYPE_CONSTRUCTOR(field.clone()),

Cargo.toml

@@ -306,15 +306,30 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
     /// - String length exceeds `u32::MAX`
     #[inline]
     pub fn append_value(&mut self, value: impl AsRef<T::Native>) {
+        self.try_append_value(value).unwrap()
+    }
+
+    /// Appends a value into the builder
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if:
+    /// - String buffer count exceeds `u32::MAX`
+    /// - String length exceeds `u32::MAX`
+    #[inline]
+    pub fn try_append_value(&mut self, value: impl AsRef<T::Native>) -> Result<(), ArrowError> {
         let v: &[u8] = value.as_ref().as_ref();
-        let length: u32 = v.len().try_into().unwrap();
+        let length: u32 = v.len().try_into().map_err(|_| {
+            ArrowError::InvalidArgumentError(format!("String length {} exceeds u32::MAX", v.len()))
+        })?;
+
         if length <= MAX_INLINE_VIEW_LEN {
             let mut view_buffer = [0; 16];
             view_buffer[0..4].copy_from_slice(&length.to_le_bytes());
             view_buffer[4..4 + v.len()].copy_from_slice(v);
             self.views_buffer.push(u128::from_le_bytes(view_buffer));
             self.null_buffer_builder.append_non_null();
-            return;
+            return Ok(());
         }
 
         // Deduplication if:
@@ -339,7 +354,7 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
                     self.views_buffer.push(self.views_buffer[*idx]);
                     self.null_buffer_builder.append_non_null();
                     self.string_tracker = Some((ht, hasher));
-                    return;
+                    return Ok(());
                 }
                 Entry::Vacant(vacant) => {
                     // o.w. we insert the (string hash -> view index)
@@ -356,17 +371,28 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
             let to_reserve = v.len().max(self.block_size.next_size() as usize);
             self.in_progress.reserve(to_reserve);
         };
+
         let offset = self.in_progress.len() as u32;
         self.in_progress.extend_from_slice(v);
 
+        let buffer_index: u32 = self.completed.len().try_into().map_err(|_| {
+            ArrowError::InvalidArgumentError(format!(
+                "Buffer count {} exceeds u32::MAX",
+                self.completed.len()
+            ))
+        })?;
+
         let view = ByteView {
             length,
+            // This won't panic as we checked the length of prefix earlier.
             prefix: u32::from_le_bytes(v[0..4].try_into().unwrap()),
-            buffer_index: self.completed.len() as u32,
+            buffer_index,
             offset,
         };
         self.views_buffer.push(view.into());
         self.null_buffer_builder.append_non_null();
+
+        Ok(())
     }
 
     /// Append an `Option` value into the builder
@@ -581,7 +607,6 @@ mod tests {
     use core::str;
 
     use super::*;
-    use crate::Array;
 
     #[test]
     fn test_string_view_deduplicate() {

arrow-array/src/array/list_array.rs

@@ -1324,7 +1324,7 @@ pub trait DecimalType:
     /// Maximum no of digits after the decimal point (note the scale can be negative)
     const MAX_SCALE: i8;
     /// The maximum value for each precision in `0..=MAX_PRECISION`: [0, 9, 99, ...]
-    const MAX_FOR_EACH_PRECISION: &[Self::Native];
+    const MAX_FOR_EACH_PRECISION: &'static [Self::Native];
     /// fn to create its [`DataType`]
     const TYPE_CONSTRUCTOR: fn(u8, i8) -> DataType;
     /// Default values for [`DataType`]
@@ -1395,7 +1395,8 @@ impl DecimalType for Decimal32Type {
     const BYTE_LENGTH: usize = 4;
     const MAX_PRECISION: u8 = DECIMAL32_MAX_PRECISION;
     const MAX_SCALE: i8 = DECIMAL32_MAX_SCALE;
-    const MAX_FOR_EACH_PRECISION: &[i32] = &arrow_data::decimal::MAX_DECIMAL32_FOR_EACH_PRECISION;
+    const MAX_FOR_EACH_PRECISION: &'static [i32] =
+        &arrow_data::decimal::MAX_DECIMAL32_FOR_EACH_PRECISION;
     const TYPE_CONSTRUCTOR: fn(u8, i8) -> DataType = DataType::Decimal32;
     const DEFAULT_TYPE: DataType =
         DataType::Decimal32(DECIMAL32_MAX_PRECISION, DECIMAL32_DEFAULT_SCALE);
@@ -1430,7 +1431,8 @@ impl DecimalType for Decimal64Type {
     const BYTE_LENGTH: usize = 8;
     const MAX_PRECISION: u8 = DECIMAL64_MAX_PRECISION;
     const MAX_SCALE: i8 = DECIMAL64_MAX_SCALE;
-    const MAX_FOR_EACH_PRECISION: &[i64] = &arrow_data::decimal::MAX_DECIMAL64_FOR_EACH_PRECISION;
+    const MAX_FOR_EACH_PRECISION: &'static [i64] =
+        &arrow_data::decimal::MAX_DECIMAL64_FOR_EACH_PRECISION;
     const TYPE_CONSTRUCTOR: fn(u8, i8) -> DataType = DataType::Decimal64;
     const DEFAULT_TYPE: DataType =
         DataType::Decimal64(DECIMAL64_MAX_PRECISION, DECIMAL64_DEFAULT_SCALE);
@@ -1465,7 +1467,8 @@ impl DecimalType for Decimal128Type {
     const BYTE_LENGTH: usize = 16;
     const MAX_PRECISION: u8 = DECIMAL128_MAX_PRECISION;
     const MAX_SCALE: i8 = DECIMAL128_MAX_SCALE;
-    const MAX_FOR_EACH_PRECISION: &[i128] = &arrow_data::decimal::MAX_DECIMAL128_FOR_EACH_PRECISION;
+    const MAX_FOR_EACH_PRECISION: &'static [i128] =
+        &arrow_data::decimal::MAX_DECIMAL128_FOR_EACH_PRECISION;
     const TYPE_CONSTRUCTOR: fn(u8, i8) -> DataType = DataType::Decimal128;
     const DEFAULT_TYPE: DataType =
         DataType::Decimal128(DECIMAL128_MAX_PRECISION, DECIMAL_DEFAULT_SCALE);
@@ -1500,7 +1503,8 @@ impl DecimalType for Decimal256Type {
     const BYTE_LENGTH: usize = 32;
     const MAX_PRECISION: u8 = DECIMAL256_MAX_PRECISION;
     const MAX_SCALE: i8 = DECIMAL256_MAX_SCALE;
-    const MAX_FOR_EACH_PRECISION: &[i256] = &arrow_data::decimal::MAX_DECIMAL256_FOR_EACH_PRECISION;
+    const MAX_FOR_EACH_PRECISION: &'static [i256] =
+        &arrow_data::decimal::MAX_DECIMAL256_FOR_EACH_PRECISION;
     const TYPE_CONSTRUCTOR: fn(u8, i8) -> DataType = DataType::Decimal256;
     const DEFAULT_TYPE: DataType =
         DataType::Decimal256(DECIMAL256_MAX_PRECISION, DECIMAL_DEFAULT_SCALE);

arrow-array/src/builder/generic_bytes_view_builder.rs

@@ -20,13 +20,13 @@
 //! This crate provides:
 //! - a [`reader`] that decodes Avro (Object Container Files, Avro Single‑Object encoding,
 //!   and Confluent Schema Registry wire format) into Arrow `RecordBatch`es,
-//! - and a [`writer`] that encodes Arrow `RecordBatch`es into Avro (OCF or raw Avro binary).
+//! - and a [`writer`] that encodes Arrow `RecordBatch`es into Avro (OCF or SOE).
 //!
 //! If you’re new to Arrow or Avro, see:
 //! - Arrow project site: <https://arrow.apache.org/>
 //! - Avro 1.11.1 specification: <https://avro.apache.org/docs/1.11.1/specification/>
 //!
-//! ## Example: OCF (Object Container File) round‑trip
+//! ## Example: OCF (Object Container File) round‑trip *(runnable)*
 //!
 //! The example below creates an Arrow table, writes an **Avro OCF** fully in memory,
 //! and then reads it back. OCF is a self‑describing file format that embeds the Avro
@@ -64,82 +64,7 @@
 //! # Ok(()) }
 //! ```
 //!
-//! ## Quickstart: Confluent wire‑format round‑trip *(runnable)*
-//!
-//! The **Confluent Schema Registry wire format** prefixes each Avro message with a
-//! 1‑byte magic `0x00` and a **4‑byte big‑endian** schema ID, followed by the Avro body.
-//! See: <https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format>
-//!
-//! In this round‑trip, we:
-//! 1) Use `AvroStreamWriter` to create a **raw Avro body** for a single‑row batch,
-//! 2) Wrap it with the Confluent prefix (magic and schema ID),
-//! 3) Decode it back to Arrow using a `Decoder` configured with a `SchemaStore` that
-//!    maps the schema ID to the Avro schema used by the writer.
-//!
-//! ```
-//! use std::collections::HashMap;
-//! use std::sync::Arc;
-//! use arrow_array::{ArrayRef, Int64Array, RecordBatch, StringArray};
-//! use arrow_schema::{DataType, Field, Schema};
-//! use arrow_avro::writer::{AvroStreamWriter, WriterBuilder};
-//! use arrow_avro::reader::ReaderBuilder;
-//! use arrow_avro::schema::{
-//!     AvroSchema, SchemaStore, Fingerprint, FingerprintAlgorithm,
-//!     FingerprintStrategy, SCHEMA_METADATA_KEY
-//! };
-//!
-//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
-//! // Writer schema registered under Schema Registry ID 1
-//! let avro_json = r#"{
-//!   "type":"record","name":"User",
-//!   "fields":[{"name":"id","type":"long"},{"name":"name","type":"string"}]
-//! }"#;
-//!
-//! let mut store = SchemaStore::new_with_type(FingerprintAlgorithm::Id);
-//! let id: u32 = 1;
-//! store.set(Fingerprint::Id(id), AvroSchema::new(avro_json.to_string()))?;
-//!
-//! // Build an Arrow schema that references the same Avro JSON
-//! let mut md = HashMap::new();
-//! md.insert(SCHEMA_METADATA_KEY.to_string(), avro_json.to_string());
-//! let schema = Schema::new_with_metadata(
-//!     vec![
-//!         Field::new("id", DataType::Int64, false),
-//!         Field::new("name", DataType::Utf8, false),
-//!     ],
-//!     md,
-//! );
-//!
-//! // One‑row batch: { id: 42, name: "alice" }
-//! let batch = RecordBatch::try_new(
-//!     Arc::new(schema.clone()),
-//!     vec![
-//!         Arc::new(Int64Array::from(vec![42])) as ArrayRef,
-//!         Arc::new(StringArray::from(vec!["alice"])) as ArrayRef,
-//!     ],
-//! )?;
-//!
-//! // Stream‑write a single record, letting the writer add the **Confluent** prefix.
-//! let sink: Vec<u8> = Vec::new();
-//! let mut w: AvroStreamWriter<Vec<u8>> = WriterBuilder::new(schema.clone())
-//!     .with_fingerprint_strategy(FingerprintStrategy::Id(id))
-//!     .build(sink)?;
-//! w.write(&batch)?;
-//! w.finish()?;
-//! let frame = w.into_inner(); // already: 0x00 + 4B BE ID + Avro body
-//! assert!(frame.len() > 5);
-//!
-//! // Decode
-//! let mut dec = ReaderBuilder::new()
-//!   .with_writer_schema_store(store)
-//!   .build_decoder()?;
-//! dec.decode(&frame)?;
-//! let out = dec.flush()?.expect("one row");
-//! assert_eq!(out.num_rows(), 1);
-//! # Ok(()) }
-//! ```
-//!
-//! ## Quickstart: Avro Single‑Object Encoding round‑trip *(runnable)*
+//! ## Quickstart: SOE (Single‑Object Encoding) round‑trip *(runnable)*
 //!
 //! Avro **Single‑Object Encoding (SOE)** wraps an Avro body with a 2‑byte marker
 //! `0xC3 0x01` and an **8‑byte little‑endian CRC‑64‑AVRO Rabin fingerprint** of the
@@ -203,7 +128,7 @@
 //! ### Modules
 //!
 //! - [`reader`]: read Avro (OCF, SOE, Confluent) into Arrow `RecordBatch`es.
-//! - [`writer`]: write Arrow `RecordBatch`es as Avro (OCF, SOE, Confluent).
+//! - [`writer`]: write Arrow `RecordBatch`es as Avro (OCF, SOE, Confluent, Apicurio).
 //! - [`schema`]: Avro schema parsing / fingerprints / registries.
 //! - [`compression`]: codecs used for **OCF block compression** (i.e., Deflate, Snappy, Zstandard, BZip2, and XZ).
 //! - [`codec`]: internal Avro-Arrow type conversion and row decode/encode plans.

arrow-array/src/types.rs

@@ -42,7 +42,7 @@
 //! * [`Reader`](crate::reader::Reader): a convenient, synchronous iterator over `RecordBatch` decoded from an OCF
 //!   input. Implements [`Iterator<Item = Result<RecordBatch, ArrowError>>`] and
 //!   `RecordBatchReader`.
-//! * [`Decoder`](crate::reader::Decoder): a push‑based row decoder that consumes raw Avro bytes and yields ready
+//! * [`Decoder`](crate::reader::Decoder): a push‑based row decoder that consumes SOE framed Avro bytes and yields ready
 //!   `RecordBatch` values when batches fill. This is suitable for integrating with async
 //!   byte streams, network protocols, or other custom data sources.
 //!
@@ -167,7 +167,7 @@
 //! use arrow_array::{ArrayRef, Int64Array, RecordBatch};
 //! use arrow_schema::{DataType, Field, Schema};
 //! use arrow_avro::schema::{AvroSchema, SchemaStore, SCHEMA_METADATA_KEY, FingerprintStrategy};
-//! use arrow_avro::writer::{WriterBuilder, format::AvroBinaryFormat};
+//! use arrow_avro::writer::{WriterBuilder, format::AvroSoeFormat};
 //! use arrow_avro::reader::ReaderBuilder;
 //!
 //! # fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -187,7 +187,7 @@
 //! )?;
 //! let mut w = WriterBuilder::new(arrow)
 //!     .with_fingerprint_strategy(FingerprintStrategy::Rabin) // SOE prefix
-//!     .build::<_, AvroBinaryFormat>(Vec::new())?;
+//!     .build::<_, AvroSoeFormat>(Vec::new())?;
 //! w.write(&batch)?;
 //! w.finish()?;
 //! let frame = w.into_inner(); // C3 01 + fp + Avro body
@@ -220,7 +220,7 @@
 //! use arrow_array::{ArrayRef, Int64Array, StringArray, RecordBatch};
 //! use arrow_schema::{DataType, Field, Schema};
 //! use arrow_avro::schema::{AvroSchema, SchemaStore, Fingerprint, FingerprintAlgorithm, SCHEMA_METADATA_KEY, FingerprintStrategy};
-//! use arrow_avro::writer::{WriterBuilder, format::AvroBinaryFormat};
+//! use arrow_avro::writer::{WriterBuilder, format::AvroSoeFormat};
 //! use arrow_avro::reader::ReaderBuilder;
 //!
 //! # fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -248,7 +248,7 @@
 //!     )?;
 //!     let mut w = WriterBuilder::new(arrow)
 //!         .with_fingerprint_strategy(FingerprintStrategy::Id(schema_id)) // 0x00 + ID + body
-//!         .build::<_, AvroBinaryFormat>(Vec::new())?;
+//!         .build::<_, AvroSoeFormat>(Vec::new())?;
 //!     w.write(&batch)?; w.finish()?;
 //!     Ok(w.into_inner())
 //! }
@@ -402,7 +402,7 @@
 //!              Arc::new(StringArray::from(vec!["v0-alice"])) as ArrayRef])?;
 //!     let mut w0 = arrow_avro::writer::WriterBuilder::new(arrow0)
 //!         .with_fingerprint_strategy(FingerprintStrategy::Id(id_v0))
-//!         .build::<_, arrow_avro::writer::format::AvroBinaryFormat>(Vec::new())?;
+//!         .build::<_, arrow_avro::writer::format::AvroSoeFormat>(Vec::new())?;
 //!     w0.write(&batch0)?; w0.finish()?;
 //!     let frame0 = w0.into_inner(); // 0x00 + id_v0 + body
 //!
@@ -420,7 +420,7 @@
 //!              Arc::new(StringArray::from(vec![Some("[email protected]")])) as ArrayRef])?;
 //!     let mut w1 = arrow_avro::writer::WriterBuilder::new(arrow1)
 //!         .with_fingerprint_strategy(FingerprintStrategy::Id(id_v1))
-//!         .build::<_, arrow_avro::writer::format::AvroBinaryFormat>(Vec::new())?;
+//!         .build::<_, arrow_avro::writer::format::AvroSoeFormat>(Vec::new())?;
 //!     w1.write(&batch1)?; w1.finish()?;
 //!     let frame1 = w1.into_inner(); // 0x00 + id_v1 + body
 //!
@@ -563,15 +563,15 @@ fn is_incomplete_data(err: &ArrowError) -> bool {
 /// # use arrow_array::{ArrayRef, Int64Array, RecordBatch};
 /// # use arrow_schema::{DataType, Field, Schema};
 /// # use arrow_avro::schema::{SCHEMA_METADATA_KEY, FingerprintStrategy};
-/// # use arrow_avro::writer::{WriterBuilder, format::AvroBinaryFormat};
+/// # use arrow_avro::writer::{WriterBuilder, format::AvroSoeFormat};
 /// # let mut md = HashMap::new();
 /// # md.insert(SCHEMA_METADATA_KEY.to_string(),
 /// #     r#"{"type":"record","name":"E","fields":[{"name":"x","type":"long"}]}"#.to_string());
 /// # let arrow = Schema::new_with_metadata(vec![Field::new("x", DataType::Int64, false)], md);
 /// # let batch = RecordBatch::try_new(Arc::new(arrow.clone()), vec![Arc::new(Int64Array::from(vec![7])) as ArrayRef])?;
 /// # let mut w = WriterBuilder::new(arrow)
 /// #     .with_fingerprint_strategy(fp.into())
-/// #     .build::<_, AvroBinaryFormat>(Vec::new())?;
+/// #     .build::<_, AvroSoeFormat>(Vec::new())?;
 /// # w.write(&batch)?; w.finish()?; let frame = w.into_inner();
 ///
 /// let mut decoder = ReaderBuilder::new()
@@ -605,7 +605,7 @@ fn is_incomplete_data(err: &ArrowError) -> bool {
 /// # use arrow_array::{ArrayRef, Int64Array, RecordBatch};
 /// # use arrow_schema::{DataType, Field, Schema};
 /// # use arrow_avro::schema::{SCHEMA_METADATA_KEY, FingerprintStrategy};
-/// # use arrow_avro::writer::{WriterBuilder, format::AvroBinaryFormat};
+/// # use arrow_avro::writer::{WriterBuilder, format::AvroSoeFormat};
 /// # fn msg(x: i64) -> Result<Vec<u8>, Box<dyn std::error::Error>> {
 /// #   let mut md = HashMap::new();
 /// #   md.insert(SCHEMA_METADATA_KEY.to_string(),
@@ -614,7 +614,7 @@ fn is_incomplete_data(err: &ArrowError) -> bool {
 /// #   let batch = RecordBatch::try_new(Arc::new(arrow.clone()), vec![Arc::new(Int64Array::from(vec![x])) as ArrayRef])?;
 /// #   let mut w = WriterBuilder::new(arrow)
 /// #       .with_fingerprint_strategy(FingerprintStrategy::Id(1234))
-/// #       .build::<_, AvroBinaryFormat>(Vec::new())?;
+/// #       .build::<_, AvroSoeFormat>(Vec::new())?;
 /// #   w.write(&batch)?; w.finish()?; Ok(w.into_inner())
 /// # }
 /// # let m1 = msg(1)?;