This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 1e285c6a fix(reader): read Avro enum columns and V2 Avro data log
blocks (#803)
1e285c6a is described below
commit 1e285c6a96d139ae1828f691e0591f99842d12b4
Author: Y Ethan Guo <[email protected]>
AuthorDate: Tue Sep 29 17:35:41 2026 -0700
fix(reader): read Avro enum columns and V2 Avro data log blocks (#803)
Accept V2 Avro data blocks, which writers before table version 6 leave in
tables upgraded to 6 until compaction; V2 delete blocks are Kryo-encoded and
stay refused with an error naming the block and version. Unpack arrow-avro's
enum dictionaries in the Avro block decoder so log rows merge with the Utf8
column the table schema and base files use.
---
crates/core/src/file_group/log_file/avro.rs | 131 ++++++++++++-
crates/core/src/file_group/log_file/content.rs | 90 +++++++--
crates/core/tests/log_block_compat_tests.rs | 250 +++++++++++++++++++++++++
3 files changed, 453 insertions(+), 18 deletions(-)
diff --git a/crates/core/src/file_group/log_file/avro.rs
b/crates/core/src/file_group/log_file/avro.rs
index d4e0f8c8..834215f9 100644
--- a/crates/core/src/file_group/log_file/avro.rs
+++ b/crates/core/src/file_group/log_file/avro.rs
@@ -176,9 +176,19 @@ impl AvroBlockDecoder {
///
/// Taken from the decoder rather than converted from the Avro JSON,
because
/// `avro_to_arrow` does not handle named-type references and the metadata
- /// table's schema uses them.
+ /// table's schema uses them. Dictionaries are unpacked to match what
+ /// [`flush`](Self::flush) emits.
pub fn schema(&self) -> SchemaRef {
- self.decoder.schema()
+ let schema = self.decoder.schema();
+ if !schema
+ .fields()
+ .iter()
+ .any(|f| has_dictionary(f.data_type()))
+ {
+ return schema;
+ }
+ let fields: Vec<Field> = schema.fields().iter().map(|f|
unpacked_field(f)).collect();
+ Arc::new(Schema::new_with_metadata(fields, schema.metadata().clone()))
}
/// Project each decoded batch to `schema` instead of resolving during the
@@ -234,7 +244,11 @@ impl AvroBlockDecoder {
let batch = self.decoder.flush().map_err(|e| {
CoreError::LogBlockError(format!("Failed to flush decoded records:
{e}"))
})?;
- let batch = batch.map(normalize_utc_timestamps).transpose()?;
+ let batch = batch
+ .map(normalize_utc_timestamps)
+ .transpose()?
+ .map(unpack_dictionaries)
+ .transpose()?;
match (batch, self.rewrite_to.as_ref()) {
(Some(batch), Some(target)) => {
crate::schema::batch_evolution::project_batch_to_schema(&batch,
target).map(Some)
@@ -300,6 +314,71 @@ fn normalize_utc_timestamps(batch: RecordBatch) ->
Result<RecordBatch> {
RecordBatch::try_new(Arc::new(Schema::new(fields)),
columns).map_err(CoreError::ArrowError)
}
+/// Unpack every dictionary column, at any depth, to its value type.
+///
+/// `arrow-avro` decodes an Avro `enum` as `Dictionary(Int32, Utf8)`, while the
+/// table schema (`avro_to_arrow::schema`) and a parquet base file both hold
the
+/// same column as `Utf8`. The merge reconciles log rows to the table schema,
+/// which refuses a dictionary leaf, so the log batch is unpacked here.
+fn unpack_dictionaries(batch: RecordBatch) -> Result<RecordBatch> {
+ let schema = batch.schema();
+ if !schema
+ .fields()
+ .iter()
+ .any(|f| has_dictionary(f.data_type()))
+ {
+ return Ok(batch);
+ }
+
+ let mut fields: Vec<Field> = Vec::with_capacity(batch.num_columns());
+ let mut columns: Vec<ArrayRef> = Vec::with_capacity(batch.num_columns());
+ for (field, column) in schema.fields().iter().zip(batch.columns()) {
+ if has_dictionary(field.data_type()) {
+ let target = unpacked_field(field);
+ columns.push(cast(column,
target.data_type()).map_err(CoreError::ArrowError)?);
+ fields.push(target);
+ } else {
+ fields.push(field.as_ref().clone());
+ columns.push(column.clone());
+ }
+ }
+ RecordBatch::try_new(
+ Arc::new(Schema::new_with_metadata(fields, schema.metadata().clone())),
+ columns,
+ )
+ .map_err(CoreError::ArrowError)
+}
+
+fn has_dictionary(data_type: &DataType) -> bool {
+ match data_type {
+ DataType::Dictionary(_, _) => true,
+ DataType::Struct(fields) => fields.iter().any(|f|
has_dictionary(f.data_type())),
+ DataType::List(f)
+ | DataType::LargeList(f)
+ | DataType::FixedSizeList(f, _)
+ | DataType::Map(f, _) => has_dictionary(f.data_type()),
+ _ => false,
+ }
+}
+
+fn unpacked_field(field: &Field) -> Field {
+ field.clone().with_data_type(unpacked(field.data_type()))
+}
+
+fn unpacked(data_type: &DataType) -> DataType {
+ match data_type {
+ DataType::Dictionary(_, value_type) => unpacked(value_type),
+ DataType::Struct(fields) => {
+ DataType::Struct(fields.iter().map(|f|
unpacked_field(f)).collect())
+ }
+ DataType::List(f) => DataType::List(Arc::new(unpacked_field(f))),
+ DataType::LargeList(f) =>
DataType::LargeList(Arc::new(unpacked_field(f))),
+ DataType::FixedSizeList(f, n) =>
DataType::FixedSizeList(Arc::new(unpacked_field(f)), *n),
+ DataType::Map(f, sorted) => DataType::Map(Arc::new(unpacked_field(f)),
*sorted),
+ other => other.clone(),
+ }
+}
+
#[cfg(test)]
mod multi_batch_tests {
use super::*;
@@ -494,6 +573,52 @@ mod tests {
);
}
+ /// An Avro enum decodes as a string dictionary; it leaves the decoder as
+ /// plain strings at every depth, matching the table schema's `Utf8`.
+ #[test]
+ fn test_enum_decodes_to_utf8_at_every_depth() {
+ use apache_avro::types::Value;
+ use arrow_array::cast::AsArray;
+ use arrow_schema::DataType;
+
+ let schema = r#"{"type":"record","name":"r","fields":[
+
{"name":"e","type":{"type":"enum","name":"Kind","symbols":["A","B"]}},
+ {"name":"n","type":["null","Kind"],"default":null},
+
{"name":"s","type":{"type":"record","name":"s","fields":[{"name":"k","type":"Kind"}]}},
+ {"name":"a","type":{"type":"array","items":"Kind"}},
+ {"name":"m","type":{"type":"map","values":"Kind"}}]}"#;
+ let avro_schema = apache_avro::Schema::parse_str(schema).unwrap();
+ let kind = |i: u32| Value::Enum(i, ["A", "B"][i as usize].to_string());
+ let record = Value::Record(vec![
+ ("e".into(), kind(1)),
+ ("n".into(), Value::Union(0, Box::new(Value::Null))),
+ ("s".into(), Value::Record(vec![("k".into(), kind(0))])),
+ ("a".into(), Value::Array(vec![kind(0), kind(1)])),
+ ("m".into(), Value::Map([("x".to_string(), kind(1))].into())),
+ ]);
+ let body = apache_avro::to_avro_datum(&avro_schema, record).unwrap();
+
+ let mut decoder =
+ AvroBlockDecoder::try_new_with_reader(schema, Some(schema),
1024).unwrap();
+ decoder.decode(&body).unwrap();
+ let declared = decoder.schema();
+ let batch = decoder.flush().unwrap().expect("a batch");
+ assert_eq!(declared.fields(), batch.schema().fields());
+
+ // `as_string` panics on a dictionary column, so each read below also
+ // pins the type.
+ assert_eq!(batch.column(0).as_string::<i32>().value(0), "B");
+ assert_eq!(batch.column(1).data_type(), &DataType::Utf8);
+ assert!(batch.column(1).is_null(0));
+ let s = batch.column(2).as_struct();
+ assert_eq!(s.column(0).as_string::<i32>().value(0), "A");
+ let a = batch.column(3).as_list::<i32>().value(0);
+ let a = a.as_string::<i32>();
+ assert_eq!((a.value(0), a.value(1)), ("A", "B"));
+ let m = batch.column(4).as_map();
+ assert_eq!(m.values().as_string::<i32>().value(0), "B");
+ }
+
/// Without a reader schema the block reads at the schema it was written
/// with, which is what a partial-update block relies on.
#[test]
diff --git a/crates/core/src/file_group/log_file/content.rs
b/crates/core/src/file_group/log_file/content.rs
index 63669db9..553e18a7 100644
--- a/crates/core/src/file_group/log_file/content.rs
+++ b/crates/core/src/file_group/log_file/content.rs
@@ -264,19 +264,24 @@ impl Decoder {
}
}
- /// Validate the log block version (first 4 bytes of block content).
+ /// Read the log block version (first 4 bytes of block content) and check
it
+ /// is one of `supported`.
///
/// This is NOT the same as [`LogFormatVersion`] (read from the file
header).
- /// Modern Hudi tables use [`LogBlockVersion::V3`].
- fn validate_log_block_version(mut reader: impl Read) -> Result<()> {
+ /// Modern Hudi tables use [`LogBlockVersion::V3`]; writers before table
+ /// version 6 (Hudi 0.13 and earlier) use [`LogBlockVersion::V2`], and a
table
+ /// upgraded from them keeps those blocks until compaction rewrites them.
+ fn validate_log_block_version(
+ mut reader: impl Read,
+ supported: &[LogBlockVersion],
+ block: &str,
+ ) -> Result<()> {
let mut version_buf = [0u8; 4];
reader.read_exact(&mut version_buf)?;
let version = LogBlockVersion::try_from(version_buf)?;
- if version != LogBlockVersion::V3 {
+ if !supported.contains(&version) {
return Err(CoreError::LogBlockError(format!(
- "Only support log block version {} but got: {:?}",
- LogBlockVersion::V3 as u32,
- version
+ "Unsupported {block} block version {version:?}; supported:
{supported:?}"
)));
}
Ok(())
@@ -435,7 +440,14 @@ impl Decoder {
mut reader: impl Read,
header: &HashMap<BlockMetadataKey, String>,
) -> Result<RecordBatches> {
- Decoder::validate_log_block_version(&mut reader)?;
+ // V2 and V3 Avro data blocks share one layout (version, record count,
+ // records): Hudi's HoodieAvroDataBlockVersion reports a record count
for
+ // every version.
+ Decoder::validate_log_block_version(
+ &mut reader,
+ &[LogBlockVersion::V2, LogBlockVersion::V3],
+ "Avro data",
+ )?;
let writer_schema_json =
header.get(&BlockMetadataKey::Schema).ok_or_else(|| {
CoreError::LogBlockError("Schema not found in block
header".to_string())
@@ -511,7 +523,9 @@ impl Decoder {
mut reader: impl Read,
header: &HashMap<BlockMetadataKey, String>,
) -> Result<RecordBatches> {
- Decoder::validate_log_block_version(&mut reader)?;
+ // A V2 delete block holds Kryo-serialized DeleteRecord objects rather
than
+ // the Avro HoodieDeleteRecordList of V3, which this reader cannot
decode.
+ Decoder::validate_log_block_version(&mut reader,
&[LogBlockVersion::V3], "delete")?;
let mut datum_len = [0u8; 4];
reader.read_exact(&mut datum_len)?;
@@ -898,24 +912,70 @@ mod tests {
);
}
- /// The block version is the first four bytes of a block's content, and
only
- /// V3 is understood. A block at another version is refused rather than
- /// decoded as if it were V3, which would misread every byte after it.
+ /// The block version is the first four bytes of a block's content. An Avro
+ /// data block is understood at V2 and V3; any other version is refused
rather
+ /// than decoded as if it were V3, which would misread every byte after it.
#[test]
- fn test_decode_content_rejects_a_non_v3_block_version() {
+ fn test_decode_content_rejects_an_unsupported_block_version() {
let mut buf = Vec::new();
buf.extend_from_slice(&1u32.to_be_bytes());
buf.extend_from_slice(&0u32.to_be_bytes());
let header = HashMap::from([(BlockMetadataKey::Schema,
"{}".to_string())]);
let err = Decoder::new(Arc::new(HudiConfigs::empty()))
.decode_avro_record_content(buf.as_slice(), &header)
- .expect_err("a non-V3 block must be refused");
+ .expect_err("a V1 block must be refused");
assert!(
- err.to_string().contains("log block version"),
+ err.to_string().contains("Avro data block version V1"),
"the error must name the version, got: {err}"
);
}
+ /// A table version 5 writer stamps its Avro data blocks V2, with the same
+ /// layout as V3 (version, record count, records), so they decode the same.
+ #[test]
+ fn test_decode_avro_content_at_block_version_2() -> Result<()> {
+ let schema_str = r#"{"type": "record", "name": "TestRecord",
+ "fields": [{"name": "id", "type": "long"}]}"#;
+ let writer_schema = apache_avro::Schema::parse_str(schema_str)?;
+ let mut buf = Vec::new();
+ buf.extend_from_slice(&2u32.to_be_bytes());
+ buf.extend_from_slice(&2u32.to_be_bytes());
+ for id in [7i64, 8] {
+ let mut record = AvroRecord::new(&writer_schema).unwrap();
+ record.put("id", id);
+ let bytes = to_avro_datum(&writer_schema, record)?;
+ buf.extend_from_slice(&(bytes.len() as u32).to_be_bytes());
+ buf.extend_from_slice(&bytes);
+ }
+ let header = HashMap::from([(BlockMetadataKey::Schema,
schema_str.to_string())]);
+ let batches = Decoder::new(Arc::new(HudiConfigs::empty()))
+ .decode_avro_record_content(Cursor::new(buf), &header)?;
+ assert_eq!(batches.num_data_rows(), 2);
+ let ids = batches.data_batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!((ids.value(0), ids.value(1)), (7, 8));
+ Ok(())
+ }
+
+ /// A V2 delete block holds Kryo-serialized records, not the Avro list of
V3,
+ /// so it is refused by name rather than misread as Avro.
+ #[test]
+ fn test_decode_delete_content_rejects_block_version_2() {
+ let mut buf = Vec::new();
+ buf.extend_from_slice(&2u32.to_be_bytes());
+ buf.extend_from_slice(&0u32.to_be_bytes());
+ let err = Decoder::new(Arc::new(HudiConfigs::empty()))
+ .decode_delete_record_content(buf.as_slice(), &HashMap::new())
+ .expect_err("a V2 delete block must be refused");
+ assert!(
+ err.to_string().contains("delete block version V2"),
+ "the error must name the block and version, got: {err}"
+ );
+ }
+
/// An Avro block records the schema it was written with in its header.
With
/// none there is nothing to decode against, so the read stops rather than
/// guessing at the table's current schema — which would misread a block
diff --git a/crates/core/tests/log_block_compat_tests.rs
b/crates/core/tests/log_block_compat_tests.rs
new file mode 100644
index 00000000..ef792983
--- /dev/null
+++ b/crates/core/tests/log_block_compat_tests.rs
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+//! Merge-on-read reads of log blocks built byte by byte, for block shapes no
+//! checked-in fixture carries: an older log block version and an Avro `enum`
+//! column that the base file stores as a string.
+
+use std::fs;
+use std::path::Path;
+use std::sync::Arc;
+
+use apache_avro::types::Value;
+use arrow::compute::concat_batches;
+use arrow_array::{ArrayRef, Int64Array, RecordBatch, StringArray};
+use arrow_schema::{DataType, Field, Schema};
+use hudi_core::error::Result;
+use hudi_core::table::{ReadOptions, Table};
+use parquet::arrow::ArrowWriter;
+
+const BASE_INSTANT: &str = "20250101000000000";
+const LOG_INSTANT: &str = "20250102000000000";
+const FILE_ID: &str = "f0000000-0000-0000-0000-000000000000-0";
+
+const AVRO_SCHEMA: &str =
r#"{"type":"record","name":"EnumRecord","namespace":"test","fields":[
+{"name":"_hoodie_commit_time","type":["null","string"],"default":null},
+{"name":"_hoodie_commit_seqno","type":["null","string"],"default":null},
+{"name":"_hoodie_record_key","type":["null","string"],"default":null},
+{"name":"_hoodie_partition_path","type":["null","string"],"default":null},
+{"name":"_hoodie_file_name","type":["null","string"],"default":null},
+{"name":"id","type":"string"},
+{"name":"kind","type":{"type":"enum","name":"Kind","symbols":["A","B","C"]}},
+{"name":"ts","type":"long"}]}"#;
+
+const SYMBOLS: [&str; 3] = ["A", "B", "C"];
+
+fn meta(v: &str) -> Value {
+ Value::Union(1, Box::new(Value::String(v.to_string())))
+}
+
+fn log_record(id: &str, kind: &str, ts: i64) -> Value {
+ let idx = SYMBOLS.iter().position(|s| *s == kind).unwrap() as u32;
+ Value::Record(vec![
+ ("_hoodie_commit_time".into(), meta(LOG_INSTANT)),
+ (
+ "_hoodie_commit_seqno".into(),
+ meta(&format!("{LOG_INSTANT}_0_{id}")),
+ ),
+ ("_hoodie_record_key".into(), meta(id)),
+ ("_hoodie_partition_path".into(), meta("")),
+ ("_hoodie_file_name".into(), meta("")),
+ ("id".into(), Value::String(id.to_string())),
+ ("kind".into(), Value::Enum(idx, kind.to_string())),
+ ("ts".into(), Value::Long(ts)),
+ ])
+}
+
+/// One log file holding a single Avro data block stamped `block_version`.
+fn write_log_file(path: &Path, block_version: u32, records: &[(&str, &str,
i64)]) -> usize {
+ let schema = apache_avro::Schema::parse_str(AVRO_SCHEMA).unwrap();
+ let mut content = Vec::new();
+ content.extend_from_slice(&block_version.to_be_bytes());
+ content.extend_from_slice(&(records.len() as u32).to_be_bytes());
+ for (id, kind, ts) in records {
+ let datum = apache_avro::to_avro_datum(&schema, log_record(id, kind,
*ts)).unwrap();
+ content.extend_from_slice(&(datum.len() as u32).to_be_bytes());
+ content.extend_from_slice(&datum);
+ }
+
+ let mut inner = Vec::new();
+ inner.extend_from_slice(&1u32.to_be_bytes()); // log format version
+ inner.extend_from_slice(&3u32.to_be_bytes()); // AvroData block
+ inner.extend_from_slice(&2u32.to_be_bytes()); // header entries
+ for (key, value) in [(0u32, LOG_INSTANT), (2u32, AVRO_SCHEMA)] {
+ inner.extend_from_slice(&key.to_be_bytes());
+ inner.extend_from_slice(&(value.len() as u32).to_be_bytes());
+ inner.extend_from_slice(value.as_bytes());
+ }
+ inner.extend_from_slice(&(content.len() as u64).to_be_bytes());
+ inner.extend_from_slice(&content);
+ inner.extend_from_slice(&0u32.to_be_bytes()); // footer entries
+
+ let block_length = (inner.len() + 8) as u64;
+ let mut out = Vec::new();
+ out.extend_from_slice(b"#HUDI#");
+ out.extend_from_slice(&block_length.to_be_bytes());
+ out.extend_from_slice(&inner);
+ out.extend_from_slice(&(block_length + 6).to_be_bytes());
+ fs::write(path, &out).unwrap();
+ out.len()
+}
+
+fn write_base_file(path: &Path, rows: &[(&str, &str, i64)]) -> u64 {
+ let utf8 = |name| Field::new(name, DataType::Utf8, true);
+ let schema = Arc::new(Schema::new(vec![
+ utf8("_hoodie_commit_time"),
+ utf8("_hoodie_commit_seqno"),
+ utf8("_hoodie_record_key"),
+ utf8("_hoodie_partition_path"),
+ utf8("_hoodie_file_name"),
+ Field::new("id", DataType::Utf8, false),
+ Field::new("kind", DataType::Utf8, false),
+ Field::new("ts", DataType::Int64, false),
+ ]));
+ let strings = |f: &dyn Fn(&(&str, &str, i64)) -> String| -> ArrayRef {
+ Arc::new(StringArray::from(rows.iter().map(f).collect::<Vec<_>>()))
+ };
+ let columns = vec![
+ strings(&|_| BASE_INSTANT.to_string()),
+ strings(&|r| format!("{BASE_INSTANT}_0_{}", r.0)),
+ strings(&|r| r.0.to_string()),
+ strings(&|_| String::new()),
+ strings(&|_| String::new()),
+ strings(&|r| r.0.to_string()),
+ strings(&|r| r.1.to_string()),
+ Arc::new(Int64Array::from(
+ rows.iter().map(|r| r.2).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ ];
+ let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
+ let mut writer = ArrowWriter::try_new(fs::File::create(path).unwrap(),
schema, None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ fs::metadata(path).unwrap().len()
+}
+
+fn write_stat(path: &str, base_file: &str, size: u64) -> serde_json::Value {
+ serde_json::json!({
+ "fileId": FILE_ID, "path": path, "baseFile": base_file, "prevCommit":
BASE_INSTANT,
+ "numWrites": 0, "numDeletes": 0, "numUpdateWrites": 0,
"totalWriteBytes": size,
+ "fileSizeInBytes": size, "partitionPath": "", "totalLogBlocks": 0,
+ })
+}
+
+fn write_commit(root: &Path, file_name: &str, operation: &str, stat:
serde_json::Value) {
+ let commit = serde_json::json!({
+ "partitionToWriteStats": { "": [stat] },
+ "compacted": false,
+ "extraMetadata": { "schema": AVRO_SCHEMA },
+ "operationType": operation,
+ });
+ fs::write(root.join(".hoodie").join(file_name),
commit.to_string()).unwrap();
+}
+
+/// A version 6 merge-on-read table: base rows `k0` and `k1`, then one log
block
+/// that updates `k1` and inserts `k2`, all carrying the enum column `kind`.
+fn create_table(root: &Path, block_version: u32) {
+ fs::create_dir_all(root.join(".hoodie")).unwrap();
+ let base_name = format!("{FILE_ID}_0-1-0_{BASE_INSTANT}.parquet");
+ let base_size = write_base_file(&root.join(&base_name), &[("k0", "A", 1),
("k1", "B", 1)]);
+ write_commit(
+ root,
+ &format!("{BASE_INSTANT}.commit"),
+ "INSERT",
+ write_stat(&base_name, &base_name, base_size),
+ );
+
+ let log_name = format!(".{FILE_ID}_{BASE_INSTANT}.log.1_0-2-0");
+ let log_size = write_log_file(
+ &root.join(&log_name),
+ block_version,
+ &[("k1", "C", 2), ("k2", "A", 2)],
+ );
+ write_commit(
+ root,
+ &format!("{LOG_INSTANT}.deltacommit"),
+ "UPSERT",
+ write_stat(&log_name, &base_name, log_size as u64),
+ );
+
+ let props = [
+ "hoodie.table.name=log_block_compat",
+ "hoodie.table.type=MERGE_ON_READ",
+ "hoodie.table.version=6",
+ "hoodie.timeline.layout.version=1",
+ "hoodie.table.recordkey.fields=id",
+ "hoodie.table.precombine.field=ts",
+ "hoodie.record.merge.mode=COMMIT_TIME_ORDERING",
+ "hoodie.table.keygenerator.type=NON_PARTITION",
+ "hoodie.datasource.write.drop.partition.columns=false",
+ ];
+ fs::write(root.join(".hoodie/hoodie.properties"),
props.join("\n")).unwrap();
+}
+
+async fn read_id_and_kind(root: &Path) -> Result<Vec<(String, String)>> {
+ let table = Table::new(root.to_str().unwrap()).await?;
+ let batches = table.read(&ReadOptions::new()).await?;
+ let batch = concat_batches(&batches[0].schema(), &batches)?;
+ assert_eq!(
+ batch.schema().field_with_name("kind")?.data_type(),
+ &DataType::Utf8
+ );
+ let column = |name| {
+ batch
+ .column_by_name(name)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .clone()
+ };
+ let (ids, kinds) = (column("id"), column("kind"));
+ let mut rows: Vec<(String, String)> = (0..batch.num_rows())
+ .map(|i| (ids.value(i).to_string(), kinds.value(i).to_string()))
+ .collect();
+ rows.sort();
+ Ok(rows)
+}
+
+fn expected() -> Vec<(String, String)> {
+ [("k0", "A"), ("k1", "C"), ("k2", "A")]
+ .iter()
+ .map(|(id, kind)| (id.to_string(), kind.to_string()))
+ .collect()
+}
+
+/// A log update and a log insert on an enum column merge onto a base file that
+/// stores the column as a string.
+#[tokio::test]
+async fn test_mor_read_merges_enum_column_from_log_block() -> Result<()> {
+ let dir = tempfile::tempdir().unwrap();
+ create_table(dir.path(), 3);
+ assert_eq!(read_id_and_kind(dir.path()).await?, expected());
+ Ok(())
+}
+
+/// Writers before table version 6 stamp their Avro data blocks V2. A table
+/// upgraded from one keeps those blocks until compaction, and they merge the
+/// same as V3.
+#[tokio::test]
+async fn test_mor_read_merges_block_version_2_avro_data_block() -> Result<()> {
+ let dir = tempfile::tempdir().unwrap();
+ create_table(dir.path(), 2);
+ assert_eq!(read_id_and_kind(dir.path()).await?, expected());
+ Ok(())
+}
