shangxinli commented on code in PR #18990:
URL: https://github.com/apache/hudi/pull/18990#discussion_r4425871145
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/HoodieDatasetBulkInsertHelper.scala:
##########
@@ -146,6 +188,84 @@ object HoodieDatasetBulkInsertHelper
partitioner.repartitionRecords(updatedDF, targetParallelism)
}
+ /**
+ * Builds the record-key / partition-path projection columns for the non
auto-keygen path,
+ * dispatching to a fast-path tier when the keygen is amenable to a pure
Catalyst projection.
+ *
+ * Tier 1: NonpartitionedKeyGenerator with a single record-key field.
+ * Tier 2: SimpleKeyGenerator with a single record-key + single
partition-path field, when the
+ * partition formatter flags are reproducible as Catalyst
expressions. Currently:
+ * default flags, or hive-style partitioning. URL-encoded and
slash-separated date
+ * partitioning fall through to Tier 3 -- the URL escape table
doesn't have an efficient
+ * pure-Catalyst equivalent, and slash-separated dates were added in
1.2.0 and exercise
+ * a separate formatter branch we'd rather not encode twice.
+ * Tier 3: Anonymous UDF that calls into the supplied
[[BuiltinKeyGenerator]].
+ */
+ private def buildKeygenColumns(df: DataFrame,
+ keyGeneratorClassName: String,
+ typedProps: TypedProperties,
+ keyGenerator: BuiltinKeyGenerator):
(org.apache.spark.sql.Column, org.apache.spark.sql.Column) = {
+ val recordKeyFields = keyGenerator.getRecordKeyFieldNames
+ val partitionPathFields = keyGenerator.getPartitionPathFields
+
+ val hiveStyle =
typedProps.getBoolean(KeyGeneratorOptions.HIVE_STYLE_PARTITIONING_ENABLE.key,
+
KeyGeneratorOptions.HIVE_STYLE_PARTITIONING_ENABLE.defaultValue.toBoolean)
+ val urlEncode =
typedProps.getBoolean(KeyGeneratorOptions.URL_ENCODE_PARTITIONING.key,
+ KeyGeneratorOptions.URL_ENCODE_PARTITIONING.defaultValue.toBoolean)
+ val slashSep =
typedProps.getBoolean(KeyGeneratorOptions.SLASH_SEPARATED_DATE_PARTITIONING.key,
+
KeyGeneratorOptions.SLASH_SEPARATED_DATE_PARTITIONING.defaultValue.toBoolean)
+
+ val tier1 = keyGeneratorClassName ==
classOf[NonpartitionedKeyGenerator].getName &&
+ recordKeyFields.size == 1
+
+ val tier2 = keyGeneratorClassName == classOf[SimpleKeyGenerator].getName &&
+ recordKeyFields.size == 1 && partitionPathFields.size == 1 &&
+ !urlEncode && !slashSep
+
+ if (tier1) {
+ (functions.col(recordKeyFields.get(0)).cast(StringType),
+ functions.lit("").cast(StringType))
+ } else if (tier2) {
+ val recordKeyCol = functions.col(recordKeyFields.get(0)).cast(StringType)
+ val partitionPathCol =
buildSimpleKeygenPartitionPathColumn(partitionPathFields.get(0), hiveStyle)
+ (recordKeyCol, partitionPathCol)
+ } else {
+ val recordKeyUdf = functions.udf(
+ new UDF1[Row, String] { override def call(row: Row): String =
keyGenerator.getRecordKey(row) },
Review Comment:
The tier 3 fallback calls the single-arg getRecordKey/getPartitionPath(Row)
overloads, which BuiltinKeyGenerator itself notes as having considerable
overhead versus the schema-qualified overloads, so these keygens may end up
slower per record than the old RDD path.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
