codope commented on code in PR #7825:
URL: https://github.com/apache/hudi/pull/7825#discussion_r1094183999
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieSqlCommonUtils.scala:
##########
@@ -35,7 +35,7 @@ import org.apache.spark.sql.catalyst.expressions.{And,
Attribute, Cast, Expressi
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
import org.apache.spark.sql.types._
-import org.apache.spark.sql.{AnalysisException, Column, DataFrame,
SparkSession}
+import org.apache.spark.sql.{AnalysisException, Column, DataFrame,
HoodieDataTypeUtils, HoodieInternalRowUtils, SparkSession}
Review Comment:
nit: optimize imports (`HoodieInternalRowUtils` is not used).
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala:
##########
@@ -188,7 +188,6 @@ trait ProvidesHoodieConfig extends Logging {
PRECOMBINE_FIELD.key -> preCombineField,
PARTITIONPATH_FIELD.key -> partitionFieldsStr,
PAYLOAD_CLASS_NAME.key -> payloadClassName,
- HoodieWriteConfig.COMBINE_BEFORE_INSERT.key ->
String.valueOf(hasPrecombineColumn),
Review Comment:
I guess the intent here was to automatically infer the COMBINE_BEFORE_INSERT
config. With this change, it not enough for user to just configure precombine
field, they also need to enable COMBINE_BEFORE_INSERT if they want to
deduplicate. Isn't it?
Is there validation in code which checks that if COMBINE_BEFORE_INSERT is
enabled then precombine field is also configured? If not, it would be better to
add as part of configs improvement story.
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/HoodieDatasetBulkInsertHelper.scala:
##########
@@ -220,4 +215,39 @@ object HoodieDatasetBulkInsertHelper
val keyGenerator = ReflectionUtils.loadClass(keyGeneratorClassName, new
TypedProperties(config.getProps)).asInstanceOf[BuiltinKeyGenerator]
keyGenerator.getPartitionPathFields.asScala
}
+
+ /**
+ * We use custom Spark [[Partitioner]] that is aware of the target table's
partitioning
+ * so that during inevitable shuffling required for de-duplication, we also
assign records
+ * into individual Spark partitions in a way affine with target table's
physical partitioning
+ * (ie records from the same table's partition will be co-located in the
same Spark's partition)
+ *
+ * This would allow us to
+ * <ul>
+ * <li>Save on additional shuffling subsequently (by
[[BulkInsertPartitioner]])</li>
+ * <li>Avoid "small files explosion" entailed by random (hash)
partitioning stemming
+ * from the fact that every Spark partition hosts records from many
table's partitions
+ * resulting into every Spark task writing into their own files in these
partitions (in
+ * case no subsequent re-partitioning is performed)
+ * </li>
+ * <ul>
+ *
+ * For more details check out HUDI-5685
+ */
+ private case class TablePartitioningAwarePartitioner(override val
numPartitions: Int,
Review Comment:
I understand the benefit but have we tested it?
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieDataTypeUtils.scala:
##########
@@ -18,10 +18,29 @@
package org.apache.spark.sql
+import org.apache.hudi.common.model.HoodieRecord
import org.apache.spark.sql.types._
+import scala.jdk.CollectionConverters.collectionAsScalaIterableConverter
+
object HoodieDataTypeUtils {
+ /**
+ * Checks whether provided schema contains Hudi's meta-fields
+ *
+ * NOTE: This method validates presence of just one field
[[HoodieRecord.RECORD_KEY_METADATA_FIELD]],
+ * however assuming that meta-fields should either be omitted or specified
in full
+ */
+ def hasMetaFields(structType: StructType): Boolean =
+ structType.getFieldIndex(HoodieRecord.RECORD_KEY_METADATA_FIELD).isDefined
+
+ // TODO scala-doc
Review Comment:
nit: remove todo?
--
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]
