This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 6171ea0e fix(spec): decode _GLOBAL_INDEX fields by name, not by
position (#843)
6171ea0e is described below
commit 6171ea0e32557c55b98eb3f2c5ba9edbaa7b3cbd
Author: jackylee <[email protected]>
AuthorDate: Thu Sep 17 13:50:32 2026 +0800
fix(spec): decode _GLOBAL_INDEX fields by name, not by position (#843)
---
.../src/spec/avro/index_manifest_entry_decode.rs | 116 +++++++++++----------
crates/paimon/src/spec/index_manifest.rs | 56 ++++++++++
2 files changed, 118 insertions(+), 54 deletions(-)
diff --git a/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs
b/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs
index 5cf75ab5..7c94ddf6 100644
--- a/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs
+++ b/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs
@@ -191,62 +191,37 @@ fn decode_nullable_global_index(
return Ok(None);
}
}
- let row_range_start = cursor.read_long()?;
- let row_range_end = cursor.read_long()?;
- let index_field_id = cursor.read_int()?;
-
- // _EXTRA_FIELD_IDS: nullable array of int
- let extra_field_ids = {
- let u_idx = cursor.read_union_index()?;
- if u_idx == 0 {
- None
- } else {
- let mut ids = Vec::new();
- loop {
- let count = cursor.read_long()?;
- if count == 0 {
- break;
- }
- let count = if count < 0 {
- cursor.skip_long()?;
- neg_count_to_usize(count)?
- } else {
- count as usize
- };
- for _ in 0..count {
- ids.push(cursor.read_int()?);
- }
+ // Walk the writer's own field list, as the deletion-vector record above
does.
+ // Nothing in a manifest says how many fields this record has: Java tried a
+ // runtime `getFieldCount() <= 5` check when `_SOURCE_META` was added
(#8549),
+ // replaced it with an entry-serializer version (#8952), then reverted to
+ // `GlobalIndexMeta.SCHEMA.getFieldCount()` (#9004) and deleted the
versioned
+ // serializer entirely (#9039). Shape compatibility is therefore delegated
to the
+ // file format's schema resolution, which is exactly what positional
decoding
+ // cannot do — the writer's schema is the only description of the record.
+ let record = extract_record_schema(schema).ok_or_else(||
crate::Error::UnexpectedError {
+ message: "global index metadata must be an Avro record".into(),
+ source: None,
+ })?;
+ let mut row_range_start = 0;
+ let mut row_range_end = 0;
+ let mut index_field_id = 0;
+ let mut extra_field_ids = None;
+ let mut index_meta = None;
+ let mut source_meta = None;
+ for field in &record.fields {
+ match field.name.as_str() {
+ "_ROW_RANGE_START" => row_range_start = read_long_field(cursor,
field.nullable)?,
+ "_ROW_RANGE_END" => row_range_end = read_long_field(cursor,
field.nullable)?,
+ "_INDEX_FIELD_ID" => index_field_id = read_int_field(cursor,
field.nullable)?,
+ "_EXTRA_FIELD_IDS" => {
+ extra_field_ids = decode_nullable_int_array(cursor,
field.nullable)?
}
- Some(ids)
- }
- };
-
- // _INDEX_META: nullable bytes
- let index_meta = {
- let u_idx = cursor.read_union_index()?;
- if u_idx == 0 {
- None
- } else {
- Some(cursor.read_bytes()?.to_vec())
- }
- };
-
- // _SOURCE_META: nullable bytes — only present in >= #8549 writer schemas.
- // Guard on the writer's nested field list so a legacy 5-field
_GLOBAL_INDEX
- // record does not misalign the cursor into the next record.
- let has_source_meta = extract_record_schema(schema)
- .map(|s| s.fields.iter().any(|f| f.name == "_SOURCE_META"))
- .unwrap_or(false);
- let source_meta = if has_source_meta {
- let u_idx = cursor.read_union_index()?;
- if u_idx == 0 {
- None
- } else {
- Some(cursor.read_bytes()?.to_vec())
+ "_INDEX_META" => index_meta = read_optional_bytes(cursor,
field.nullable)?,
+ "_SOURCE_META" => source_meta = read_optional_bytes(cursor,
field.nullable)?,
+ _ => skip_nullable_field(cursor, &field.schema, field.nullable)?,
}
- } else {
- None
- };
+ }
Ok(Some(GlobalIndexMeta {
row_range_start,
@@ -257,3 +232,36 @@ fn decode_nullable_global_index(
source_meta,
}))
}
+
+fn read_optional_bytes(cursor: &mut AvroCursor, nullable: bool) ->
crate::Result<Option<Vec<u8>>> {
+ if nullable && cursor.read_union_index()? == 0 {
+ return Ok(None);
+ }
+ Ok(Some(cursor.read_bytes()?.to_vec()))
+}
+
+fn decode_nullable_int_array(
+ cursor: &mut AvroCursor,
+ nullable: bool,
+) -> crate::Result<Option<Vec<i32>>> {
+ if nullable && cursor.read_union_index()? == 0 {
+ return Ok(None);
+ }
+ let mut ids = Vec::new();
+ loop {
+ let count = cursor.read_long()?;
+ if count == 0 {
+ break;
+ }
+ let count = if count < 0 {
+ cursor.skip_long()?;
+ neg_count_to_usize(count)?
+ } else {
+ count as usize
+ };
+ for _ in 0..count {
+ ids.push(cursor.read_int()?);
+ }
+ }
+ Ok(Some(ids))
+}
diff --git a/crates/paimon/src/spec/index_manifest.rs
b/crates/paimon/src/spec/index_manifest.rs
index 7b767813..967bab8c 100644
--- a/crates/paimon/src/spec/index_manifest.rs
+++ b/crates/paimon/src/spec/index_manifest.rs
@@ -309,6 +309,62 @@ mod tests {
assert_eq!(IndexManifest::read_from_bytes(&bytes).unwrap(),
vec![entry]);
}
+ #[test]
+ fn test_global_index_records_allow_reordered_and_unknown_fields() {
+ // Sibling of the deletion-vector test above, for the other nested
record in
+ // the same entry. `_GLOBAL_INDEX` has already grown once
(`_SOURCE_META`), and
+ // Java documents the record as appendable, so the decoder must not
depend on
+ // field order.
+ let mut schema: serde_json::Value =
+ serde_json::from_str(INDEX_MANIFEST_ENTRY_SCHEMA).unwrap();
+ assert_eq!(schema["fields"][10]["name"], "_GLOBAL_INDEX");
+ let fields = schema["fields"][10]["type"][1]["fields"]
+ .as_array_mut()
+ .unwrap();
+ assert_eq!(
+ fields.len(),
+ 6,
+ "a seventh field means someone must confirm the walk handles it"
+ );
+ fields.reverse();
+ fields.insert(
+ 1,
+ serde_json::json!({"name": "future", "type": ["null", "bytes"],
"default": null}),
+ );
+
+ let entry = global_index_entry(Some(vec![4, 5, 6]));
+ let schema = Schema::parse_str(&schema.to_string()).unwrap();
+ let original =
+ crate::spec::to_avro_bytes(INDEX_MANIFEST_ENTRY_SCHEMA,
std::slice::from_ref(&entry))
+ .unwrap();
+ let mut value = apache_avro::Reader::new(original.as_slice())
+ .unwrap()
+ .next()
+ .unwrap()
+ .unwrap();
+ let Value::Record(fields) = &mut value else {
+ panic!("record");
+ };
+ let (_, global_index) = fields
+ .iter_mut()
+ .find(|(name, _)| name == "_GLOBAL_INDEX")
+ .unwrap();
+ let Value::Union(_, global_index) = global_index else {
+ panic!("nullable global index");
+ };
+ let Value::Record(fields) = global_index.as_mut() else {
+ panic!("global index record");
+ };
+ fields.push((
+ "future".into(),
+ Value::Union(1, Box::new(Value::Bytes(vec![9, 9]))),
+ ));
+ let mut writer = apache_avro::Writer::new(&schema, Vec::new());
+ writer.append(value.resolve(&schema).unwrap()).unwrap();
+ let bytes = writer.into_inner().unwrap();
+ assert_eq!(IndexManifest::read_from_bytes(&bytes).unwrap(),
vec![entry]);
+ }
+
#[test]
fn test_read_index_manifest_file() {
let workdir =
