[HUDI-8637] Refactor broadcast to a more generic reader context factory

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieCompactionHandler.java

@@ -28,8 +28,6 @@
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieNotSupportedException;
 
-import org.apache.hadoop.conf.Configuration;
-
 import java.io.IOException;
 import java.util.Iterator;
 import java.util.List;
@@ -50,8 +48,7 @@ Iterator<List<WriteStatus>> handleInsert(String instantTime, String partitionPat
   default List<WriteStatus> compactUsingFileGroupReader(String instantTime,
                                                         CompactionOperation operation,
                                                         HoodieWriteConfig writeConfig,
-                                                        HoodieReaderContext readerContext,
-                                                        Configuration conf) {
+                                                        HoodieReaderContext readerContext) {
     throw new HoodieNotSupportedException("This engine does not support file group reader based compaction.");
   }
 

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java

@@ -24,6 +24,8 @@
 import org.apache.hudi.common.config.HoodieReaderConfig;
 import org.apache.hudi.common.data.HoodieData;
 import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.engine.ReaderContextFactory;
 import org.apache.hudi.common.engine.TaskContextSupplier;
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.model.CompactionOperation;
@@ -46,7 +48,6 @@
 import org.apache.hudi.io.IOUtils;
 import org.apache.hudi.storage.HoodieStorage;
 import org.apache.hudi.storage.StoragePath;
-import org.apache.hudi.table.EngineBroadcastManager;
 import org.apache.hudi.table.HoodieCompactionHandler;
 import org.apache.hudi.table.HoodieTable;
 import org.apache.hudi.table.action.compact.strategy.CompactionStrategy;
@@ -89,16 +90,6 @@ public abstract void preCompact(
    */
   public abstract void maybePersist(HoodieData<WriteStatus> writeStatus, HoodieEngineContext context, HoodieWriteConfig config, String instantTime);
 
-  /**
-   * @param context {@link HoodieEngineContext} instance
-   *
-   * @return the {@link EngineBroadcastManager} if available.
-   */
-  public Option<EngineBroadcastManager> getEngineBroadcastManager(HoodieEngineContext context,
-                                                                  HoodieTableMetaClient metaClient) {
-    return Option.empty();
-  }
-
   /**
    * Execute compaction operations and report back status.
    */
@@ -152,11 +143,9 @@ public HoodieData<WriteStatus> compact(
         && config.populateMetaFields();                                             // Virtual key support by fg reader is not ready
 
     if (useFileGroupReaderBasedCompaction) {
-      Option<EngineBroadcastManager> broadcastManagerOpt = getEngineBroadcastManager(context, metaClient);
-      // Broadcast required information.
-      broadcastManagerOpt.ifPresent(EngineBroadcastManager::prepareAndBroadcast);
+      ReaderContextFactory<T> readerContextFactory = context.getReaderContextFactory(metaClient);
       return context.parallelize(operations).map(
-              operation -> compact(compactionHandler, metaClient, config, operation, compactionInstantTime, broadcastManagerOpt))
+              operation -> compact(compactionHandler, config, operation, compactionInstantTime, readerContextFactory.getContext()))
           .flatMap(List::iterator);
     } else {
       return context.parallelize(operations).map(
@@ -297,14 +286,12 @@ public List<WriteStatus> compact(HoodieCompactionHandler compactionHandler,
    * Execute a single compaction operation and report back status.
    */
   public List<WriteStatus> compact(HoodieCompactionHandler compactionHandler,
-                                   HoodieTableMetaClient metaClient,
                                    HoodieWriteConfig writeConfig,
                                    CompactionOperation operation,
                                    String instantTime,
-                                   Option<EngineBroadcastManager> broadcastManagerOpt) throws IOException {
+                                   HoodieReaderContext<T> hoodieReaderContext) throws IOException {
     return compactionHandler.compactUsingFileGroupReader(instantTime, operation,
-        writeConfig, broadcastManagerOpt.get().retrieveFileGroupReaderContext(metaClient.getBasePath()).get(),
-        broadcastManagerOpt.get().retrieveStorageConfig().get());
+        writeConfig, hoodieReaderContext);
   }
 
   public String getMaxInstantTime(HoodieTableMetaClient metaClient) {

hudi-client/hudi-java-client/src/test/java/org/apache/hudi/hadoop/TestHoodieFileGroupReaderOnHive.java

@@ -142,7 +142,7 @@ public HoodieReaderContext<ArrayWritable> getHoodieReaderContext(String tablePat
     setupJobconf(jobConf);
     return new HiveHoodieReaderContext(readerCreator, getRecordKeyField(metaClient),
         getStoredPartitionFieldNames(new JobConf(storageConf.unwrapAs(Configuration.class)), avroSchema),
-        new ObjectInspectorCache(avroSchema, jobConf));
+        new ObjectInspectorCache(avroSchema, jobConf), storageConf);
   }
 
   @Override

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java

@@ -33,6 +33,7 @@
 import org.apache.hudi.common.data.HoodieData;
 import org.apache.hudi.common.engine.HoodieEngineContext;
 import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.engine.ReaderContextFactory;
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.model.ClusteringOperation;
 import org.apache.hudi.common.model.FileSlice;
@@ -72,10 +73,8 @@
 import org.apache.hudi.storage.HoodieStorage;
 import org.apache.hudi.storage.StorageConfiguration;
 import org.apache.hudi.storage.StoragePath;
-import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
 import org.apache.hudi.table.BulkInsertPartitioner;
 import org.apache.hudi.table.HoodieTable;
-import org.apache.hudi.table.SparkBroadcastManager;
 import org.apache.hudi.table.action.HoodieWriteMetadata;
 import org.apache.hudi.table.action.cluster.strategy.ClusteringExecutionStrategy;
 
@@ -455,8 +454,7 @@ private Dataset<Row> readRecordsForGroupAsRowWithFileGroupReader(JavaSparkContex
     SerializableSchema serializableTableSchemaWithMetaFields = new SerializableSchema(tableSchemaWithMetaFields);
 
     // broadcast reader context.
-    SparkBroadcastManager broadcastManager = new SparkBroadcastManager(getEngineContext(), getHoodieTable().getMetaClient());
-    broadcastManager.prepareAndBroadcast();
+    ReaderContextFactory<InternalRow> readerContextFactory = getEngineContext().getReaderContextFactory(getHoodieTable().getMetaClient());
     StructType sparkSchemaWithMetaFields = AvroConversionUtils.convertAvroSchemaToStructType(tableSchemaWithMetaFields);
 
     RDD<InternalRow> internalRowRDD = jsc.parallelize(clusteringOps, clusteringOps.size()).flatMap(new FlatMapFunction<ClusteringOperation, InternalRow>() {
@@ -469,20 +467,17 @@ public Iterator<InternalRow> call(ClusteringOperation clusteringOperation) throw
         if (isInternalSchemaPresent) {
           internalSchemaOption = SerDeHelper.fromJson(internalSchemaStr);
         }
-        Option<HoodieReaderContext> readerContextOpt = broadcastManager.retrieveFileGroupReaderContext(new StoragePath(basePath));
-        Configuration conf = broadcastManager.retrieveStorageConfig().get();
 
         // instantiate FG reader
-        HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(readerContextOpt.get(),
-            getHoodieTable().getMetaClient().getStorage().newInstance(new StoragePath(basePath), new HadoopStorageConfiguration(conf)),
+        HoodieReaderContext<InternalRow> readerContext = readerContextFactory.getContext();
+        HoodieFileGroupReader<InternalRow> fileGroupReader = new HoodieFileGroupReader<>(readerContext,
+            getHoodieTable().getMetaClient().getStorage().newInstance(new StoragePath(basePath), readerContext.getStorageConfiguration()),
             basePath, instantTime, fileSlice, readerSchema, readerSchema, internalSchemaOption,
             getHoodieTable().getMetaClient(), getHoodieTable().getMetaClient().getTableConfig().getProps(),
             0, Long.MAX_VALUE, usePosition, false);
         fileGroupReader.initRecordIterators();
         // read records from the FG reader
-        HoodieFileGroupReader.HoodieFileGroupReaderIterator<InternalRow> recordIterator
-            = (HoodieFileGroupReader.HoodieFileGroupReaderIterator<InternalRow>) fileGroupReader.getClosableIterator();
-        return recordIterator;
+        return fileGroupReader.getClosableIterator();
       }
     }).rdd();
 

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/HoodieSparkEngineContext.java

@@ -25,11 +25,13 @@
 import org.apache.hudi.common.data.HoodiePairData;
 import org.apache.hudi.common.engine.EngineProperty;
 import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.ReaderContextFactory;
 import org.apache.hudi.common.function.SerializableBiFunction;
 import org.apache.hudi.common.function.SerializableConsumer;
 import org.apache.hudi.common.function.SerializableFunction;
 import org.apache.hudi.common.function.SerializablePairFlatMapFunction;
 import org.apache.hudi.common.function.SerializablePairFunction;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.util.Functions;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.collection.ImmutablePair;
@@ -48,6 +50,7 @@
 import org.apache.spark.api.java.function.Function2;
 import org.apache.spark.api.java.function.PairFlatMapFunction;
 import org.apache.spark.sql.SQLContext;
+import org.apache.spark.sql.catalyst.InternalRow;
 
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -250,6 +253,11 @@ public <I, O> O aggregate(HoodieData<I> data, O zeroValue, Functions.Function2<O
     return HoodieJavaRDD.getJavaRDD(data).aggregate(zeroValue, seqOpFunc, combOpFunc);
   }
 
+  @Override
+  public ReaderContextFactory<InternalRow> getReaderContextFactory(HoodieTableMetaClient metaClient) {
+    return new SparkReaderContextFactory(this, metaClient);
+  }
+
   public SparkConf getConf() {
     return javaSparkContext.getConf();
   }