[HUDI-3549] Removing dependency on "spark-avro"

README.md

@@ -90,21 +90,9 @@ mvn clean package -DskipTests -Dspark3
 mvn clean package -DskipTests -Dspark3.1.x
 ```
 
-### Build without spark-avro module
+### What about "spark-avro" module? 
 
-The default hudi-jar bundles spark-avro module. To build without spark-avro module, build using `spark-shade-unbundle-avro` profile
-
-```
-# Checkout code and build
-git clone https://github.com/apache/hudi.git && cd hudi
-mvn clean package -DskipTests -Pspark-shade-unbundle-avro
-
-# Start command
-spark-2.4.4-bin-hadoop2.7/bin/spark-shell \
-  --packages org.apache.spark:spark-avro_2.11:2.4.4 \
-  --jars `ls packaging/hudi-spark-bundle/target/hudi-spark-bundle_2.11-*.*.*-SNAPSHOT.jar` \
-  --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'
-```
+Starting from versions 0.11, Hudi no longer requires `spark-avro` to be specified using `--packages`
 
 ## Running Tests
 

docker/demo/config/test-suite/templates/spark_command.txt.template

@@ -15,7 +15,6 @@
 # limitations under the License.
 
 spark-submit \
---packages org.apache.spark:spark-avro_2.11:2.4.0 \
 --conf spark.task.cpus=1 \
 --conf spark.executor.cores=1 \
 --conf spark.task.maxFailures=100 \

hudi-cli/pom.xml

@@ -225,10 +225,6 @@
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-sql_${scala.binary.version}</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.apache.spark</groupId>
-      <artifactId>spark-avro_${scala.binary.version}</artifactId>
-    </dependency>
 
     <dependency>
       <groupId>org.springframework.shell</groupId>

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

@@ -18,6 +18,11 @@
 
 package org.apache.hudi.table;
 
+import org.apache.avro.Schema;
+import org.apache.avro.specific.SpecificRecordBase;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hudi.avro.HoodieAvroUtils;
 import org.apache.hudi.avro.model.HoodieCleanMetadata;
 import org.apache.hudi.avro.model.HoodieCleanerPlan;
@@ -72,17 +77,9 @@
 import org.apache.hudi.table.marker.WriteMarkersFactory;
 import org.apache.hudi.table.storage.HoodieLayoutFactory;
 import org.apache.hudi.table.storage.HoodieStorageLayout;
-
-import org.apache.avro.Schema;
-import org.apache.avro.specific.SpecificRecordBase;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
-import javax.annotation.Nonnull;
-
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -261,19 +258,6 @@ public abstract HoodieWriteMetadata<O> bulkInsertPrepped(HoodieEngineContext con
    */
   public abstract HoodieWriteMetadata<O> insertOverwriteTable(HoodieEngineContext context, String instantTime, I records);
 
-  /**
-   * Updates Metadata Indexes (like Column Stats index)
-   * TODO rebase onto metadata table (post RFC-27)
-   *
-   * @param context instance of {@link HoodieEngineContext}
-   * @param instantTime instant of the carried operation triggering the update
-   */
-  public abstract void updateMetadataIndexes(
-      @Nonnull HoodieEngineContext context,
-      @Nonnull List<HoodieWriteStat> stats,
-      @Nonnull String instantTime
-  ) throws Exception;
-
   public HoodieWriteConfig getConfig() {
     return config;
   }

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java

@@ -34,7 +34,6 @@
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
-import org.apache.hudi.common.model.HoodieWriteStat;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
@@ -63,12 +62,9 @@
 import org.apache.hudi.table.action.commit.FlinkUpsertPreppedCommitActionExecutor;
 import org.apache.hudi.table.action.rollback.BaseRollbackPlanActionExecutor;
 import org.apache.hudi.table.action.rollback.CopyOnWriteRollbackActionExecutor;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.annotation.Nonnull;
-
 import java.io.IOException;
 import java.util.Collections;
 import java.util.Iterator;
@@ -246,11 +242,6 @@ public HoodieWriteMetadata deletePartitions(HoodieEngineContext context, String
     throw new HoodieNotSupportedException("DeletePartitions is not supported yet");
   }
 
-  @Override
-  public void updateMetadataIndexes(@Nonnull HoodieEngineContext context, @Nonnull List<HoodieWriteStat> stats, @Nonnull String instantTime) {
-    throw new HoodieNotSupportedException("update statistics is not supported yet");
-  }
-
   @Override
   public HoodieWriteMetadata<List<WriteStatus>> upsertPrepped(HoodieEngineContext context, String instantTime, List<HoodieRecord<T>> preppedRecords) {
     throw new HoodieNotSupportedException("This method should not be invoked");

hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaCopyOnWriteTable.java

@@ -34,7 +34,6 @@
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
-import org.apache.hudi.common.model.HoodieWriteStat;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
@@ -66,12 +65,9 @@
 import org.apache.hudi.table.action.rollback.CopyOnWriteRollbackActionExecutor;
 import org.apache.hudi.table.action.rollback.RestorePlanActionExecutor;
 import org.apache.hudi.table.action.savepoint.SavepointActionExecutor;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.annotation.Nonnull;
-
 import java.io.IOException;
 import java.util.Collections;
 import java.util.Iterator;
@@ -173,11 +169,6 @@ public HoodieWriteMetadata<List<WriteStatus>> insertOverwriteTable(HoodieEngineC
         context, config, this, instantTime, records).execute();
   }
 
-  @Override
-  public void updateMetadataIndexes(@Nonnull HoodieEngineContext context, @Nonnull List<HoodieWriteStat> stats, @Nonnull String instantTime) {
-    throw new HoodieNotSupportedException("update statistics is not supported yet");
-  }
-
   @Override
   public Option<HoodieCompactionPlan> scheduleCompaction(HoodieEngineContext context,
                                                          String instantTime,

hudi-client/hudi-spark-client/pom.xml

@@ -53,11 +53,6 @@
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-sql_${scala.binary.version}</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.apache.spark</groupId>
-      <artifactId>spark-avro_${scala.binary.version}</artifactId>
-      <scope>provided</scope>
-    </dependency>
 
     <!-- Parquet -->
     <dependency>

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java

@@ -389,9 +389,6 @@ private void completeClustering(HoodieReplaceCommitMetadata metadata,
       finalizeWrite(table, clusteringCommitTime, writeStats);
       // Update table's metadata (table)
       updateTableMetadata(table, metadata, clusteringInstant);
-      // Update tables' metadata indexes
-      // NOTE: This overlaps w/ metadata table (above) and will be reconciled in the future
-      table.updateMetadataIndexes(context, writeStats, clusteringCommitTime);
 
       LOG.info("Committing Clustering " + clusteringCommitTime + ". Finished with result " + metadata);
 

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java

@@ -18,8 +18,6 @@
 
 package org.apache.hudi.table;
 
-import org.apache.hudi.AvroConversionUtils;
-import org.apache.hudi.avro.HoodieAvroUtils;
 import org.apache.hudi.avro.model.HoodieCleanMetadata;
 import org.apache.hudi.avro.model.HoodieCleanerPlan;
 import org.apache.hudi.avro.model.HoodieClusteringPlan;
@@ -38,18 +36,14 @@
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
-import org.apache.hudi.common.model.HoodieWriteStat;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.common.table.TableSchemaResolver;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.exception.HoodieNotSupportedException;
 import org.apache.hudi.exception.HoodieUpsertException;
-import org.apache.hudi.index.columnstats.ColumnStatsIndexHelper;
 import org.apache.hudi.io.HoodieCreateHandle;
 import org.apache.hudi.io.HoodieMergeHandle;
 import org.apache.hudi.io.HoodieSortedMergeHandle;
@@ -78,21 +72,14 @@
 import org.apache.hudi.table.action.rollback.CopyOnWriteRollbackActionExecutor;
 import org.apache.hudi.table.action.rollback.RestorePlanActionExecutor;
 import org.apache.hudi.table.action.savepoint.SavepointActionExecutor;
-
-import org.apache.avro.Schema;
-import org.apache.hadoop.fs.Path;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
-import javax.annotation.Nonnull;
-
 import java.io.IOException;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 /**
  * Implementation of a very heavily read-optimized Hoodie Table where, all data is stored in base files, with
@@ -172,63 +159,6 @@ public HoodieWriteMetadata<HoodieData<WriteStatus>> insertOverwriteTable(HoodieE
     return new SparkInsertOverwriteTableCommitActionExecutor(context, config, this, instantTime, records).execute();
   }
 
-  @Override
-  public void updateMetadataIndexes(@Nonnull HoodieEngineContext context, @Nonnull List<HoodieWriteStat> stats, @Nonnull String instantTime) throws Exception {
-    updateColumnsStatsIndex(context, stats, instantTime);
-  }
-
-  private void updateColumnsStatsIndex(
-      @Nonnull HoodieEngineContext context,
-      @Nonnull List<HoodieWriteStat> updatedFilesStats,
-      @Nonnull String instantTime
-  ) throws Exception {
-    String sortColsList = config.getClusteringSortColumns();
-    String basePath = metaClient.getBasePath();
-    String indexPath = metaClient.getColumnStatsIndexPath();
-
-    List<String> touchedFiles =
-        updatedFilesStats.stream()
-            .map(s -> new Path(basePath, s.getPath()).toString())
-            .collect(Collectors.toList());
-
-    if (touchedFiles.isEmpty() || StringUtils.isNullOrEmpty(sortColsList) || StringUtils.isNullOrEmpty(indexPath)) {
-      return;
-    }
-
-    LOG.info(String.format("Updating column-statistics index table (%s)", indexPath));
-
-    List<String> sortCols = Arrays.stream(sortColsList.split(","))
-        .map(String::trim)
-        .collect(Collectors.toList());
-
-    HoodieSparkEngineContext sparkEngineContext = (HoodieSparkEngineContext)context;
-
-    // Fetch table schema to appropriately construct col-stats index schema
-    Schema tableWriteSchema =
-        HoodieAvroUtils.createHoodieWriteSchema(
-            new TableSchemaResolver(metaClient).getTableAvroSchemaWithoutMetadataFields()
-        );
-
-    List<String> completedCommits =
-        metaClient.getCommitsTimeline()
-            .filterCompletedInstants()
-            .getInstants()
-            .map(HoodieInstant::getTimestamp)
-            .collect(Collectors.toList());
-
-    ColumnStatsIndexHelper.updateColumnStatsIndexFor(
-        sparkEngineContext.getSqlContext().sparkSession(),
-        AvroConversionUtils.convertAvroSchemaToStructType(tableWriteSchema),
-        touchedFiles,
-        sortCols,
-        indexPath,
-        instantTime,
-        completedCommits
-    );
-
-    LOG.info(String.format("Successfully updated column-statistics index at instant (%s)", instantTime));
-  }
-
   @Override
   public Option<HoodieCompactionPlan> scheduleCompaction(HoodieEngineContext context, String instantTime, Option<Map<String, String>> extraMetadata) {
     throw new HoodieNotSupportedException("Compaction is not supported on a CopyOnWrite table");

hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/AvroConversionUtils.scala

@@ -17,13 +17,13 @@
  */
 
 package org.apache.hudi
+
 import org.apache.avro.Schema.Type
 import org.apache.avro.generic.{GenericRecord, GenericRecordBuilder, IndexedRecord}
 import org.apache.avro.{AvroRuntimeException, JsonProperties, Schema}
 import org.apache.hudi.HoodieSparkUtils.sparkAdapter
 import org.apache.hudi.avro.HoodieAvroUtils
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.avro.SchemaConverters
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.encoders.RowEncoder
 import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
@@ -136,27 +136,36 @@ object AvroConversionUtils {
   }
 
   /**
-    *
-    * Returns avro schema from spark StructType.
-    *
-    * @param structType       Dataframe Struct Type.
-    * @param structName       Avro record name.
-    * @param recordNamespace  Avro record namespace.
-    * @return                 Avro schema corresponding to given struct type.
-    */
+   *
+   * Returns avro schema from spark StructType.
+   *
+   * @param structType      Dataframe Struct Type.
+   * @param structName      Avro record name.
+   * @param recordNamespace Avro record namespace.
+   * @return Avro schema corresponding to given struct type.
+   */
   def convertStructTypeToAvroSchema(structType: DataType,
                                     structName: String,
                                     recordNamespace: String): Schema = {
-    getAvroSchemaWithDefaults(SchemaConverters.toAvroType(structType, nullable = false, structName, recordNamespace), structType)
+    val schemaConverters = sparkAdapter.getAvroSchemaConverters
+    val avroSchema = schemaConverters.toAvroType(structType, nullable = false, structName, recordNamespace)
+    getAvroSchemaWithDefaults(avroSchema, structType)
+  }
+
+  def convertAvroSchemaToStructType(avroSchema: Schema): StructType = {
+    val schemaConverters = sparkAdapter.getAvroSchemaConverters
+    schemaConverters.toSqlType(avroSchema) match {
+      case (dataType, _) => dataType.asInstanceOf[StructType]
+    }
   }
 
   /**
-    *
-    * Method to add default value of null to nullable fields in given avro schema
-    *
-    * @param schema     input avro schema
-    * @return           Avro schema with null default set to nullable fields
-    */
+   *
+   * Method to add default value of null to nullable fields in given avro schema
+   *
+   * @param schema input avro schema
+   * @return Avro schema with null default set to nullable fields
+   */
   def getAvroSchemaWithDefaults(schema: Schema, dataType: DataType): Schema = {
 
     schema.getType match {
@@ -205,10 +214,6 @@ object AvroConversionUtils {
     }
   }
 
-  def convertAvroSchemaToStructType(avroSchema: Schema): StructType = {
-    SchemaConverters.toSqlType(avroSchema).dataType.asInstanceOf[StructType]
-  }
-
   def getAvroRecordNameAndNamespace(tableName: String): (String, String) = {
     val name = HoodieAvroUtils.sanitizeName(tableName)
     (s"${name}_record", s"hoodie.${name}")