[HUDI-1234] DO NOT MERGE use fg reader in cdc test

hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java

@@ -37,12 +37,14 @@
 import org.apache.spark.sql.HoodieInternalRowUtils;
 import org.apache.spark.sql.HoodieUnsafeRowUtils;
 import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.catalyst.expressions.UnsafeProjection;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
 import org.apache.spark.sql.types.StructType;
 
 import java.util.Map;
 import java.util.function.UnaryOperator;
 
+import scala.Function1;
+
 import static org.apache.hudi.common.model.HoodieRecord.RECORD_KEY_METADATA_FIELD;
 import static org.apache.hudi.common.model.HoodieRecordMerger.DEFAULT_MERGER_STRATEGY_UUID;
 import static org.apache.spark.sql.HoodieInternalRowUtils.getCachedSchema;
@@ -137,8 +139,14 @@ private Object getFieldValueFromInternalRow(InternalRow row, Schema recordSchema
  }
 
  @Override
- public UnaryOperator<InternalRow> projectRecord(Schema from, Schema to) {
- UnsafeProjection projection = HoodieInternalRowUtils.generateUnsafeProjectionAlias(getCachedSchema(from), getCachedSchema(to));
- return projection::apply;
+ public UnaryOperator<InternalRow> projectRecord(Schema from, Schema to, Map<String, String> renamedColumns) {
+ Function1<InternalRow, UnsafeRow> unsafeRowWriter =
+ HoodieInternalRowUtils.getCachedUnsafeRowWriter(getCachedSchema(from), getCachedSchema(to), renamedColumns);
+ return row -> (InternalRow) unsafeRowWriter.apply(row);
+
+ }
+
+ protected UnaryOperator<InternalRow> getIdentityProjection() {
+ return row -> row;
  }
 }

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

@@ -19,48 +19,59 @@
 
 package org.apache.hudi
 
+import org.apache.avro.Schema
+import org.apache.avro.generic.IndexedRecord
+import org.apache.hadoop.conf.Configuration
+import org.apache.hudi.SparkFileFormatInternalRowReaderContext.getAppliedRequiredSchema
+import org.apache.hudi.avro.AvroSchemaUtils
 import org.apache.hudi.common.engine.HoodieReaderContext
 import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.read.HoodiePositionBasedFileGroupRecordBuffer.ROW_INDEX_TEMPORARY_COLUMN_NAME
 import org.apache.hudi.common.util.ValidationUtils.checkState
-import org.apache.hudi.common.util.collection.{ClosableIterator, CloseableMappingIterator}
+import org.apache.hudi.common.util.collection.{CachingIterator, ClosableIterator, CloseableMappingIterator}
 import org.apache.hudi.io.storage.{HoodieSparkFileReaderFactory, HoodieSparkParquetReader}
-import org.apache.hudi.storage.{HoodieStorage, StoragePath}
+import org.apache.hudi.storage.{HoodieStorage, StorageConfiguration, StoragePath}
 import org.apache.hudi.util.CloseableInternalRowIterator
-
-import org.apache.avro.Schema
-import org.apache.avro.generic.IndexedRecord
 import org.apache.spark.sql.HoodieInternalRowUtils
 import org.apache.spark.sql.avro.HoodieAvroDeserializer
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{JoinedRow, UnsafeProjection, UnsafeRow}
 import org.apache.spark.sql.execution.datasources.PartitionedFile
-import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, SparkParquetReader}
+import org.apache.spark.sql.hudi.SparkAdapter
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.{LongType, MetadataBuilder, StructField, StructType}
 import org.apache.spark.sql.vectorized.{ColumnVector, ColumnarBatch}
 
 import scala.collection.mutable
 
 /**
- * Implementation of {@link HoodieReaderContext} to read {@link InternalRow}s with
- * {@link ParquetFileFormat} on Spark.
+ * Implementation of [[HoodieReaderContext]] to read [[InternalRow]]s with
+ * [[ParquetFileFormat]] on Spark.
  *
  * This uses Spark parquet reader to read parquet data files or parquet log blocks.
  *
- * @param readermaps our intention is to build the reader inside of getFileRecordIterator, but since it is called from
- * the executor, we will need to port a bunch of the code from ParquetFileFormat for each spark version
- * for now, we pass in a map of the different readers we expect to create
+ * @param parquetFileReader A reader that transforms a [[PartitionedFile]] to an iterator of
+ * [[InternalRow]]. This is required for reading the base file and
+ * not required for reading a file group with only log files.
+ * @param recordKeyColumn column name for the recordkey
+ * @param filters spark filters that might be pushed down into the reader
  */
-class SparkFileFormatInternalRowReaderContext(readerMaps: mutable.Map[Long, PartitionedFile => Iterator[InternalRow]]) extends BaseSparkInternalRowReaderContext {
- lazy val sparkAdapter = SparkAdapterSupport.sparkAdapter
- val deserializerMap: mutable.Map[Schema, HoodieAvroDeserializer] = mutable.Map()
-
- override def getFileRecordIterator(filePath: StoragePath, start: Long, length: Long, dataSchema: Schema, requiredSchema: Schema, storage: HoodieStorage): ClosableIterator[InternalRow] = {
- // partition value is empty because the spark parquet reader will append the partition columns to
- // each row if they are given. That is the only usage of the partition values in the reader.
- val fileInfo = sparkAdapter.getSparkPartitionedFileUtils
- .createPartitionedFile(InternalRow.empty, filePath, start, length)
+class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetReader,
+ recordKeyColumn: String,
+ filters: Seq[Filter]) extends BaseSparkInternalRowReaderContext {
+ lazy val sparkAdapter: SparkAdapter = SparkAdapterSupport.sparkAdapter
+ lazy val recordKeyFilters: Seq[Filter] = filters.filter(f => f.references.exists(c => c.equalsIgnoreCase(recordKeyColumn)))
+ private val deserializerMap: mutable.Map[Schema, HoodieAvroDeserializer] = mutable.Map()
+
+ override def getFileRecordIterator(filePath: StoragePath,
+ start: Long,
+ length: Long,
+ dataSchema: Schema,
+ requiredSchema: Schema,
+ storage: HoodieStorage): ClosableIterator[InternalRow] = {
+ val structType: StructType = HoodieInternalRowUtils.getCachedSchema(requiredSchema)
  if (FSUtils.isLogFile(filePath)) {
- val structType: StructType = HoodieInternalRowUtils.getCachedSchema(requiredSchema)
  val projection: UnsafeProjection = HoodieInternalRowUtils.getCachedUnsafeProjection(structType, structType)
  new CloseableMappingIterator[InternalRow, UnsafeRow](
  new HoodieSparkFileReaderFactory(storage).newParquetFileReader(filePath)
@@ -73,16 +84,27 @@ class SparkFileFormatInternalRowReaderContext(readerMaps: mutable.Map[Long, Part
  }
  }).asInstanceOf[ClosableIterator[InternalRow]]
  } else {
- val schemaPairHashKey = generateSchemaPairHashKey(dataSchema, requiredSchema)
- if (!readerMaps.contains(schemaPairHashKey)) {
- throw new IllegalStateException("schemas don't hash to a known reader")
- }
- new CloseableInternalRowIterator(readerMaps(schemaPairHashKey).apply(fileInfo))
+ // partition value is empty because the spark parquet reader will append the partition columns to
+ // each row if they are given. That is the only usage of the partition values in the reader.
+ val fileInfo = sparkAdapter.getSparkPartitionedFileUtils
+ .createPartitionedFile(InternalRow.empty, filePath, start, length)
+ val (readSchema, readFilters) = getSchemaAndFiltersForRead(structType)
+ new CloseableInternalRowIterator(parquetFileReader.read(fileInfo,
+ readSchema, StructType(Seq.empty), readFilters, storage.getConf.asInstanceOf[StorageConfiguration[Configuration]]))
  }
  }
 
- private def generateSchemaPairHashKey(dataSchema: Schema, requestedSchema: Schema): Long = {
- dataSchema.hashCode() + requestedSchema.hashCode()
+ private def getSchemaAndFiltersForRead(structType: StructType): (StructType, Seq[Filter]) = {
+ (getHasLogFiles, getNeedsBootstrapMerge, getUseRecordPosition) match {
+ case (false, false, _) =>
+ (structType, filters)
+ case (false, true, true) =>
+ (getAppliedRequiredSchema(structType), filters)
+ case (true, _, true) =>
+ (getAppliedRequiredSchema(structType), recordKeyFilters)
+ case (_, _, _) =>
+ (structType, Seq.empty)
+ }
  }
 
  /**
@@ -101,46 +123,147 @@ class SparkFileFormatInternalRowReaderContext(readerMaps: mutable.Map[Long, Part
  }
 
  override def mergeBootstrapReaders(skeletonFileIterator: ClosableIterator[InternalRow],
- dataFileIterator: ClosableIterator[InternalRow]): ClosableIterator[InternalRow] = {
- doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]],
- dataFileIterator.asInstanceOf[ClosableIterator[Any]])
+ skeletonRequiredSchema: Schema,
+ dataFileIterator: ClosableIterator[InternalRow],
+ dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
+ doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]], skeletonRequiredSchema,
+ dataFileIterator.asInstanceOf[ClosableIterator[Any]], dataRequiredSchema)
  }
 
- protected def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any], dataFileIterator: ClosableIterator[Any]): ClosableIterator[InternalRow] = {
- new ClosableIterator[Any] {
- val combinedRow = new JoinedRow()
-
- override def hasNext: Boolean = {
- //If the iterators are out of sync it is probably due to filter pushdown
- checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
- "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
- dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ private def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any],
+ skeletonRequiredSchema: Schema,
+ dataFileIterator: ClosableIterator[Any],
+ dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
+ if (getUseRecordPosition) {
+ assert(AvroSchemaUtils.containsFieldInSchema(skeletonRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ assert(AvroSchemaUtils.containsFieldInSchema(dataRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ val javaSet = new java.util.HashSet[String]()
+ javaSet.add(ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ val skeletonProjection = projectRecord(skeletonRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(skeletonRequiredSchema, javaSet))
+ //If we have log files, we will want to do position based merging with those as well,
+ //so leave the row index column at the end
+ val dataProjection = if (getHasLogFiles) {
+ getIdentityProjection
+ } else {
+ projectRecord(dataRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(dataRequiredSchema, javaSet))
  }
 
- override def next(): Any = {
- (skeletonFileIterator.next(), dataFileIterator.next()) match {
- case (s: ColumnarBatch, d: ColumnarBatch) =>
- val numCols = s.numCols() + d.numCols()
- val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
- for (i <- 0 until numCols) {
- if (i < s.numCols()) {
- vecs(i) = s.column(i)
+ //Always use internal row for positional merge because
+ //we need to iterate row by row when merging
+ new CachingIterator[InternalRow] {
+ val combinedRow = new JoinedRow()
+
+ //position column will always be at the end of the row
+ private def getPos(row: InternalRow): Long = {
+ row.getLong(row.numFields-1)
+ }
+
+ private def getNextSkeleton: (InternalRow, Long) = {
+ val nextSkeletonRow = skeletonFileIterator.next().asInstanceOf[InternalRow]
+ (nextSkeletonRow, getPos(nextSkeletonRow))
+ }
+
+ private def getNextData: (InternalRow, Long) = {
+ val nextSkeletonRow = skeletonFileIterator.next().asInstanceOf[InternalRow]
+ (nextSkeletonRow, getPos(nextSkeletonRow))
+ }
+
+ override def close(): Unit = {
+ skeletonFileIterator.close()
+ dataFileIterator.close()
+ }
+
+ override protected def doHasNext(): Boolean = {
+ if (!dataFileIterator.hasNext || !skeletonFileIterator.hasNext) {
+ false
+ } else {
+ var nextSkeleton = getNextSkeleton
+ var nextData = getNextData
+ while (nextSkeleton._2 != nextData._2) {
+ if (nextSkeleton._2 > nextData._2) {
+ if (!dataFileIterator.hasNext) {
+ return false
+ } else {
+ nextData = getNextData
+ }
  } else {
- vecs(i) = d.column(i - s.numCols())
+ if (!skeletonFileIterator.hasNext) {
+ return false
+ } else {
+ nextSkeleton = getNextSkeleton
+ }
  }
  }
- assert(s.numRows() == d.numRows())
- sparkAdapter.makeColumnarBatch(vecs, s.numRows())
- case (_: ColumnarBatch, _: InternalRow) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
- case (_: InternalRow, _: ColumnarBatch) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
- case (s: InternalRow, d: InternalRow) => combinedRow(s, d)
+ nextRecord = combinedRow(skeletonProjection.apply(nextSkeleton._1), dataProjection.apply(nextData._1))
+ true
+ }
  }
  }
+ } else {
+ new ClosableIterator[Any] {
+ val combinedRow = new JoinedRow()
 
- override def close(): Unit = {
- skeletonFileIterator.close()
- dataFileIterator.close()
- }
- }.asInstanceOf[ClosableIterator[InternalRow]]
+ override def hasNext: Boolean = {
+ //If the iterators are out of sync it is probably due to filter pushdown
+ checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
+ "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
+ dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ }
+
+ override def next(): Any = {
+ (skeletonFileIterator.next(), dataFileIterator.next()) match {
+ case (s: ColumnarBatch, d: ColumnarBatch) =>
+ //This will not be used until [HUDI-7693] is implemented
+ val numCols = s.numCols() + d.numCols()
+ val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
+ for (i <- 0 until numCols) {
+ if (i < s.numCols()) {
+ vecs(i) = s.column(i)
+ } else {
+ vecs(i) = d.column(i - s.numCols())
+ }
+ }
+ assert(s.numRows() == d.numRows())
+ sparkAdapter.makeColumnarBatch(vecs, s.numRows())
+ case (_: ColumnarBatch, _: InternalRow) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
+ case (_: InternalRow, _: ColumnarBatch) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
+ case (s: InternalRow, d: InternalRow) => combinedRow(s, d)
+ }
+ }
+
+ override def close(): Unit = {
+ skeletonFileIterator.close()
+ dataFileIterator.close()
+ }
+ }.asInstanceOf[ClosableIterator[InternalRow]]
+ }
  }
 }
+
+object SparkFileFormatInternalRowReaderContext {
+ // From "namedExpressions.scala": Used to construct to record position field metadata.
+ private val FILE_SOURCE_GENERATED_METADATA_COL_ATTR_KEY = "__file_source_generated_metadata_col"
+ private val FILE_SOURCE_METADATA_COL_ATTR_KEY = "__file_source_metadata_col"
+ private val METADATA_COL_ATTR_KEY = "__metadata_col"
+
+ def getRecordKeyRelatedFilters(filters: Seq[Filter], recordKeyColumn: String): Seq[Filter] = {
+ filters.filter(f => f.references.exists(c => c.equalsIgnoreCase(recordKeyColumn)))
+ }
+
+ def getAppliedRequiredSchema(requiredSchema: StructType): StructType = {
+ val metadata = new MetadataBuilder()
+ .putString(METADATA_COL_ATTR_KEY, ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ .putBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY, value = true)
+ .putString(FILE_SOURCE_GENERATED_METADATA_COL_ATTR_KEY, ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ .build()
+ val rowIndexField = StructField(ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType, nullable = false, metadata)
+ StructType(requiredSchema.fields.filterNot(isIndexTempColumn) :+ rowIndexField)
+ }
+
+ private def isIndexTempColumn(field: StructField): Boolean = {
+ field.name.equals(ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ }
+
+}

hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieInternalRowUtils.scala

@@ -75,13 +75,6 @@ object HoodieInternalRowUtils {
  .getOrElseUpdate((from, to), generateUnsafeProjection(from, to))
  }
 
- /**
- * due to scala2.11 and HoodieCatalystExpressionUtils is both an object and trait,
- * we can't directly call generateUnsafeProjection from java code
- */
- def generateUnsafeProjectionAlias(from: StructType, to: StructType): UnsafeProjection = {
- generateUnsafeProjection(from, to)
- }
  /**
  * Provides cached instance of [[UnsafeRowWriter]] transforming provided [[InternalRow]]s from
  * one [[StructType]] and into another [[StructType]]

hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkParquetReader.scala

@@ -20,6 +20,7 @@
 package org.apache.spark.sql.execution.datasources.parquet
 
 import org.apache.hadoop.conf.Configuration
+import org.apache.hudi.storage.StorageConfiguration
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.execution.datasources.PartitionedFile
 import org.apache.spark.sql.sources.Filter
@@ -33,12 +34,12 @@ trait SparkParquetReader extends Serializable {
  * @param requiredSchema desired output schema of the data
  * @param partitionSchema schema of the partition columns. Partition values will be appended to the end of every row
  * @param filters filters for data skipping. Not guaranteed to be used; the spark plan will also apply the filters.
- * @param sharedConf  the hadoop conf
+ * @param storageConf the hadoop conf
  * @return iterator of rows read from the file output type says [[InternalRow]] but could be [[ColumnarBatch]]
  */
  def read(file: PartitionedFile,
  requiredSchema: StructType,
  partitionSchema: StructType,
  filters: Seq[Filter],
- sharedConf: Configuration): Iterator[InternalRow]
+ storageConf: StorageConfiguration[Configuration]): Iterator[InternalRow]
 }