lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-4132] Fixing determining target table schema for delta sync with empty batch (#5648)

Browse Source
This commit is contained in:
Sivabalan Narayanan
2022-05-24 08:17:15 -04:00
committed by GitHub
parent 0caa55ecb4
commit 10363c1412
1 changed files with 9 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
View File

@@ -840,8 +840,15 @@ public class DeltaSync implements Serializable {
&& SchemaCompatibility.checkReaderWriterCompatibility(InputBatch.NULL_SCHEMA, targetSchema).getType() == SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE) { && SchemaCompatibility.checkReaderWriterCompatibility(InputBatch.NULL_SCHEMA, targetSchema).getType() == SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE) {
// target schema is null. fetch schema from commit metadata and use it // target schema is null. fetch schema from commit metadata and use it
HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(new Configuration(fs.getConf())).setBasePath(cfg.targetBasePath).setPayloadClassName(cfg.payloadClassName).build(); HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(new Configuration(fs.getConf())).setBasePath(cfg.targetBasePath).setPayloadClassName(cfg.payloadClassName).build();
int totalCompleted = meta.getActiveTimeline().getCommitsTimeline().filterCompletedInstants().countInstants();
if (totalCompleted > 0) {
try {
TableSchemaResolver schemaResolver = new TableSchemaResolver(meta); TableSchemaResolver schemaResolver = new TableSchemaResolver(meta);
newWriteSchema = schemaResolver.getTableAvroSchema(false); newWriteSchema = schemaResolver.getTableAvroSchema(false);
} catch (IllegalArgumentException e) {
LOG.warn("Could not fetch schema from table. Falling back to using target schema from schema provider");
}
}
} }
} }
return newWriteSchema; return newWriteSchema;