[HUDI-7237] Hudi Streamer: Handle edge case with null schema, minor cleanups (apache#10342)

Author: the-other-tim-brown

hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java

@@ -144,7 +144,7 @@ public static Schema reconcileSchemaRequirements(Schema sourceSchema, Schema tar
       return sourceSchema;
     }
 
-    if (sourceSchema.getType() == Schema.Type.NULL || sourceSchema.getFields().isEmpty()) {
+    if (sourceSchema == null || sourceSchema.getType() == Schema.Type.NULL || sourceSchema.getFields().isEmpty()) {
       return targetSchema;
     }
 

hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaProviderWithPostProcessor.java

@@ -18,9 +18,10 @@
 
 package org.apache.hudi.utilities.schema;
 
-import org.apache.avro.Schema;
 import org.apache.hudi.common.util.Option;
 
+import org.apache.avro.Schema;
+
 /**
  * A schema provider which applies schema post process hook on schema.
  */
@@ -38,14 +39,16 @@ public SchemaProviderWithPostProcessor(SchemaProvider schemaProvider,
 
   @Override
   public Schema getSourceSchema() {
-    return schemaPostProcessor.map(processor -> processor.processSchema(schemaProvider.getSourceSchema()))
-        .orElse(schemaProvider.getSourceSchema());
+    Schema sourceSchema = schemaProvider.getSourceSchema();
+    return schemaPostProcessor.map(processor -> processor.processSchema(sourceSchema))
+        .orElse(sourceSchema);
   }
 
   @Override
   public Schema getTargetSchema() {
-    return schemaPostProcessor.map(processor -> processor.processSchema(schemaProvider.getTargetSchema()))
-        .orElse(schemaProvider.getTargetSchema());
+    Schema targetSchema = schemaProvider.getTargetSchema();
+    return schemaPostProcessor.map(processor -> processor.processSchema(targetSchema))
+        .orElse(targetSchema);
   }
 
   public SchemaProvider getOriginalSchemaProvider() {

hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/InputBatch.java

@@ -55,12 +55,16 @@ public SchemaProvider getSchemaProvider() {
     if (batch.isPresent() && schemaProvider == null) {
       throw new HoodieException("Please provide a valid schema provider class!");
     }
-    return Option.ofNullable(schemaProvider).orElse(new NullSchemaProvider());
+    return Option.ofNullable(schemaProvider).orElseGet(NullSchemaProvider::getInstance);
   }
 
   public static class NullSchemaProvider extends SchemaProvider {
+    private static final NullSchemaProvider INSTANCE = new NullSchemaProvider();
+    public static NullSchemaProvider getInstance() {
+      return INSTANCE;
+    }
 
-    public NullSchemaProvider() {
+    private NullSchemaProvider() {
       this(null, null);
     }
 

hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java

@@ -274,18 +274,16 @@ public StreamSync(HoodieStreamer.Config cfg, SparkSession sparkSession, SchemaPr
     this.processedSchema = new SchemaSet();
     this.autoGenerateRecordKeys = KeyGenUtils.enableAutoGenerateRecordKeys(props);
     this.keyGenClassName = getKeyGeneratorClassName(new TypedProperties(props));
-    refreshTimeline();
-    // Register User Provided schema first
-    registerAvroSchemas(schemaProvider);
-
-
-    this.metrics = (HoodieIngestionMetrics) ReflectionUtils.loadClass(cfg.ingestionMetricsClass, getHoodieClientConfig(this.schemaProvider));
-    this.hoodieMetrics = new HoodieMetrics(getHoodieClientConfig(this.schemaProvider));
     this.conf = conf;
+
+    HoodieWriteConfig hoodieWriteConfig = getHoodieClientConfig();
+    this.metrics = (HoodieIngestionMetrics) ReflectionUtils.loadClass(cfg.ingestionMetricsClass, hoodieWriteConfig);
+    this.hoodieMetrics = new HoodieMetrics(hoodieWriteConfig);
     if (props.getBoolean(ERROR_TABLE_ENABLED.key(), ERROR_TABLE_ENABLED.defaultValue())) {
       this.errorTableWriter = ErrorTableUtils.getErrorTableWriter(cfg, sparkSession, props, hoodieSparkContext, fs);
       this.errorWriteFailureStrategy = ErrorTableUtils.getErrorWriteFailureStrategy(props);
     }
+    refreshTimeline();
     Source source = UtilHelpers.createSource(cfg.sourceClassName, props, hoodieSparkContext.jsc(), sparkSession, schemaProvider, metrics);
     this.formatAdapter = new SourceFormatAdapter(source, this.errorTableWriter, Option.of(props));
 
@@ -309,7 +307,7 @@ public void refreshTimeline() throws IOException {
     if (fs.exists(new Path(cfg.targetBasePath))) {
       try {
         HoodieTableMetaClient meta = HoodieTableMetaClient.builder()
-            .setConf(new Configuration(fs.getConf()))
+            .setConf(conf)
             .setBasePath(cfg.targetBasePath)
             .setPayloadClassName(cfg.payloadClassName)
             .setRecordMergerStrategy(props.getProperty(HoodieWriteConfig.RECORD_MERGER_STRATEGY.key(), HoodieWriteConfig.RECORD_MERGER_STRATEGY.defaultValue()))
@@ -337,7 +335,7 @@ public void refreshTimeline() throws IOException {
             LOG.warn("Base path exists, but table is not fully initialized. Re-initializing again");
             initializeEmptyTable();
             // reload the timeline from metaClient and validate that its empty table. If there are any instants found, then we should fail the pipeline, bcoz hoodie.properties got deleted by mistake.
-            HoodieTableMetaClient metaClientToValidate = HoodieTableMetaClient.builder().setConf(new Configuration(fs.getConf())).setBasePath(cfg.targetBasePath).build();
+            HoodieTableMetaClient metaClientToValidate = HoodieTableMetaClient.builder().setConf(conf).setBasePath(cfg.targetBasePath).build();
             if (metaClientToValidate.reloadActiveTimeline().countInstants() > 0) {
               // Deleting the recreated hoodie.properties and throwing exception.
               fs.delete(new Path(String.format("%s%s/%s", basePathWithForwardSlash, HoodieTableMetaClient.METAFOLDER_NAME, HoodieTableConfig.HOODIE_PROPERTIES_FILE)));
@@ -395,7 +393,7 @@ public Pair<Option<String>, JavaRDD<WriteStatus>> syncOnce() throws IOException
     // Refresh Timeline
     refreshTimeline();
     HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder()
-        .setConf(new Configuration(fs.getConf()))
+        .setConf(conf)
         .setBasePath(cfg.targetBasePath)
         .setRecordMergerStrategy(props.getProperty(HoodieWriteConfig.RECORD_MERGER_STRATEGY.key(), HoodieWriteConfig.RECORD_MERGER_STRATEGY.defaultValue()))
         .setTimeGeneratorConfig(HoodieTimeGeneratorConfig.newBuilder().fromProperties(props).withPath(cfg.targetBasePath).build())
@@ -432,15 +430,15 @@ public Pair<Option<String>, JavaRDD<WriteStatus>> syncOnce() throws IOException
       }
 
       // complete the pending compaction before writing to sink
-      if (cfg.retryLastPendingInlineCompactionJob && getHoodieClientConfig(this.schemaProvider).inlineCompactionEnabled()) {
+      if (cfg.retryLastPendingInlineCompactionJob && writeClient.getConfig().inlineCompactionEnabled()) {
         Option<String> pendingCompactionInstant = getLastPendingCompactionInstant(allCommitsTimelineOpt);
         if (pendingCompactionInstant.isPresent()) {
           HoodieWriteMetadata<JavaRDD<WriteStatus>> writeMetadata = writeClient.compact(pendingCompactionInstant.get());
           writeClient.commitCompaction(pendingCompactionInstant.get(), writeMetadata.getCommitMetadata().get(), Option.empty());
           refreshTimeline();
           reInitWriteClient(schemaProvider.getSourceSchema(), schemaProvider.getTargetSchema(), null);
         }
-      } else if (cfg.retryLastPendingInlineClusteringJob && getHoodieClientConfig(this.schemaProvider).inlineClusteringEnabled()) {
+      } else if (cfg.retryLastPendingInlineClusteringJob && writeClient.getConfig().inlineClusteringEnabled()) {
         // complete the pending clustering before writing to sink
         Option<String> pendingClusteringInstant = getLastPendingClusteringInstant(allCommitsTimelineOpt);
         if (pendingClusteringInstant.isPresent()) {
@@ -1001,7 +999,7 @@ public void runMetaSync() {
    * this constraint.
    */
   private void setupWriteClient(Option<JavaRDD<HoodieRecord>> recordsOpt) throws IOException {
-    if ((null != schemaProvider)) {
+    if (null != schemaProvider) {
       Schema sourceSchema = schemaProvider.getSourceSchema();
       Schema targetSchema = schemaProvider.getTargetSchema();
       reInitWriteClient(sourceSchema, targetSchema, recordsOpt);
@@ -1013,8 +1011,9 @@ private void reInitWriteClient(Schema sourceSchema, Schema targetSchema, Option<
     if (HoodieStreamerUtils.isDropPartitionColumns(props)) {
       targetSchema = HoodieAvroUtils.removeFields(targetSchema, HoodieStreamerUtils.getPartitionColumns(props));
     }
-    registerAvroSchemas(sourceSchema, targetSchema);
-    final HoodieWriteConfig initialWriteConfig = getHoodieClientConfig(targetSchema);
+    final Pair<HoodieWriteConfig, Schema> initialWriteConfigAndSchema = getHoodieClientConfigAndWriterSchema(targetSchema, true);
+    final HoodieWriteConfig initialWriteConfig = initialWriteConfigAndSchema.getLeft();
+    registerAvroSchemas(sourceSchema, initialWriteConfigAndSchema.getRight());
     final HoodieWriteConfig writeConfig = SparkSampleWritesUtils
         .getWriteConfigWithRecordSizeEstimate(hoodieSparkContext.jsc(), recordsOpt, initialWriteConfig)
         .orElse(initialWriteConfig);
@@ -1036,20 +1035,21 @@ private void reInitWriteClient(Schema sourceSchema, Schema targetSchema, Option<
   }
 
   /**
-   * Helper to construct Write Client config.
-   *
-   * @param schemaProvider Schema Provider
+   * Helper to construct Write Client config without a schema.
    */
-  private HoodieWriteConfig getHoodieClientConfig(SchemaProvider schemaProvider) {
-    return getHoodieClientConfig(schemaProvider != null ? schemaProvider.getTargetSchema() : null);
+  private HoodieWriteConfig getHoodieClientConfig() {
+    return getHoodieClientConfigAndWriterSchema(null, false).getLeft();
   }
 
   /**
    * Helper to construct Write Client config.
    *
-   * @param schema Schema
+   * @param schema initial writer schema. If null or Avro Null type, the schema will be fetched from previous commit metadata for the table.
+   * @param requireSchemaInConfig whether the schema should be present in the config. This is an optimization to avoid fetching schema from previous commits if not needed.
+   *
+   * @return Pair of HoodieWriteConfig and writer schema.
    */
-  private HoodieWriteConfig getHoodieClientConfig(Schema schema) {
+  private Pair<HoodieWriteConfig, Schema> getHoodieClientConfigAndWriterSchema(Schema schema, boolean requireSchemaInConfig) {
     final boolean combineBeforeUpsert = true;
     final boolean autoCommit = false;
 
@@ -1075,8 +1075,13 @@ private HoodieWriteConfig getHoodieClientConfig(Schema schema) {
             .withAutoCommit(autoCommit)
             .withProps(props);
 
-    if (schema != null) {
-      builder.withSchema(getSchemaForWriteConfig(schema).toString());
+    // If schema is required in the config, we need to handle the case where the target schema is null and should be fetched from previous commits
+    final Schema returnSchema;
+    if (requireSchemaInConfig) {
+      returnSchema = getSchemaForWriteConfig(schema);
+      builder.withSchema(returnSchema.toString());
+    } else {
+      returnSchema = schema;
     }
 
     HoodieWriteConfig config = builder.build();
@@ -1108,30 +1113,28 @@ private HoodieWriteConfig getHoodieClientConfig(Schema schema) {
         String.format("%s should be set to %s", COMBINE_BEFORE_INSERT.key(), cfg.filterDupes));
     ValidationUtils.checkArgument(config.shouldCombineBeforeUpsert(),
         String.format("%s should be set to %s", COMBINE_BEFORE_UPSERT.key(), combineBeforeUpsert));
-    return config;
+    return Pair.of(config, returnSchema);
   }
 
   private Schema getSchemaForWriteConfig(Schema targetSchema) {
     Schema newWriteSchema = targetSchema;
     try {
-      if (targetSchema != null) {
-        // check if targetSchema is equal to NULL schema
-        if (SchemaCompatibility.checkReaderWriterCompatibility(targetSchema, InputBatch.NULL_SCHEMA).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
-          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) {
-            TableSchemaResolver schemaResolver = new TableSchemaResolver(meta);
-            Option<Schema> tableSchema = schemaResolver.getTableAvroSchemaIfPresent(false);
-            if (tableSchema.isPresent()) {
-              newWriteSchema = tableSchema.get();
-            } else {
-              LOG.warn("Could not fetch schema from table. Falling back to using target schema from schema provider");
-            }
+      // check if targetSchema is equal to NULL schema
+      if (targetSchema == null || (SchemaCompatibility.checkReaderWriterCompatibility(targetSchema, InputBatch.NULL_SCHEMA).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
+        HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(conf)
+            .setBasePath(cfg.targetBasePath)
+            .setPayloadClassName(cfg.payloadClassName)
+            .build();
+        int totalCompleted = meta.getActiveTimeline().getCommitsTimeline().filterCompletedInstants().countInstants();
+        if (totalCompleted > 0) {
+          TableSchemaResolver schemaResolver = new TableSchemaResolver(meta);
+          Option<Schema> tableSchema = schemaResolver.getTableAvroSchemaIfPresent(false);
+          if (tableSchema.isPresent()) {
+            newWriteSchema = tableSchema.get();
+          } else {
+            LOG.warn("Could not fetch schema from table. Falling back to using target schema from schema provider");
           }
         }
       }