TengHuo commented on issue #7284: URL: https://github.com/apache/hudi/issues/7284#issuecomment-1324899843
@voonhous and me did some trouble shooting on this issue. And we found it is cased by the difference between writer schema and reader schema at this line: https://github.com/apache/hudi/blob/76a28daeb08e7192d75dfc447624c827643bef0d/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieAvroDataBlock.java#L171 Writer schema: ```json { "type": "record", "name": "test_mor_tab_record", "namespace": "hoodie.test_mor_tab", "fields": [ { "name": "_hoodie_commit_time", "type": [ "null", "string" ], "doc": "", "default": null }, { "name": "_hoodie_commit_seqno", "type": [ "null", "string" ], "doc": "", "default": null }, { "name": "_hoodie_record_key", "type": [ "null", "string" ], "doc": "", "default": null }, { "name": "_hoodie_partition_path", "type": [ "null", "string" ], "doc": "", "default": null }, { "name": "_hoodie_file_name", "type": [ "null", "string" ], "doc": "", "default": null }, { "name": "id", "type": "int" }, { "name": "name", "type": "string" }, { "name": "price", "type": "double" }, { "name": "ts", "type": "long" }, { "name": "new_test_col", "type": { "type": "fixed", "name": "fixed", "namespace": "hoodie.test_mor_tab.test_mor_tab_record.new_test_col", "size": 11, "logicalType": "decimal", "precision": 25, "scale": 4 }, "doc": "a column for test decimal type" }, { "name": "dt", "type": "string" } ] } ``` Reader schema: ```json { "type": "record", "name": "Record", "fields": [ { "name": "_hoodie_commit_time", "type": [ "string", "null" ] }, { "name": "_hoodie_commit_seqno", "type": [ "string", "null" ] }, { "name": "_hoodie_record_key", "type": [ "string", "null" ] }, { "name": "_hoodie_partition_path", "type": [ "string", "null" ] }, { "name": "_hoodie_file_name", "type": [ "string", "null" ] }, { "name": "id", "type": [ "int", "null" ] }, { "name": "name", "type": [ "string", "null" ] }, { "name": "price", "type": [ "double", "null" ] }, { "name": "ts", "type": [ "long", "null" ] }, { "name": "new_test_col", "type": [ { "type": "fixed", "name": "fixed", "namespace": "Record.new_test_col", "size": 11, "logicalType": "decimal", "precision": 25, "scale": 4 }, "null" ] }, { "name": "dt", "type": [ "string", "null" ] } ] } ``` It can be saw in writer schema, the type of column `new_test_col` is a `fixed` type, and with namespace is `hoodie.test_mor_tab.test_mor_tab_record.new_test_col`. ```json { "name": "new_test_col", "type": { "type": "fixed", "name": "fixed", "namespace": "hoodie.test_mor_tab.test_mor_tab_record.new_test_col", "size": 11, "logicalType": "decimal", "precision": 25, "scale": 4 }, "doc": "a column for test decimal type" } ``` But in reader schema, the type of column `new_test_col` is a `union` type, and with namespace is`Record.new_test_col`. ```json { "name": "new_test_col", "type": [ { "type": "fixed", "name": "fixed", "namespace": "Record.new_test_col", "size": 11, "logicalType": "decimal", "precision": 25, "scale": 4 }, "null" ] } ``` According to [Avro doc](https://avro.apache.org/docs/1.8.2/spec.html#Schema+Resolution), `UNION` type is compatible in schema evolution with other primitive types. So, it is acceptable to read "fixed" type data with `union` type. However, the namespace in reader schema is different with writer schema, it causes the exception mentioned above `org.apache.avro.AvroTypeException: Found hoodie.test_mor_tab.test_mor_tab_record.new_test_col.fixed, expecting union`. If I replace the reader schema with the same namespace as writer schema, the test case can run properly. ```java ... private RecordIterator(Schema readerSchema, Schema writerSchema, byte[] content, InternalSchema internalSchema) throws IOException { this.content = content; this.dis = new SizeAwareDataInputStream(new DataInputStream(new ByteArrayInputStream(this.content))); // 1. Read version for this data block int version = this.dis.readInt(); HoodieAvroDataBlockVersion logBlockVersion = new HoodieAvroDataBlockVersion(version); Schema finalReadSchema = readerSchema; if (!internalSchema.isEmptySchema()) { // we should use write schema to read log file, // since when we have done some DDL operation, the readerSchema maybe different from writeSchema, avro reader will throw exception. // eg: origin writeSchema is: "a String, b double" then we add a new column now the readerSchema will be: "a string, c int, b double". it's wrong to use readerSchema to read old log file. // after we read those record by writeSchema, we rewrite those record with readerSchema in AbstractHoodieLogRecordReader finalReadSchema = writerSchema; } Schema readSchema = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Record\",\"fields\":[{\"name\":\"_hoodie_commit_time\",\"type\":[\"string\",\"null\"]},{\"name\":\"_hoodie_commit_seqno\",\"type\":[\"string\",\"null\"]},{\"name\":\"_hoodie_record_key\",\"type\":[\"string\",\"null\"]},{\"name\":\"_hoodie_partition_path\",\"type\":[\"string\",\"null\"]},{\"name\":\"_hoodie_file_name\",\"type\":[\"string\",\"null\"]},{\"name\":\"id\",\"type\":[\"int\",\"null\"]},{\"name\":\"name\",\"type\":[\"string\",\"null\"]},{\"name\":\"price\",\"type\":[\"double\",\"null\"]},{\"name\":\"ts\",\"type\":[\"long\",\"null\"]},{\"name\":\"new_test_col\",\"type\":[{\"type\":\"fixed\",\"name\":\"fixed\",\"namespace\":\"hoodie.test_mor_tab.test_mor_tab_record.new_test_col\",\"size\":11,\"logicalType\":\"decimal\",\"precision\":25,\"scale\":4},\"null\"]},{\"name\":\"dt\",\"type\":[\"string\",\"null\"]}]}"); this.reader = new GenericDatumReader<>(writerSchema, readSchema); if (logBlockVersion.hasRecordCount()) { this.totalRecords = this.dis.readInt(); } } ... ``` -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
