You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "nsivabalan (via GitHub)" <gi...@apache.org> on 2023/06/09 00:25:07 UTC
[GitHub] [hudi] nsivabalan commented on a diff in pull request #8879: [HUDI-6315] [WIP] Optimize UPSERT codepath to use meta fields instead of key generation and index lookup
nsivabalan commented on code in PR #8879:
URL: https://github.com/apache/hudi/pull/8879#discussion_r1222275056
##########
hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java:
##########
@@ -233,15 +238,25 @@ public static HoodieWriteResult doDeletePartitionsOperation(SparkRDDWriteClient
}
public static HoodieRecord createHoodieRecord(GenericRecord gr, Comparable orderingVal, HoodieKey hKey,
- String payloadClass) throws IOException {
+ String payloadClass, HoodieRecordLocation recordLocation) throws IOException {
HoodieRecordPayload payload = DataSourceUtils.createPayload(payloadClass, gr, orderingVal);
- return new HoodieAvroRecord<>(hKey, payload);
+
+ HoodieAvroRecord record = new HoodieAvroRecord<>(hKey, payload);
+ if (recordLocation != null) {
+ record.setCurrentLocation(recordLocation);
+ }
+ return record;
}
+ // AKL_TODO: check if this change is needed. Also validate change if needed.
public static HoodieRecord createHoodieRecord(GenericRecord gr, HoodieKey hKey,
- String payloadClass) throws IOException {
+ String payloadClass, HoodieRecordLocation recordLocation) throws IOException {
Review Comment:
same here
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala:
##########
@@ -144,20 +144,25 @@ class DefaultSource extends RelationProvider
mode: SaveMode,
optParams: Map[String, String],
df: DataFrame): BaseRelation = {
- val dfWithoutMetaCols = df.drop(HoodieRecord.HOODIE_META_COLUMNS.asScala:_*)
+ val dfPrepped = if (optParams.getOrDefault(DATASOURCE_WRITE_PREPPED_KEY, "false")
Review Comment:
dfPrepped -> processedDf
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -1160,21 +1171,29 @@ object HoodieSparkSqlWriter {
// handle dropping partition columns
it.map { avroRec =>
- val processedRecord = if (shouldDropPartitionColumns) {
- HoodieAvroUtils.rewriteRecord(avroRec, dataFileSchema)
+ val (hoodieKey: HoodieKey, recordLocation: Option[HoodieRecordLocation]) = getKeyAndLocatorFromAvroRecord(keyGenerator, avroRec,
+ isPrepped)
+
+ val avroRecWithoutMeta: GenericRecord = if (isPrepped) {
+ HoodieAvroUtils.rewriteRecord(avroRec, HoodieAvroUtils.removeMetadataFields(dataFileSchema))
} else {
avroRec
}
- val hoodieKey = new HoodieKey(keyGenerator.getRecordKey(avroRec), keyGenerator.getPartitionPath(avroRec))
+ val processedRecord = if (shouldDropPartitionColumns) {
+ HoodieAvroUtils.rewriteRecord(avroRecWithoutMeta, dataFileSchema)
+ } else {
+ avroRecWithoutMeta
+ }
+
val hoodieRecord = if (shouldCombine) {
val orderingVal = HoodieAvroUtils.getNestedFieldVal(avroRec, config.getString(PRECOMBINE_FIELD),
false, consistentLogicalTimestampEnabled).asInstanceOf[Comparable[_]]
DataSourceUtils.createHoodieRecord(processedRecord, orderingVal, hoodieKey,
- config.getString(PAYLOAD_CLASS_NAME))
+ config.getString(PAYLOAD_CLASS_NAME), recordLocation.getOrElse(null))
} else {
- DataSourceUtils.createHoodieRecord(processedRecord, hoodieKey,
- config.getString(PAYLOAD_CLASS_NAME))
+ // AKL_TODO: check if this change is needed.
Review Comment:
fix the comments
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -1195,18 +1214,108 @@ object HoodieSparkSqlWriter {
}
val sparkKeyGenerator = HoodieSparkKeyGeneratorFactory.createKeyGenerator(keyGenProps).asInstanceOf[SparkKeyGeneratorInterface]
val targetStructType = if (shouldDropPartitionColumns) dataFileStructType else writerStructType
+ val finalStructType = if (isPrepped) {
+ val fieldsToExclude = HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION.toArray()
+ StructType(targetStructType.fields.filterNot(field => fieldsToExclude.contains(field.name)))
+ } else {
+ targetStructType
+ }
// NOTE: To make sure we properly transform records
- val targetStructTypeRowWriter = getCachedUnsafeRowWriter(sourceStructType, targetStructType)
+ val finalStructTypeRowWriter = getCachedUnsafeRowWriter(sourceStructType, finalStructType)
it.map { sourceRow =>
- val recordKey = sparkKeyGenerator.getRecordKey(sourceRow, sourceStructType)
- val partitionPath = sparkKeyGenerator.getPartitionPath(sourceRow, sourceStructType)
- val key = new HoodieKey(recordKey.toString, partitionPath.toString)
- val targetRow = targetStructTypeRowWriter(sourceRow)
+ val (hoodieKey: HoodieKey, recordLocation: Option[HoodieRecordLocation]) = getKeyAndLocatorFromSparkRecord(sparkKeyGenerator, sourceRow, sourceStructType, isPrepped)
- new HoodieSparkRecord(key, targetRow, dataFileStructType, false)
+ val finalRow = finalStructTypeRowWriter(sourceRow)
+ var hoodieSparkRecord = new HoodieSparkRecord(hoodieKey, finalRow, dataFileStructType, false)
+ if (recordLocation.getOrElse(null) != null) {
Review Comment:
lets use Option
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -1195,18 +1214,108 @@ object HoodieSparkSqlWriter {
}
val sparkKeyGenerator = HoodieSparkKeyGeneratorFactory.createKeyGenerator(keyGenProps).asInstanceOf[SparkKeyGeneratorInterface]
val targetStructType = if (shouldDropPartitionColumns) dataFileStructType else writerStructType
+ val finalStructType = if (isPrepped) {
+ val fieldsToExclude = HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION.toArray()
+ StructType(targetStructType.fields.filterNot(field => fieldsToExclude.contains(field.name)))
+ } else {
+ targetStructType
+ }
// NOTE: To make sure we properly transform records
- val targetStructTypeRowWriter = getCachedUnsafeRowWriter(sourceStructType, targetStructType)
+ val finalStructTypeRowWriter = getCachedUnsafeRowWriter(sourceStructType, finalStructType)
it.map { sourceRow =>
- val recordKey = sparkKeyGenerator.getRecordKey(sourceRow, sourceStructType)
- val partitionPath = sparkKeyGenerator.getPartitionPath(sourceRow, sourceStructType)
- val key = new HoodieKey(recordKey.toString, partitionPath.toString)
- val targetRow = targetStructTypeRowWriter(sourceRow)
+ val (hoodieKey: HoodieKey, recordLocation: Option[HoodieRecordLocation]) = getKeyAndLocatorFromSparkRecord(sparkKeyGenerator, sourceRow, sourceStructType, isPrepped)
- new HoodieSparkRecord(key, targetRow, dataFileStructType, false)
+ val finalRow = finalStructTypeRowWriter(sourceRow)
+ var hoodieSparkRecord = new HoodieSparkRecord(hoodieKey, finalRow, dataFileStructType, false)
+ if (recordLocation.getOrElse(null) != null) {
+ hoodieSparkRecord.setCurrentLocation(recordLocation.get)
+ }
+ hoodieSparkRecord
}
}.toJavaRDD().asInstanceOf[JavaRDD[HoodieRecord[_]]]
}
}
+
+ private def getKeyAndLocatorFromAvroRecord(keyGenerator: BaseKeyGenerator, avroRec: GenericRecord,
Review Comment:
getHoodieKeyAndMayBeLocation
##########
hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/bootstrap/SparkFullBootstrapDataProviderBase.java:
##########
@@ -84,7 +84,7 @@ public JavaRDD<HoodieRecord> generateInputRecords(String tableName, String sourc
Boolean.parseBoolean(KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.defaultValue())));
try {
return DataSourceUtils.createHoodieRecord(gr, orderingVal, keyGenerator.getKey(gr),
- props.getString("hoodie.datasource.write.payload.class"));
+ props.getString("hoodie.datasource.write.payload.class"), null);
Review Comment:
why is this change ?
##########
hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java:
##########
@@ -233,15 +238,25 @@ public static HoodieWriteResult doDeletePartitionsOperation(SparkRDDWriteClient
}
public static HoodieRecord createHoodieRecord(GenericRecord gr, Comparable orderingVal, HoodieKey hKey,
- String payloadClass) throws IOException {
+ String payloadClass, HoodieRecordLocation recordLocation) throws IOException {
Review Comment:
Lets do Option<HoodieRecordLocation>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala:
##########
@@ -144,20 +144,25 @@ class DefaultSource extends RelationProvider
mode: SaveMode,
optParams: Map[String, String],
df: DataFrame): BaseRelation = {
- val dfWithoutMetaCols = df.drop(HoodieRecord.HOODIE_META_COLUMNS.asScala:_*)
+ val dfPrepped = if (optParams.getOrDefault(DATASOURCE_WRITE_PREPPED_KEY, "false")
Review Comment:
or we can rename the argument (L146) to unProcessedDf and so we can rename dfPrepped -> df and so rest of the code is untouched.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -1195,18 +1214,108 @@ object HoodieSparkSqlWriter {
}
val sparkKeyGenerator = HoodieSparkKeyGeneratorFactory.createKeyGenerator(keyGenProps).asInstanceOf[SparkKeyGeneratorInterface]
val targetStructType = if (shouldDropPartitionColumns) dataFileStructType else writerStructType
+ val finalStructType = if (isPrepped) {
+ val fieldsToExclude = HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION.toArray()
Review Comment:
lets ensure we drop 6 meta fields across both avro and spark record type code paths
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -1195,18 +1214,108 @@ object HoodieSparkSqlWriter {
}
val sparkKeyGenerator = HoodieSparkKeyGeneratorFactory.createKeyGenerator(keyGenProps).asInstanceOf[SparkKeyGeneratorInterface]
val targetStructType = if (shouldDropPartitionColumns) dataFileStructType else writerStructType
+ val finalStructType = if (isPrepped) {
+ val fieldsToExclude = HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION.toArray()
+ StructType(targetStructType.fields.filterNot(field => fieldsToExclude.contains(field.name)))
+ } else {
+ targetStructType
+ }
// NOTE: To make sure we properly transform records
- val targetStructTypeRowWriter = getCachedUnsafeRowWriter(sourceStructType, targetStructType)
+ val finalStructTypeRowWriter = getCachedUnsafeRowWriter(sourceStructType, finalStructType)
it.map { sourceRow =>
- val recordKey = sparkKeyGenerator.getRecordKey(sourceRow, sourceStructType)
- val partitionPath = sparkKeyGenerator.getPartitionPath(sourceRow, sourceStructType)
- val key = new HoodieKey(recordKey.toString, partitionPath.toString)
- val targetRow = targetStructTypeRowWriter(sourceRow)
+ val (hoodieKey: HoodieKey, recordLocation: Option[HoodieRecordLocation]) = getKeyAndLocatorFromSparkRecord(sparkKeyGenerator, sourceRow, sourceStructType, isPrepped)
- new HoodieSparkRecord(key, targetRow, dataFileStructType, false)
+ val finalRow = finalStructTypeRowWriter(sourceRow)
+ var hoodieSparkRecord = new HoodieSparkRecord(hoodieKey, finalRow, dataFileStructType, false)
+ if (recordLocation.getOrElse(null) != null) {
+ hoodieSparkRecord.setCurrentLocation(recordLocation.get)
+ }
+ hoodieSparkRecord
}
}.toJavaRDD().asInstanceOf[JavaRDD[HoodieRecord[_]]]
}
}
+
+ private def getKeyAndLocatorFromAvroRecord(keyGenerator: BaseKeyGenerator, avroRec: GenericRecord,
+ isPrepped: Boolean): (HoodieKey, Option[HoodieRecordLocation]) = {
+ val recordKey = if (isPrepped) {
+ avroRec.get("_hoodie_record_key").toString
+ } else {
+ keyGenerator.getRecordKey(avroRec)
+ };
+
+ val partitionPath = if (isPrepped) {
+ avroRec.get("_hoodie_partition_path").toString
+ } else {
+ keyGenerator.getPartitionPath(avroRec)
+ };
+
+ val hoodieKey = new HoodieKey(recordKey, partitionPath)
+ val instantTime: Option[String] = if (isPrepped) {
+ Option(avroRec.get("_hoodie_commit_time")).map(_.toString) }
+ else {
+ None
+ }
+ val fileName: Option[String] = if (isPrepped) {
+ Option(avroRec.get("_hoodie_file_name")).map(_.toString) }
+ else {
+ None
+ }
+ val recordLocation: Option[HoodieRecordLocation] = if (instantTime.isDefined && fileName.isDefined) {
+ val fileId = FSUtils.getFileId(fileName.get)
+ Some(new HoodieRecordLocation(instantTime.get, fileId))
+ } else {
+ None
+ }
+ (hoodieKey, recordLocation)
+ }
+
+ private def getKeyAndLocatorFromSparkRecord(sparkKeyGenerator: SparkKeyGeneratorInterface, sourceRow: InternalRow,
+ schema: StructType, isPrepped: Boolean): (HoodieKey, Option[HoodieRecordLocation]) = {
+ def getFieldIndex(fieldName: String): Int = {
+ if (schema.fieldNames.contains(fieldName)) {
Review Comment:
HoodieRecord already has the name to position mapping
HOODIE_META_COLUMNS_NAME_TO_POS
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -1195,18 +1214,108 @@ object HoodieSparkSqlWriter {
}
val sparkKeyGenerator = HoodieSparkKeyGeneratorFactory.createKeyGenerator(keyGenProps).asInstanceOf[SparkKeyGeneratorInterface]
val targetStructType = if (shouldDropPartitionColumns) dataFileStructType else writerStructType
+ val finalStructType = if (isPrepped) {
+ val fieldsToExclude = HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION.toArray()
+ StructType(targetStructType.fields.filterNot(field => fieldsToExclude.contains(field.name)))
+ } else {
+ targetStructType
+ }
// NOTE: To make sure we properly transform records
- val targetStructTypeRowWriter = getCachedUnsafeRowWriter(sourceStructType, targetStructType)
+ val finalStructTypeRowWriter = getCachedUnsafeRowWriter(sourceStructType, finalStructType)
it.map { sourceRow =>
- val recordKey = sparkKeyGenerator.getRecordKey(sourceRow, sourceStructType)
- val partitionPath = sparkKeyGenerator.getPartitionPath(sourceRow, sourceStructType)
- val key = new HoodieKey(recordKey.toString, partitionPath.toString)
- val targetRow = targetStructTypeRowWriter(sourceRow)
+ val (hoodieKey: HoodieKey, recordLocation: Option[HoodieRecordLocation]) = getKeyAndLocatorFromSparkRecord(sparkKeyGenerator, sourceRow, sourceStructType, isPrepped)
- new HoodieSparkRecord(key, targetRow, dataFileStructType, false)
+ val finalRow = finalStructTypeRowWriter(sourceRow)
+ var hoodieSparkRecord = new HoodieSparkRecord(hoodieKey, finalRow, dataFileStructType, false)
+ if (recordLocation.getOrElse(null) != null) {
+ hoodieSparkRecord.setCurrentLocation(recordLocation.get)
+ }
+ hoodieSparkRecord
}
}.toJavaRDD().asInstanceOf[JavaRDD[HoodieRecord[_]]]
}
}
+
+ private def getKeyAndLocatorFromAvroRecord(keyGenerator: BaseKeyGenerator, avroRec: GenericRecord,
+ isPrepped: Boolean): (HoodieKey, Option[HoodieRecordLocation]) = {
+ val recordKey = if (isPrepped) {
+ avroRec.get("_hoodie_record_key").toString
+ } else {
+ keyGenerator.getRecordKey(avroRec)
+ };
+
+ val partitionPath = if (isPrepped) {
+ avroRec.get("_hoodie_partition_path").toString
+ } else {
+ keyGenerator.getPartitionPath(avroRec)
+ };
+
+ val hoodieKey = new HoodieKey(recordKey, partitionPath)
+ val instantTime: Option[String] = if (isPrepped) {
+ Option(avroRec.get("_hoodie_commit_time")).map(_.toString) }
+ else {
+ None
+ }
+ val fileName: Option[String] = if (isPrepped) {
+ Option(avroRec.get("_hoodie_file_name")).map(_.toString) }
+ else {
+ None
+ }
+ val recordLocation: Option[HoodieRecordLocation] = if (instantTime.isDefined && fileName.isDefined) {
+ val fileId = FSUtils.getFileId(fileName.get)
+ Some(new HoodieRecordLocation(instantTime.get, fileId))
+ } else {
+ None
+ }
+ (hoodieKey, recordLocation)
+ }
+
+ private def getKeyAndLocatorFromSparkRecord(sparkKeyGenerator: SparkKeyGeneratorInterface, sourceRow: InternalRow,
+ schema: StructType, isPrepped: Boolean): (HoodieKey, Option[HoodieRecordLocation]) = {
+ def getFieldIndex(fieldName: String): Int = {
+ if (schema.fieldNames.contains(fieldName)) {
+ schema.fieldIndex(fieldName)
+ } else {
+ -1
+ }
+ }
+
+ val hoodieRecordKeyIndex = getFieldIndex("_hoodie_record_key")
Review Comment:
lets re-use the const variable and avoid hard coding
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -1195,18 +1214,108 @@ object HoodieSparkSqlWriter {
}
val sparkKeyGenerator = HoodieSparkKeyGeneratorFactory.createKeyGenerator(keyGenProps).asInstanceOf[SparkKeyGeneratorInterface]
val targetStructType = if (shouldDropPartitionColumns) dataFileStructType else writerStructType
+ val finalStructType = if (isPrepped) {
+ val fieldsToExclude = HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION.toArray()
+ StructType(targetStructType.fields.filterNot(field => fieldsToExclude.contains(field.name)))
+ } else {
+ targetStructType
+ }
// NOTE: To make sure we properly transform records
- val targetStructTypeRowWriter = getCachedUnsafeRowWriter(sourceStructType, targetStructType)
+ val finalStructTypeRowWriter = getCachedUnsafeRowWriter(sourceStructType, finalStructType)
it.map { sourceRow =>
- val recordKey = sparkKeyGenerator.getRecordKey(sourceRow, sourceStructType)
- val partitionPath = sparkKeyGenerator.getPartitionPath(sourceRow, sourceStructType)
- val key = new HoodieKey(recordKey.toString, partitionPath.toString)
- val targetRow = targetStructTypeRowWriter(sourceRow)
+ val (hoodieKey: HoodieKey, recordLocation: Option[HoodieRecordLocation]) = getKeyAndLocatorFromSparkRecord(sparkKeyGenerator, sourceRow, sourceStructType, isPrepped)
- new HoodieSparkRecord(key, targetRow, dataFileStructType, false)
+ val finalRow = finalStructTypeRowWriter(sourceRow)
+ var hoodieSparkRecord = new HoodieSparkRecord(hoodieKey, finalRow, dataFileStructType, false)
+ if (recordLocation.getOrElse(null) != null) {
+ hoodieSparkRecord.setCurrentLocation(recordLocation.get)
+ }
+ hoodieSparkRecord
}
}.toJavaRDD().asInstanceOf[JavaRDD[HoodieRecord[_]]]
}
}
+
+ private def getKeyAndLocatorFromAvroRecord(keyGenerator: BaseKeyGenerator, avroRec: GenericRecord,
+ isPrepped: Boolean): (HoodieKey, Option[HoodieRecordLocation]) = {
+ val recordKey = if (isPrepped) {
+ avroRec.get("_hoodie_record_key").toString
Review Comment:
can we reference the static variables.
HoodieRecord.RECORD_KEY_METADATA_FIELD
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala:
##########
@@ -144,20 +144,25 @@ class DefaultSource extends RelationProvider
mode: SaveMode,
optParams: Map[String, String],
df: DataFrame): BaseRelation = {
- val dfWithoutMetaCols = df.drop(HoodieRecord.HOODIE_META_COLUMNS.asScala:_*)
+ val dfPrepped = if (optParams.getOrDefault(DATASOURCE_WRITE_PREPPED_KEY, "false")
+ .equalsIgnoreCase("true")) {
+ df // Don't remove meta columns for prepped write.
+ } else {
+ df.drop(HoodieRecord.HOODIE_META_COLUMNS.asScala: _*)
Review Comment:
do you know what happens if we drop a non existant column?
we have another optional meta field named operation(_hoodie_operation).
HOODIE_META_COLUMNS_WITH_OPERATION
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala:
##########
@@ -55,8 +55,8 @@ trait ProvidesHoodieConfig extends Logging {
// TODO(HUDI-3456) clean up
val preCombineField = Option(tableConfig.getPreCombineField).getOrElse("")
- require(hoodieCatalogTable.primaryKeys.nonEmpty,
- s"There are no primary key in table ${hoodieCatalogTable.table.identifier}, cannot execute update operator")
+// require(hoodieCatalogTable.primaryKeys.nonEmpty,
Review Comment:
lets fix unintended changes or remove commented out code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org