[HUDI-1453] Throw Exception when input data schema is not equal to th… #2334
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -28,7 +28,6 @@ | |
| import org.apache.hudi.common.util.HoodieTimer; | ||
| import org.apache.hudi.common.util.Option; | ||
| import org.apache.hudi.common.util.ReflectionUtils; | ||
| import org.apache.hudi.config.HoodieWriteConfig; | ||
| import org.apache.hudi.exception.HoodieIOException; | ||
| import org.apache.hudi.io.storage.HoodieFileWriter; | ||
|
|
@@ -55,47 +54,70 @@ public abstract class HoodieWriteHandle<T extends HoodieRecordPayload, I, K, O> | |
|
|
||
| private static final Logger LOG = LogManager.getLogger(HoodieWriteHandle.class); | ||
|
|
||
| /** | ||
| * The table schema is the schema of the table which used to read/write record from table. | ||
| */ | ||
| protected final Schema tableSchema; | ||
|
||
| /** | ||
| * The table schema with meta fields. | ||
| */ | ||
| protected final Schema tableSchemaWithMetaFields; | ||
| /** | ||
| * The input schema is the input data schema which used to parse data from incoming record. | ||
| */ | ||
| protected final Schema inputSchema; | ||
| /** | ||
| * The input schema with meta fields. | ||
| */ | ||
| protected final Schema inputSchemaWithMetaFields; | ||
|
|
||
| protected HoodieTimer timer; | ||
| protected WriteStatus writeStatus; | ||
| protected final String partitionPath; | ||
| protected final String fileId; | ||
| protected final String writeToken; | ||
| protected final TaskContextSupplier taskContextSupplier; | ||
|
|
||
| /** | ||
| * | ||
| * @param config the write config | ||
| * @param instantTime the instance time | ||
| * @param partitionPath the partition path | ||
| * @param fileId the file id | ||
| * @param hoodieTable the hoodie table | ||
| * @param schemaOption the option schema specified for HoodieBootstrapHandle | ||
| * @param taskContextSupplier | ||
| */ | ||
| protected HoodieWriteHandle(HoodieWriteConfig config, String instantTime, String partitionPath, | ||
| String fileId, HoodieTable<T, I, K, O> hoodieTable, | ||
| Option<Schema> schemaOption, | ||
| TaskContextSupplier taskContextSupplier) { | ||
|
|
||
| super(config, instantTime, hoodieTable); | ||
| this.partitionPath = partitionPath; | ||
| this.fileId = fileId; | ||
| // If the schemaOption has specified,use it,otherwise load the schema from the table. | ||
| // The schemaOption will be specified only for HoodieBootstrapHandle. | ||
| this.tableSchema = schemaOption.orElseGet(() -> hoodieTable.getTableSchema(config, false)); | ||
|
||
| this.tableSchemaWithMetaFields = HoodieAvroUtils.addMetadataFields(tableSchema); | ||
| this.inputSchema = schemaOption.orElseGet(() -> getInputSchema(config)); | ||
| this.inputSchemaWithMetaFields = HoodieAvroUtils.addMetadataFields(inputSchema); | ||
| this.timer = new HoodieTimer().startTimer(); | ||
| this.writeStatus = (WriteStatus) ReflectionUtils.loadClass(config.getWriteStatusClassName(), | ||
| !hoodieTable.getIndex().isImplicitWithStorage(), config.getWriteStatusFailureFraction()); | ||
| this.taskContextSupplier = taskContextSupplier; | ||
| this.writeToken = makeWriteToken(); | ||
| } | ||
|
|
||
| public HoodieWriteHandle(HoodieWriteConfig config, String instantTime, String partitionPath, | ||
| String fileId, HoodieTable<T, I, K, O> hoodieTable, | ||
| TaskContextSupplier taskContextSupplier) { | ||
| this(config, instantTime, partitionPath, fileId, hoodieTable, | ||
| Option.empty(), taskContextSupplier); | ||
| } | ||
|
|
||
| private static Schema getInputSchema(HoodieWriteConfig config) { | ||
| return new Schema.Parser().parse(config.getSchema()); | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -127,8 +149,12 @@ protected void createMarkerFile(String partitionPath, String dataFileName) { | |
| markerFiles.create(partitionPath, dataFileName, getIOType()); | ||
| } | ||
|
|
||
| public Schema getTableSchemaWithMetaFields() { | ||
| return tableSchemaWithMetaFields; | ||
| } | ||
|
|
||
| public Schema getInputSchemaWithMetaFields() { | ||
| return inputSchemaWithMetaFields; | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -166,7 +192,7 @@ public void write(HoodieRecord record, Option<IndexedRecord> avroRecord, Option< | |
| * Rewrite the GenericRecord with the Schema containing the Hoodie Metadata fields. | ||
| */ | ||
| protected GenericRecord rewriteRecord(GenericRecord record) { | ||
| return HoodieAvroUtils.rewriteRecord(record, inputSchemaWithMetaFields); | ||
| } | ||
|
|
||
| public abstract List<WriteStatus> close(); | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@bvaradar : can you confirm that this change looks good.