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/09/08 02:37:26 UTC
[PR] Add array/map/fixed schema resolution and default value support to arrow-avro codec [arrow-rs]
jecsand838 opened a new pull request, #8292:
URL: https://github.com/apache/arrow-rs/pull/8292
# Which issue does this PR close?
This work continues arrow-avro schema resolution support and aligns behavior with the Avro spec.
- **Related to**: #4886 (“Add Avro Support”): ongoing work to round out the reader/decoder, including schema resolution and type promotion.
- **Follow-ups/Context**: #8124 (schema resolution & type promotion for the decoder), #8223 (enum mapping for schema resolution). These previous efforts established the foundations that this PR extends to default values and additional resolvable types.
# Rationale for this change
Avro’s **schema resolution** requires readers to reconcile differences between the writer and reader schemas, including:
- Using record-field **default values** when the writer lacks a field present in the reader; defaults must be type-correct (i.e., union defaults match the first union member; bytes/fixed defaults are JSON strings).
- Recursively resolving **arrays** (by item schema) and **maps** (by value schema).
- Resolving **fixed** types (size and unqualified name must match) and erroring when they do not.
Prior to this change, arrow-avro’s resolution handled some cases but lacked full Codec support for **default values** and for resolving **array/map/fixed** shapes between writer and reader. This led to gaps when reading evolved data or datasets produced by heterogeneous systems. This PR implements these missing pieces so the Arrow reader behaves per the spec in common evolution scenarios.
# What changes are included in this PR?
This PR modifies **`arrow-avro/src/codec.rs`** to extend the schema-resolution path
- **Default value handling** for record fields
- Reads and applies default values when the reader expects a field absent from the writer, including **nested defaults**.
- Validates defaults per the Avro spec (e.g., union defaults match the first schema; bytes/fixed defaults are JSON strings).
- **Array / Map / Fixed schema resolution**
- **Array**: recursively resolves item schemas (writer↔reader).
- **Map**: recursively resolves value schemas.
- **Fixed**: enforces matching size and (unqualified) name; otherwise signals an error, consistent with the spec.
- **Codec updates**
- Refactors internal codec logic to support the above during decoding, including resolution for **record fields** and **nested defaults**. (See commit message for the high-level summary.)
# Are these changes tested?
**Yes.** This PR includes new unit tests in `arrow-avro/src/codec.rs` covering:
1) **Default validation & persistence**
- `Null`/union‑nullability rules; metadata persistence of defaults (`AVRO_FIELD_DEFAULT_METADATA_KEY`).
2) **`AvroLiteral` Parsing**
- Range checks for `i32`/`f32`; correct literals for `i64`/`f64`; `Utf8`/`Utf8View`; `uuid` strings (RFC‑4122).
- Byte‑range mapping for `bytes`/`fixed` defaults; `Fixed(n)` length enforcement; `decimal` on `fixed` vs `bytes`; `duration`/interval fixed **12**‑byte enforcement.
3) **Collections & records**
- Array/map defaults shape; enum symbol validity; record defaults for missing fields, required‑field errors, and honoring field‑level defaults; skip‑fields retained for writer‑only fields.
4) **Resolution mechanics**
- Element **promotion** (`int` to `long`) for arrays; **reader metadata precedence** for colliding attributes; `fixed` name/size match including **alias**.
# 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 array/map/fixed schema resolution and default value support to arrow-avro codec [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #8292:
URL: https://github.com/apache/arrow-rs/pull/8292#discussion_r2337661495
##########
arrow-avro/src/codec.rs:
##########
@@ -1712,4 +2032,440 @@ mod tests {
panic!("Top-level schema is not a struct");
}
}
+
+ fn json_string(s: &str) -> Value {
+ Value::String(s.to_string())
+ }
+
+ fn assert_default_stored(dt: &AvroDataType, default_json: &Value) {
+ let stored = dt
+ .metadata
+ .get(AVRO_FIELD_DEFAULT_METADATA_KEY)
+ .cloned()
+ .unwrap_or_default();
+ let expected = serde_json::to_string(default_json).unwrap();
+ assert_eq!(stored, expected, "stored default metadata should match");
+ }
+
+ #[test]
+ fn test_validate_and_store_default_null_and_nullability_rules() {
+ let mut dt_null = AvroDataType::new(Codec::Null, HashMap::new(), None);
+ let lit = dt_null.parse_and_store_default(&Value::Null).unwrap();
+ assert_eq!(lit, AvroLiteral::Null);
+ assert_default_stored(&dt_null, &Value::Null);
+ let mut dt_int = AvroDataType::new(Codec::Int32, HashMap::new(), None);
+ let err = dt_int.parse_and_store_default(&Value::Null).unwrap_err();
+ assert!(
+ err.to_string()
+ .contains("JSON null default is only valid for `null` type"),
+ "unexpected error: {err}"
+ );
+ let mut dt_int_nf =
+ AvroDataType::new(Codec::Int32, HashMap::new(), Some(Nullability::NullFirst));
+ let lit2 = dt_int_nf.parse_and_store_default(&Value::Null).unwrap();
+ assert_eq!(lit2, AvroLiteral::Null);
+ assert_default_stored(&dt_int_nf, &Value::Null);
+ let mut dt_int_ns =
+ AvroDataType::new(Codec::Int32, HashMap::new(), Some(Nullability::NullSecond));
+ let err2 = dt_int_ns.parse_and_store_default(&Value::Null).unwrap_err();
+ assert!(
+ err2.to_string()
+ .contains("JSON null default is only valid for `null` type"),
+ "unexpected error: {err2}"
+ );
+ }
+
+ #[test]
+ fn test_validate_and_store_default_primitives_and_temporal() {
+ let mut dt_bool = AvroDataType::new(Codec::Boolean, HashMap::new(), None);
+ let lit = dt_bool.parse_and_store_default(&Value::Bool(true)).unwrap();
+ assert_eq!(lit, AvroLiteral::Boolean(true));
+ assert_default_stored(&dt_bool, &Value::Bool(true));
+ let mut dt_i32 = AvroDataType::new(Codec::Int32, HashMap::new(), None);
+ let lit = dt_i32
+ .parse_and_store_default(&serde_json::json!(123))
+ .unwrap();
+ assert_eq!(lit, AvroLiteral::Int(123));
+ assert_default_stored(&dt_i32, &serde_json::json!(123));
+ let err = dt_i32
+ .parse_and_store_default(&serde_json::json!(i64::from(i32::MAX) + 1))
+ .unwrap_err();
+ assert!(format!("{err}").contains("out of i32 range"));
+ let mut dt_i64 = AvroDataType::new(Codec::Int64, HashMap::new(), None);
+ let lit = dt_i64
+ .parse_and_store_default(&serde_json::json!(1234567890))
+ .unwrap();
+ assert_eq!(lit, AvroLiteral::Long(1234567890));
+ assert_default_stored(&dt_i64, &serde_json::json!(1234567890));
+ let mut dt_f32 = AvroDataType::new(Codec::Float32, HashMap::new(), None);
+ let lit = dt_f32
+ .parse_and_store_default(&serde_json::json!(1.25))
+ .unwrap();
+ assert_eq!(lit, AvroLiteral::Float(1.25));
+ assert_default_stored(&dt_f32, &serde_json::json!(1.25));
+ let err = dt_f32
+ .parse_and_store_default(&serde_json::json!(1e39))
+ .unwrap_err();
+ assert!(format!("{err}").contains("out of f32 range"));
+ let mut dt_f64 = AvroDataType::new(Codec::Float64, HashMap::new(), None);
+ let lit = dt_f64
+ .parse_and_store_default(&serde_json::json!(std::f64::consts::PI))
+ .unwrap();
+ assert_eq!(lit, AvroLiteral::Double(std::f64::consts::PI));
+ assert_default_stored(&dt_f64, &serde_json::json!(std::f64::consts::PI));
+ let mut dt_str = AvroDataType::new(Codec::Utf8, HashMap::new(), None);
+ let l = dt_str
+ .parse_and_store_default(&json_string("hello"))
+ .unwrap();
+ assert_eq!(l, AvroLiteral::String("hello".into()));
+ assert_default_stored(&dt_str, &json_string("hello"));
+ let mut dt_strv = AvroDataType::new(Codec::Utf8View, HashMap::new(), None);
+ let l = dt_strv
+ .parse_and_store_default(&json_string("view"))
+ .unwrap();
+ assert_eq!(l, AvroLiteral::String("view".into()));
+ assert_default_stored(&dt_strv, &json_string("view"));
+ let mut dt_uuid = AvroDataType::new(Codec::Uuid, HashMap::new(), None);
+ let l = dt_uuid
+ .parse_and_store_default(&json_string("00000000-0000-0000-0000-000000000000"))
+ .unwrap();
+ assert_eq!(
+ l,
+ AvroLiteral::String("00000000-0000-0000-0000-000000000000".into())
+ );
+ let mut dt_bin = AvroDataType::new(Codec::Binary, HashMap::new(), None);
+ let l = dt_bin.parse_and_store_default(&json_string("ABC")).unwrap();
+ assert_eq!(l, AvroLiteral::Bytes(vec![65, 66, 67]));
+ let err = dt_bin
+ .parse_and_store_default(&json_string("€")) // U+20AC
+ .unwrap_err();
+ assert!(format!("{err}").contains("Invalid codepoint"));
+ let mut dt_date = AvroDataType::new(Codec::Date32, HashMap::new(), None);
+ let ld = dt_date
+ .parse_and_store_default(&serde_json::json!(1))
+ .unwrap();
+ assert_eq!(ld, AvroLiteral::Int(1));
+ let mut dt_tmill = AvroDataType::new(Codec::TimeMillis, HashMap::new(), None);
+ let lt = dt_tmill
+ .parse_and_store_default(&serde_json::json!(86_400_000))
+ .unwrap();
+ assert_eq!(lt, AvroLiteral::Int(86_400_000));
+ let mut dt_tmicros = AvroDataType::new(Codec::TimeMicros, HashMap::new(), None);
+ let ltm = dt_tmicros
+ .parse_and_store_default(&serde_json::json!(1_000_000))
+ .unwrap();
+ assert_eq!(ltm, AvroLiteral::Long(1_000_000));
+ let mut dt_ts_milli = AvroDataType::new(Codec::TimestampMillis(true), HashMap::new(), None);
+ let l1 = dt_ts_milli
+ .parse_and_store_default(&serde_json::json!(123))
+ .unwrap();
+ assert_eq!(l1, AvroLiteral::Long(123));
+ let mut dt_ts_micro =
+ AvroDataType::new(Codec::TimestampMicros(false), HashMap::new(), None);
+ let l2 = dt_ts_micro
+ .parse_and_store_default(&serde_json::json!(456))
+ .unwrap();
+ assert_eq!(l2, AvroLiteral::Long(456));
+ }
+
+ #[test]
+ fn test_validate_and_store_default_fixed_decimal_interval() {
+ let mut dt_fixed = AvroDataType::new(Codec::Fixed(4), HashMap::new(), None);
+ let l = dt_fixed
+ .parse_and_store_default(&json_string("WXYZ"))
+ .unwrap();
+ assert_eq!(l, AvroLiteral::Bytes(vec![87, 88, 89, 90]));
+ let err = dt_fixed
+ .parse_and_store_default(&json_string("TOO LONG"))
+ .unwrap_err();
+ assert!(format!("{err}").contains("Default length"));
Review Comment:
minor thing is that you can write this like
```suggestion
assert!(err.to_string().contains("Default length"));
```
And also below, but I think this also works fine
--
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 array/map/fixed schema resolution and default value support to arrow-avro codec [arrow-rs]
Posted by "jecsand838 (via GitHub)" <gi...@apache.org>.
jecsand838 commented on PR #8292:
URL: https://github.com/apache/arrow-rs/pull/8292#issuecomment-3272467821
@alamb Would you be able to review this PR whenever you get a chance? About 60% of the new code are tests.
--
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 array/map/fixed schema resolution and default value support to arrow-avro codec [arrow-rs]
Posted by "jecsand838 (via GitHub)" <gi...@apache.org>.
jecsand838 commented on code in PR #8292:
URL: https://github.com/apache/arrow-rs/pull/8292#discussion_r2337928702
##########
arrow-avro/src/codec.rs:
##########
@@ -193,6 +200,224 @@ impl AvroDataType {
pub fn nullability(&self) -> Option<Nullability> {
self.nullability
}
+
+ #[inline]
+ fn parse_default_literal(&self, default_json: &Value) -> Result<AvroLiteral, ArrowError> {
+ fn expect_string<'v>(
+ default_json: &'v Value,
+ data_type: &str,
+ ) -> Result<&'v str, ArrowError> {
+ match default_json {
+ Value::String(s) => Ok(s.as_str()),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default value must be a JSON string for {data_type}"
+ ))),
+ }
+ }
+
+ fn parse_bytes_default(
+ default_json: &Value,
+ expected_len: Option<usize>,
+ ) -> Result<Vec<u8>, ArrowError> {
+ let s = expect_string(default_json, "bytes/fixed logical types")?;
+ let mut out = Vec::with_capacity(s.len());
+ for ch in s.chars() {
+ let cp = ch as u32;
+ if cp > 0xFF {
+ return Err(ArrowError::SchemaError(format!(
+ "Invalid codepoint U+{cp:04X} in bytes/fixed default; must be ≤ 0xFF"
+ )));
+ }
+ out.push(cp as u8);
+ }
+ if let Some(len) = expected_len {
+ if out.len() != len {
+ return Err(ArrowError::SchemaError(format!(
+ "Default length {} does not match expected fixed size {len}",
+ out.len(),
+ )));
+ }
+ }
+ Ok(out)
+ }
+
+ fn parse_json_i64(default_json: &Value, data_type: &str) -> Result<i64, ArrowError> {
+ match default_json {
+ Value::Number(n) => n.as_i64().ok_or_else(|| {
+ ArrowError::SchemaError(format!("Default {data_type} must be an integer"))
+ }),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default {data_type} must be a JSON integer"
+ ))),
+ }
+ }
+
+ fn parse_json_f64(default_json: &Value, data_type: &str) -> Result<f64, ArrowError> {
+ match default_json {
+ Value::Number(n) => n.as_f64().ok_or_else(|| {
+ ArrowError::SchemaError(format!("Default {data_type} must be a number"))
+ }),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default {data_type} must be a JSON number"
+ ))),
+ }
+ }
+
+ // Handle JSON nulls per-spec: allowed only for `null` type or unions with null FIRST
Review Comment:
That's a really good catch. I was planning to bring in `NullSecond` support here as a part of the full Dense `Union` type decoding support I'm finishing up right now. Which is why I left the comment calling it out.
--
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 array/map/fixed schema resolution and default value support to arrow-avro codec [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #8292:
URL: https://github.com/apache/arrow-rs/pull/8292#issuecomment-3281580097
Thank you @jecsand838 and @nathaniel-d-ef
--
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 array/map/fixed schema resolution and default value support to arrow-avro codec [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #8292:
URL: https://github.com/apache/arrow-rs/pull/8292
--
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 array/map/fixed schema resolution and default value support to arrow-avro codec [arrow-rs]
Posted by "jecsand838 (via GitHub)" <gi...@apache.org>.
jecsand838 commented on code in PR #8292:
URL: https://github.com/apache/arrow-rs/pull/8292#discussion_r2338008633
##########
arrow-avro/src/codec.rs:
##########
@@ -193,6 +200,224 @@ impl AvroDataType {
pub fn nullability(&self) -> Option<Nullability> {
self.nullability
}
+
+ #[inline]
+ fn parse_default_literal(&self, default_json: &Value) -> Result<AvroLiteral, ArrowError> {
+ fn expect_string<'v>(
+ default_json: &'v Value,
+ data_type: &str,
+ ) -> Result<&'v str, ArrowError> {
+ match default_json {
+ Value::String(s) => Ok(s.as_str()),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default value must be a JSON string for {data_type}"
+ ))),
+ }
+ }
+
+ fn parse_bytes_default(
+ default_json: &Value,
+ expected_len: Option<usize>,
+ ) -> Result<Vec<u8>, ArrowError> {
+ let s = expect_string(default_json, "bytes/fixed logical types")?;
+ let mut out = Vec::with_capacity(s.len());
+ for ch in s.chars() {
+ let cp = ch as u32;
+ if cp > 0xFF {
+ return Err(ArrowError::SchemaError(format!(
+ "Invalid codepoint U+{cp:04X} in bytes/fixed default; must be ≤ 0xFF"
+ )));
+ }
+ out.push(cp as u8);
+ }
+ if let Some(len) = expected_len {
+ if out.len() != len {
+ return Err(ArrowError::SchemaError(format!(
+ "Default length {} does not match expected fixed size {len}",
+ out.len(),
+ )));
+ }
+ }
+ Ok(out)
+ }
+
+ fn parse_json_i64(default_json: &Value, data_type: &str) -> Result<i64, ArrowError> {
+ match default_json {
+ Value::Number(n) => n.as_i64().ok_or_else(|| {
+ ArrowError::SchemaError(format!("Default {data_type} must be an integer"))
+ }),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default {data_type} must be a JSON integer"
+ ))),
+ }
+ }
+
+ fn parse_json_f64(default_json: &Value, data_type: &str) -> Result<f64, ArrowError> {
+ match default_json {
+ Value::Number(n) => n.as_f64().ok_or_else(|| {
+ ArrowError::SchemaError(format!("Default {data_type} must be a number"))
+ }),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default {data_type} must be a JSON number"
+ ))),
+ }
+ }
+
+ // Handle JSON nulls per-spec: allowed only for `null` type or unions with null FIRST
+ if default_json.is_null() {
+ return match self.codec() {
+ Codec::Null => Ok(AvroLiteral::Null),
+ _ if self.nullability() == Some(Nullability::NullFirst) => Ok(AvroLiteral::Null),
+ _ => Err(ArrowError::SchemaError(
+ "JSON null default is only valid for `null` type or for a union whose first branch is `null`"
+ .to_string(),
+ )),
+ };
+ }
+ let lit = match self.codec() {
+ Codec::Null => {
+ return Err(ArrowError::SchemaError(
+ "Default for `null` type must be JSON null".to_string(),
+ ))
+ }
+ Codec::Boolean => match default_json {
+ Value::Bool(b) => AvroLiteral::Boolean(*b),
+ _ => {
+ return Err(ArrowError::SchemaError(
+ "Boolean default must be a JSON boolean".to_string(),
+ ))
+ }
+ },
+ Codec::Int32 | Codec::Date32 | Codec::TimeMillis => {
+ let i = parse_json_i64(default_json, "int")?;
+ if i < i32::MIN as i64 || i > i32::MAX as i64 {
+ return Err(ArrowError::SchemaError(format!(
+ "Default int {i} out of i32 range"
+ )));
+ }
+ AvroLiteral::Int(i as i32)
+ }
+ Codec::Int64
+ | Codec::TimeMicros
+ | Codec::TimestampMillis(_)
+ | Codec::TimestampMicros(_) => AvroLiteral::Long(parse_json_i64(default_json, "long")?),
+ Codec::Float32 => {
+ let f = parse_json_f64(default_json, "float")?;
+ if !f.is_finite() || f < f32::MIN as f64 || f > f32::MAX as f64 {
+ return Err(ArrowError::SchemaError(format!(
+ "Default float {f} out of f32 range or not finite"
+ )));
+ }
+ AvroLiteral::Float(f as f32)
+ }
+ Codec::Float64 => AvroLiteral::Double(parse_json_f64(default_json, "double")?),
+ Codec::Utf8 | Codec::Utf8View | Codec::Uuid => {
+ AvroLiteral::String(expect_string(default_json, "string/uuid")?.to_string())
+ }
+ Codec::Binary => AvroLiteral::Bytes(parse_bytes_default(default_json, None)?),
+ Codec::Fixed(sz) => {
+ AvroLiteral::Bytes(parse_bytes_default(default_json, Some(*sz as usize))?)
+ }
+ Codec::Decimal(_, _, fixed_size) => {
+ AvroLiteral::Bytes(parse_bytes_default(default_json, *fixed_size)?)
+ }
+ Codec::Enum(symbols) => {
+ let s = expect_string(default_json, "enum")?;
+ if symbols.iter().any(|sym| sym == s) {
+ AvroLiteral::Enum(s.to_string())
+ } else {
+ return Err(ArrowError::SchemaError(format!(
+ "Default enum symbol {s:?} not found in reader enum symbols"
+ )));
+ }
+ }
+ Codec::Interval => AvroLiteral::Bytes(parse_bytes_default(default_json, Some(12))?),
+ Codec::List(item_dt) => match default_json {
+ Value::Array(items) => AvroLiteral::Array(
+ items
+ .iter()
+ .map(|v| item_dt.parse_default_literal(v))
+ .collect::<Result<_, _>>()?,
+ ),
+ _ => {
+ return Err(ArrowError::SchemaError(
+ "Default value must be a JSON array for Avro array type".to_string(),
+ ))
+ }
+ },
+ Codec::Map(val_dt) => match default_json {
+ Value::Object(map) => {
+ let mut out = IndexMap::with_capacity(map.len());
+ for (k, v) in map {
+ out.insert(k.clone(), val_dt.parse_default_literal(v)?);
+ }
+ AvroLiteral::Map(out)
+ }
+ _ => {
+ return Err(ArrowError::SchemaError(
+ "Default value must be a JSON object for Avro map type".to_string(),
+ ))
+ }
+ },
+ Codec::Struct(fields) => match default_json {
+ Value::Object(obj) => {
+ let mut out: IndexMap<String, AvroLiteral> =
+ IndexMap::with_capacity(fields.len());
+ for f in fields.as_ref() {
+ let name = f.name().to_string();
+ if let Some(sub) = obj.get(&name) {
+ // Explicit value provided in the record default object
+ let lit = f.data_type().parse_default_literal(sub)?;
+ out.insert(name, lit);
+ } else if let Some(default_json) =
+ f.data_type().metadata.get(AVRO_FIELD_DEFAULT_METADATA_KEY)
+ {
+ // Use the subfield's own stored default (validate and parse)
+ let v: Value = serde_json::from_str(default_json).map_err(|e| {
+ ArrowError::SchemaError(format!(
+ "Failed to parse stored subfield default JSON for '{}': {e}",
+ f.name(),
+ ))
+ })?;
+ let lit = f.data_type().parse_default_literal(&v)?;
+ out.insert(name, lit);
+ } else if f.data_type().nullability() == Some(Nullability::default()) {
Review Comment:
@nathaniel-d-ef Good callout. I see what you're getting at. I went ahead and cleaned this code up a bit more.
--
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 array/map/fixed schema resolution and default value support to arrow-avro codec [arrow-rs]
Posted by "nathaniel-d-ef (via GitHub)" <gi...@apache.org>.
nathaniel-d-ef commented on code in PR #8292:
URL: https://github.com/apache/arrow-rs/pull/8292#discussion_r2337360146
##########
arrow-avro/src/codec.rs:
##########
@@ -193,6 +200,224 @@ impl AvroDataType {
pub fn nullability(&self) -> Option<Nullability> {
self.nullability
}
+
+ #[inline]
+ fn parse_default_literal(&self, default_json: &Value) -> Result<AvroLiteral, ArrowError> {
+ fn expect_string<'v>(
+ default_json: &'v Value,
+ data_type: &str,
+ ) -> Result<&'v str, ArrowError> {
+ match default_json {
+ Value::String(s) => Ok(s.as_str()),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default value must be a JSON string for {data_type}"
+ ))),
+ }
+ }
+
+ fn parse_bytes_default(
+ default_json: &Value,
+ expected_len: Option<usize>,
+ ) -> Result<Vec<u8>, ArrowError> {
+ let s = expect_string(default_json, "bytes/fixed logical types")?;
+ let mut out = Vec::with_capacity(s.len());
+ for ch in s.chars() {
+ let cp = ch as u32;
+ if cp > 0xFF {
+ return Err(ArrowError::SchemaError(format!(
+ "Invalid codepoint U+{cp:04X} in bytes/fixed default; must be ≤ 0xFF"
+ )));
+ }
+ out.push(cp as u8);
+ }
+ if let Some(len) = expected_len {
+ if out.len() != len {
+ return Err(ArrowError::SchemaError(format!(
+ "Default length {} does not match expected fixed size {len}",
+ out.len(),
+ )));
+ }
+ }
+ Ok(out)
+ }
+
+ fn parse_json_i64(default_json: &Value, data_type: &str) -> Result<i64, ArrowError> {
+ match default_json {
+ Value::Number(n) => n.as_i64().ok_or_else(|| {
+ ArrowError::SchemaError(format!("Default {data_type} must be an integer"))
+ }),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default {data_type} must be a JSON integer"
+ ))),
+ }
+ }
+
+ fn parse_json_f64(default_json: &Value, data_type: &str) -> Result<f64, ArrowError> {
+ match default_json {
+ Value::Number(n) => n.as_f64().ok_or_else(|| {
+ ArrowError::SchemaError(format!("Default {data_type} must be a number"))
+ }),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default {data_type} must be a JSON number"
+ ))),
+ }
+ }
+
+ // Handle JSON nulls per-spec: allowed only for `null` type or unions with null FIRST
Review Comment:
I recall we're supporting first and second positions elsewhere in this crate re: impala - can you clarify why it's not an issue that we're diverging from that here?
##########
arrow-avro/src/codec.rs:
##########
@@ -193,6 +200,224 @@ impl AvroDataType {
pub fn nullability(&self) -> Option<Nullability> {
self.nullability
}
+
+ #[inline]
+ fn parse_default_literal(&self, default_json: &Value) -> Result<AvroLiteral, ArrowError> {
+ fn expect_string<'v>(
+ default_json: &'v Value,
+ data_type: &str,
+ ) -> Result<&'v str, ArrowError> {
+ match default_json {
+ Value::String(s) => Ok(s.as_str()),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default value must be a JSON string for {data_type}"
+ ))),
+ }
+ }
+
+ fn parse_bytes_default(
+ default_json: &Value,
+ expected_len: Option<usize>,
+ ) -> Result<Vec<u8>, ArrowError> {
+ let s = expect_string(default_json, "bytes/fixed logical types")?;
+ let mut out = Vec::with_capacity(s.len());
+ for ch in s.chars() {
+ let cp = ch as u32;
+ if cp > 0xFF {
+ return Err(ArrowError::SchemaError(format!(
+ "Invalid codepoint U+{cp:04X} in bytes/fixed default; must be ≤ 0xFF"
+ )));
+ }
+ out.push(cp as u8);
+ }
+ if let Some(len) = expected_len {
+ if out.len() != len {
+ return Err(ArrowError::SchemaError(format!(
+ "Default length {} does not match expected fixed size {len}",
+ out.len(),
+ )));
+ }
+ }
+ Ok(out)
+ }
+
+ fn parse_json_i64(default_json: &Value, data_type: &str) -> Result<i64, ArrowError> {
+ match default_json {
+ Value::Number(n) => n.as_i64().ok_or_else(|| {
+ ArrowError::SchemaError(format!("Default {data_type} must be an integer"))
+ }),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default {data_type} must be a JSON integer"
+ ))),
+ }
+ }
+
+ fn parse_json_f64(default_json: &Value, data_type: &str) -> Result<f64, ArrowError> {
+ match default_json {
+ Value::Number(n) => n.as_f64().ok_or_else(|| {
+ ArrowError::SchemaError(format!("Default {data_type} must be a number"))
+ }),
+ _ => Err(ArrowError::SchemaError(format!(
+ "Default {data_type} must be a JSON number"
+ ))),
+ }
+ }
+
+ // Handle JSON nulls per-spec: allowed only for `null` type or unions with null FIRST
+ if default_json.is_null() {
+ return match self.codec() {
+ Codec::Null => Ok(AvroLiteral::Null),
+ _ if self.nullability() == Some(Nullability::NullFirst) => Ok(AvroLiteral::Null),
+ _ => Err(ArrowError::SchemaError(
+ "JSON null default is only valid for `null` type or for a union whose first branch is `null`"
+ .to_string(),
+ )),
+ };
+ }
+ let lit = match self.codec() {
+ Codec::Null => {
+ return Err(ArrowError::SchemaError(
+ "Default for `null` type must be JSON null".to_string(),
+ ))
+ }
+ Codec::Boolean => match default_json {
+ Value::Bool(b) => AvroLiteral::Boolean(*b),
+ _ => {
+ return Err(ArrowError::SchemaError(
+ "Boolean default must be a JSON boolean".to_string(),
+ ))
+ }
+ },
+ Codec::Int32 | Codec::Date32 | Codec::TimeMillis => {
+ let i = parse_json_i64(default_json, "int")?;
+ if i < i32::MIN as i64 || i > i32::MAX as i64 {
+ return Err(ArrowError::SchemaError(format!(
+ "Default int {i} out of i32 range"
+ )));
+ }
+ AvroLiteral::Int(i as i32)
+ }
+ Codec::Int64
+ | Codec::TimeMicros
+ | Codec::TimestampMillis(_)
+ | Codec::TimestampMicros(_) => AvroLiteral::Long(parse_json_i64(default_json, "long")?),
+ Codec::Float32 => {
+ let f = parse_json_f64(default_json, "float")?;
+ if !f.is_finite() || f < f32::MIN as f64 || f > f32::MAX as f64 {
+ return Err(ArrowError::SchemaError(format!(
+ "Default float {f} out of f32 range or not finite"
+ )));
+ }
+ AvroLiteral::Float(f as f32)
+ }
+ Codec::Float64 => AvroLiteral::Double(parse_json_f64(default_json, "double")?),
+ Codec::Utf8 | Codec::Utf8View | Codec::Uuid => {
+ AvroLiteral::String(expect_string(default_json, "string/uuid")?.to_string())
+ }
+ Codec::Binary => AvroLiteral::Bytes(parse_bytes_default(default_json, None)?),
+ Codec::Fixed(sz) => {
+ AvroLiteral::Bytes(parse_bytes_default(default_json, Some(*sz as usize))?)
+ }
+ Codec::Decimal(_, _, fixed_size) => {
+ AvroLiteral::Bytes(parse_bytes_default(default_json, *fixed_size)?)
+ }
+ Codec::Enum(symbols) => {
+ let s = expect_string(default_json, "enum")?;
+ if symbols.iter().any(|sym| sym == s) {
+ AvroLiteral::Enum(s.to_string())
+ } else {
+ return Err(ArrowError::SchemaError(format!(
+ "Default enum symbol {s:?} not found in reader enum symbols"
+ )));
+ }
+ }
+ Codec::Interval => AvroLiteral::Bytes(parse_bytes_default(default_json, Some(12))?),
+ Codec::List(item_dt) => match default_json {
+ Value::Array(items) => AvroLiteral::Array(
+ items
+ .iter()
+ .map(|v| item_dt.parse_default_literal(v))
+ .collect::<Result<_, _>>()?,
+ ),
+ _ => {
+ return Err(ArrowError::SchemaError(
+ "Default value must be a JSON array for Avro array type".to_string(),
+ ))
+ }
+ },
+ Codec::Map(val_dt) => match default_json {
+ Value::Object(map) => {
+ let mut out = IndexMap::with_capacity(map.len());
+ for (k, v) in map {
+ out.insert(k.clone(), val_dt.parse_default_literal(v)?);
+ }
+ AvroLiteral::Map(out)
+ }
+ _ => {
+ return Err(ArrowError::SchemaError(
+ "Default value must be a JSON object for Avro map type".to_string(),
+ ))
+ }
+ },
+ Codec::Struct(fields) => match default_json {
+ Value::Object(obj) => {
+ let mut out: IndexMap<String, AvroLiteral> =
+ IndexMap::with_capacity(fields.len());
+ for f in fields.as_ref() {
+ let name = f.name().to_string();
+ if let Some(sub) = obj.get(&name) {
+ // Explicit value provided in the record default object
+ let lit = f.data_type().parse_default_literal(sub)?;
+ out.insert(name, lit);
+ } else if let Some(default_json) =
+ f.data_type().metadata.get(AVRO_FIELD_DEFAULT_METADATA_KEY)
+ {
+ // Use the subfield's own stored default (validate and parse)
+ let v: Value = serde_json::from_str(default_json).map_err(|e| {
+ ArrowError::SchemaError(format!(
+ "Failed to parse stored subfield default JSON for '{}': {e}",
+ f.name(),
+ ))
+ })?;
+ let lit = f.data_type().parse_default_literal(&v)?;
+ out.insert(name, lit);
+ } else if f.data_type().nullability() == Some(Nullability::default()) {
Review Comment:
This is probably a cheap check - worth it to move it up in the chain before the others?
--
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 array/map/fixed schema resolution and default value support to arrow-avro codec [arrow-rs]
Posted by "jecsand838 (via GitHub)" <gi...@apache.org>.
jecsand838 commented on code in PR #8292:
URL: https://github.com/apache/arrow-rs/pull/8292#discussion_r2338011238
##########
arrow-avro/src/codec.rs:
##########
@@ -1712,4 +2032,440 @@ mod tests {
panic!("Top-level schema is not a struct");
}
}
+
+ fn json_string(s: &str) -> Value {
+ Value::String(s.to_string())
+ }
+
+ fn assert_default_stored(dt: &AvroDataType, default_json: &Value) {
+ let stored = dt
+ .metadata
+ .get(AVRO_FIELD_DEFAULT_METADATA_KEY)
+ .cloned()
+ .unwrap_or_default();
+ let expected = serde_json::to_string(default_json).unwrap();
+ assert_eq!(stored, expected, "stored default metadata should match");
+ }
+
+ #[test]
+ fn test_validate_and_store_default_null_and_nullability_rules() {
+ let mut dt_null = AvroDataType::new(Codec::Null, HashMap::new(), None);
+ let lit = dt_null.parse_and_store_default(&Value::Null).unwrap();
+ assert_eq!(lit, AvroLiteral::Null);
+ assert_default_stored(&dt_null, &Value::Null);
+ let mut dt_int = AvroDataType::new(Codec::Int32, HashMap::new(), None);
+ let err = dt_int.parse_and_store_default(&Value::Null).unwrap_err();
+ assert!(
+ err.to_string()
+ .contains("JSON null default is only valid for `null` type"),
+ "unexpected error: {err}"
+ );
+ let mut dt_int_nf =
+ AvroDataType::new(Codec::Int32, HashMap::new(), Some(Nullability::NullFirst));
+ let lit2 = dt_int_nf.parse_and_store_default(&Value::Null).unwrap();
+ assert_eq!(lit2, AvroLiteral::Null);
+ assert_default_stored(&dt_int_nf, &Value::Null);
+ let mut dt_int_ns =
+ AvroDataType::new(Codec::Int32, HashMap::new(), Some(Nullability::NullSecond));
+ let err2 = dt_int_ns.parse_and_store_default(&Value::Null).unwrap_err();
+ assert!(
+ err2.to_string()
+ .contains("JSON null default is only valid for `null` type"),
+ "unexpected error: {err2}"
+ );
+ }
+
+ #[test]
+ fn test_validate_and_store_default_primitives_and_temporal() {
+ let mut dt_bool = AvroDataType::new(Codec::Boolean, HashMap::new(), None);
+ let lit = dt_bool.parse_and_store_default(&Value::Bool(true)).unwrap();
+ assert_eq!(lit, AvroLiteral::Boolean(true));
+ assert_default_stored(&dt_bool, &Value::Bool(true));
+ let mut dt_i32 = AvroDataType::new(Codec::Int32, HashMap::new(), None);
+ let lit = dt_i32
+ .parse_and_store_default(&serde_json::json!(123))
+ .unwrap();
+ assert_eq!(lit, AvroLiteral::Int(123));
+ assert_default_stored(&dt_i32, &serde_json::json!(123));
+ let err = dt_i32
+ .parse_and_store_default(&serde_json::json!(i64::from(i32::MAX) + 1))
+ .unwrap_err();
+ assert!(format!("{err}").contains("out of i32 range"));
+ let mut dt_i64 = AvroDataType::new(Codec::Int64, HashMap::new(), None);
+ let lit = dt_i64
+ .parse_and_store_default(&serde_json::json!(1234567890))
+ .unwrap();
+ assert_eq!(lit, AvroLiteral::Long(1234567890));
+ assert_default_stored(&dt_i64, &serde_json::json!(1234567890));
+ let mut dt_f32 = AvroDataType::new(Codec::Float32, HashMap::new(), None);
+ let lit = dt_f32
+ .parse_and_store_default(&serde_json::json!(1.25))
+ .unwrap();
+ assert_eq!(lit, AvroLiteral::Float(1.25));
+ assert_default_stored(&dt_f32, &serde_json::json!(1.25));
+ let err = dt_f32
+ .parse_and_store_default(&serde_json::json!(1e39))
+ .unwrap_err();
+ assert!(format!("{err}").contains("out of f32 range"));
+ let mut dt_f64 = AvroDataType::new(Codec::Float64, HashMap::new(), None);
+ let lit = dt_f64
+ .parse_and_store_default(&serde_json::json!(std::f64::consts::PI))
+ .unwrap();
+ assert_eq!(lit, AvroLiteral::Double(std::f64::consts::PI));
+ assert_default_stored(&dt_f64, &serde_json::json!(std::f64::consts::PI));
+ let mut dt_str = AvroDataType::new(Codec::Utf8, HashMap::new(), None);
+ let l = dt_str
+ .parse_and_store_default(&json_string("hello"))
+ .unwrap();
+ assert_eq!(l, AvroLiteral::String("hello".into()));
+ assert_default_stored(&dt_str, &json_string("hello"));
+ let mut dt_strv = AvroDataType::new(Codec::Utf8View, HashMap::new(), None);
+ let l = dt_strv
+ .parse_and_store_default(&json_string("view"))
+ .unwrap();
+ assert_eq!(l, AvroLiteral::String("view".into()));
+ assert_default_stored(&dt_strv, &json_string("view"));
+ let mut dt_uuid = AvroDataType::new(Codec::Uuid, HashMap::new(), None);
+ let l = dt_uuid
+ .parse_and_store_default(&json_string("00000000-0000-0000-0000-000000000000"))
+ .unwrap();
+ assert_eq!(
+ l,
+ AvroLiteral::String("00000000-0000-0000-0000-000000000000".into())
+ );
+ let mut dt_bin = AvroDataType::new(Codec::Binary, HashMap::new(), None);
+ let l = dt_bin.parse_and_store_default(&json_string("ABC")).unwrap();
+ assert_eq!(l, AvroLiteral::Bytes(vec![65, 66, 67]));
+ let err = dt_bin
+ .parse_and_store_default(&json_string("€")) // U+20AC
+ .unwrap_err();
+ assert!(format!("{err}").contains("Invalid codepoint"));
+ let mut dt_date = AvroDataType::new(Codec::Date32, HashMap::new(), None);
+ let ld = dt_date
+ .parse_and_store_default(&serde_json::json!(1))
+ .unwrap();
+ assert_eq!(ld, AvroLiteral::Int(1));
+ let mut dt_tmill = AvroDataType::new(Codec::TimeMillis, HashMap::new(), None);
+ let lt = dt_tmill
+ .parse_and_store_default(&serde_json::json!(86_400_000))
+ .unwrap();
+ assert_eq!(lt, AvroLiteral::Int(86_400_000));
+ let mut dt_tmicros = AvroDataType::new(Codec::TimeMicros, HashMap::new(), None);
+ let ltm = dt_tmicros
+ .parse_and_store_default(&serde_json::json!(1_000_000))
+ .unwrap();
+ assert_eq!(ltm, AvroLiteral::Long(1_000_000));
+ let mut dt_ts_milli = AvroDataType::new(Codec::TimestampMillis(true), HashMap::new(), None);
+ let l1 = dt_ts_milli
+ .parse_and_store_default(&serde_json::json!(123))
+ .unwrap();
+ assert_eq!(l1, AvroLiteral::Long(123));
+ let mut dt_ts_micro =
+ AvroDataType::new(Codec::TimestampMicros(false), HashMap::new(), None);
+ let l2 = dt_ts_micro
+ .parse_and_store_default(&serde_json::json!(456))
+ .unwrap();
+ assert_eq!(l2, AvroLiteral::Long(456));
+ }
+
+ #[test]
+ fn test_validate_and_store_default_fixed_decimal_interval() {
+ let mut dt_fixed = AvroDataType::new(Codec::Fixed(4), HashMap::new(), None);
+ let l = dt_fixed
+ .parse_and_store_default(&json_string("WXYZ"))
+ .unwrap();
+ assert_eq!(l, AvroLiteral::Bytes(vec![87, 88, 89, 90]));
+ let err = dt_fixed
+ .parse_and_store_default(&json_string("TOO LONG"))
+ .unwrap_err();
+ assert!(format!("{err}").contains("Default length"));
Review Comment:
@alamb That's a good callout! I just pushed up a commit with the other instances of this changed.
--
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