You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2025/01/15 19:41:12 UTC
(hudi) branch master updated: [HUDI-8758] Enforce insert dedup policy for across Spark SQL and Spark datasource writes (#12588)
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 1e265042576 [HUDI-8758] Enforce insert dedup policy for across Spark SQL and Spark datasource writes (#12588)
1e265042576 is described below
commit 1e265042576fccb6af2014afe106032a11d90ae0
Author: Lin Liu <14...@users.noreply.github.com>
AuthorDate: Wed Jan 15 11:41:04 2025 -0800
[HUDI-8758] Enforce insert dedup policy for across Spark SQL and Spark datasource writes (#12588)
---
.../main/java/org/apache/hudi/DataSourceUtils.java | 67 +++++++++---
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 63 +++++++++--
.../spark/sql/hudi/ProvidesHoodieConfig.scala | 41 +------
.../hudi/command/ValidateDuplicateKeyPayload.scala | 44 --------
.../java/org/apache/hudi/TestDataSourceUtils.java | 2 +-
.../org/apache/hudi/TestHoodieSparkSqlWriter.scala | 55 +++++++++-
.../org/apache/hudi/TestInsertDedupPolicy.scala | 121 +++++++++++++++++++++
.../spark/sql/hudi/dml/TestInsertTable.scala | 95 ++++++++++++++--
.../apache/hudi/utilities/streamer/StreamSync.java | 2 +-
9 files changed, 370 insertions(+), 120 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
index c8666711189..94926ec224e 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
@@ -38,6 +38,7 @@ import org.apache.hudi.common.util.TablePathUtils;
import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodiePayloadConfig;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieDuplicateKeyException;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.TableNotFoundException;
import org.apache.hudi.storage.HoodieStorage;
@@ -277,31 +278,69 @@ public class DataSourceUtils {
}
/**
- * Drop records already present in the dataset.
+ * Drop records already present in the dataset if {@code failOnDuplicates} is {@code false}.
+ * Otherwise, throw a {@link HoodieDuplicateKeyException} if duplicates are found.
*
- * @param engineContext the spark engine context
- * @param incomingHoodieRecords HoodieRecords to deduplicate
- * @param writeConfig HoodieWriteConfig
+ * @param engineContext the Spark engine context
+ * @param incomingHoodieRecords the HoodieRecords to deduplicate
+ * @param writeConfig the HoodieWriteConfig
+ * @param failOnDuplicates a flag indicating whether to fail when duplicates are found
+ * @return a JavaRDD of deduplicated HoodieRecords
*/
@SuppressWarnings("unchecked")
- public static JavaRDD<HoodieRecord> dropDuplicates(HoodieSparkEngineContext engineContext, JavaRDD<HoodieRecord> incomingHoodieRecords,
- HoodieWriteConfig writeConfig) {
+ public static JavaRDD<HoodieRecord> handleDuplicates(HoodieSparkEngineContext engineContext,
+ JavaRDD<HoodieRecord> incomingHoodieRecords,
+ HoodieWriteConfig writeConfig,
+ boolean failOnDuplicates) {
try {
SparkRDDReadClient client = new SparkRDDReadClient<>(engineContext, writeConfig);
return client.tagLocation(incomingHoodieRecords)
- .filter(r -> !((HoodieRecord<HoodieRecordPayload>) r).isCurrentLocationKnown());
+ .filter(r -> shouldIncludeRecord((HoodieRecord<HoodieRecordPayload>) r, failOnDuplicates));
} catch (TableNotFoundException e) {
- // this will be executed when there is no hoodie table yet
- // so no dups to drop
+ // No table exists yet, so no duplicates to drop
return incomingHoodieRecords;
}
}
+ /**
+ * Determines if a record should be included in the result after deduplication.
+ *
+ * @param record The Hoodie record to evaluate.
+ * @param failOnDuplicates Whether to fail on detecting duplicates.
+ * @return true if the record should be included; false otherwise.
+ */
+ private static boolean shouldIncludeRecord(HoodieRecord<?> record, boolean failOnDuplicates) {
+ if (!record.isCurrentLocationKnown()) {
+ return true;
+ }
+ if (failOnDuplicates) {
+ // Fail if duplicates are found and the flag is set
+ throw new HoodieDuplicateKeyException(record.getRecordKey());
+ }
+ return false;
+ }
+
+ /**
+ * Resolves duplicate records in the provided {@code incomingHoodieRecords}.
+ *
+ * <p>If {@code failOnDuplicates} is {@code false}, duplicate records already present in the dataset
+ * are dropped. Otherwise, a {@link HoodieDuplicateKeyException} is thrown if duplicates are found.</p>
+ *
+ * @param jssc the Spark context used for executing the deduplication
+ * @param incomingHoodieRecords the input {@link JavaRDD} of {@link HoodieRecord} objects to process
+ * @param parameters a map of configuration parameters, including the dataset path under the key {@code "path"}
+ * @param failOnDuplicates a flag indicating whether to fail when duplicates are found
+ * @return a {@link JavaRDD} of deduplicated {@link HoodieRecord} objects
+ */
@SuppressWarnings("unchecked")
- public static JavaRDD<HoodieRecord> dropDuplicates(JavaSparkContext jssc, JavaRDD<HoodieRecord> incomingHoodieRecords,
- Map<String, String> parameters) {
- HoodieWriteConfig writeConfig =
- HoodieWriteConfig.newBuilder().withPath(parameters.get("path")).withProps(parameters).build();
- return dropDuplicates(new HoodieSparkEngineContext(jssc), incomingHoodieRecords, writeConfig);
+ public static JavaRDD<HoodieRecord> resolveDuplicates(JavaSparkContext jssc,
+ JavaRDD<HoodieRecord> incomingHoodieRecords,
+ Map<String, String> parameters,
+ boolean failOnDuplicates) {
+ HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
+ .withPath(parameters.get("path"))
+ .withProps(parameters).build();
+ return handleDuplicates(
+ new HoodieSparkEngineContext(jssc), incomingHoodieRecords, writeConfig, failOnDuplicates);
}
}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
index 1c411db1a15..c84d95abff9 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
@@ -29,6 +29,7 @@ import org.apache.hudi.DataSourceOptionsHelper.fetchMissingWriteConfigsFromTable
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.HoodieConversionUtils.{toProperties, toScalaOption}
import org.apache.hudi.HoodieSparkSqlWriter.StreamingWriteParams
+import org.apache.hudi.HoodieSparkSqlWriterInternal.{handleInsertDuplicates, shouldDropDuplicatesForInserts, shouldFailWhenDuplicatesFound}
import org.apache.hudi.HoodieSparkUtils.sparkAdapter
import org.apache.hudi.HoodieWriterUtils._
import org.apache.hudi.avro.AvroSchemaUtils.resolveNullableSchema
@@ -66,8 +67,7 @@ import org.apache.hudi.sync.common.HoodieSyncConfig
import org.apache.hudi.sync.common.util.SyncUtilHelpers
import org.apache.hudi.sync.common.util.SyncUtilHelpers.getHoodieMetaSyncException
import org.apache.hudi.util.SparkKeyGenUtils
-
-import org.apache.spark.api.java.JavaSparkContext
+import org.apache.spark.api.java.{JavaRDD, JavaSparkContext}
import org.apache.spark.sql.HoodieDataTypeUtils.tryOverrideParquetWriteLegacyFormatProperty
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
@@ -79,7 +79,6 @@ import org.apache.spark.{SPARK_VERSION, SparkContext}
import org.slf4j.LoggerFactory
import java.util.function.BiConsumer
-
import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.util.{Failure, Success, Try}
@@ -506,12 +505,8 @@ class HoodieSparkSqlWriterInternal {
case Failure(e) => throw new HoodieRecordCreationException("Failed to create Hoodie Spark Record", e)
}
- val dedupedHoodieRecords =
- if (hoodieConfig.getBoolean(INSERT_DROP_DUPS) && operation != WriteOperationType.INSERT_OVERWRITE_TABLE && operation != WriteOperationType.INSERT_OVERWRITE) {
- DataSourceUtils.dropDuplicates(jsc, hoodieRecords, parameters.asJava)
- } else {
- hoodieRecords
- }
+ // Remove duplicates from incoming records based on existing keys from storage.
+ val dedupedHoodieRecords = handleInsertDuplicates(hoodieRecords, hoodieConfig, operation, jsc, parameters)
client.startCommitWithTime(instantTime, commitActionType)
try {
val writeResult = DataSourceUtils.doWriteOperation(client, dedupedHoodieRecords, instantTime, operation,
@@ -554,13 +549,16 @@ class HoodieSparkSqlWriterInternal {
var operation = WriteOperationType.fromValue(hoodieConfig.getString(OPERATION))
// TODO clean up
// It does not make sense to allow upsert() operation if INSERT_DROP_DUPS is true
+ // or INSERT_DUP_POLICY is `drop` or `fail`.
// Auto-correct the operation to "insert" if OPERATION is set to "upsert" wrongly
// or not set (in which case it will be set as "upsert" by parametersWithWriteDefaults()) .
- if (hoodieConfig.getBoolean(INSERT_DROP_DUPS) &&
- operation == WriteOperationType.UPSERT) {
+ if ((hoodieConfig.getBoolean(INSERT_DROP_DUPS) ||
+ shouldDropDuplicatesForInserts(hoodieConfig) ||
+ shouldFailWhenDuplicatesFound(hoodieConfig))
+ && operation == WriteOperationType.UPSERT) {
log.warn(s"$UPSERT_OPERATION_OPT_VAL is not applicable " +
- s"when $INSERT_DROP_DUPS is set to be true, " +
+ s"when $INSERT_DROP_DUPS is set to be true, or $INSERT_DUP_POLICY is set to fail or drop, " +
s"overriding the $OPERATION to be $INSERT_OPERATION_OPT_VAL")
operation = WriteOperationType.INSERT
@@ -1165,3 +1163,44 @@ class HoodieSparkSqlWriterInternal {
.map(_.toLong).filter(SQLExecution.getQueryExecution(_) eq newQueryExecution)
}
}
+
+object HoodieSparkSqlWriterInternal {
+ // Check if duplicates should be dropped.
+ def shouldDropDuplicatesForInserts(hoodieConfig: HoodieConfig): Boolean = {
+ hoodieConfig.contains(INSERT_DUP_POLICY) &&
+ hoodieConfig.getString(INSERT_DUP_POLICY).equalsIgnoreCase(DROP_INSERT_DUP_POLICY)
+ }
+
+ // Check if we should fail if duplicates are found.
+ def shouldFailWhenDuplicatesFound(hoodieConfig: HoodieConfig): Boolean = {
+ hoodieConfig.contains(INSERT_DUP_POLICY) &&
+ hoodieConfig.getString(INSERT_DUP_POLICY).equalsIgnoreCase(FAIL_INSERT_DUP_POLICY)
+ }
+
+ // Check if deduplication is required.
+ def isDeduplicationRequired(hoodieConfig: HoodieConfig): Boolean = {
+ hoodieConfig.getBoolean(INSERT_DROP_DUPS) ||
+ shouldFailWhenDuplicatesFound(hoodieConfig) ||
+ shouldDropDuplicatesForInserts(hoodieConfig)
+ }
+
+ // Check if deduplication is needed.
+ def isDeduplicationNeeded(operation: WriteOperationType): Boolean = {
+ operation == WriteOperationType.INSERT
+ }
+
+ def handleInsertDuplicates(incomingRecords: JavaRDD[HoodieRecord[_]],
+ hoodieConfig: HoodieConfig,
+ operation: WriteOperationType,
+ jsc: JavaSparkContext,
+ parameters: Map[String, String]): JavaRDD[HoodieRecord[_]] = {
+ // If no deduplication is needed, return the incoming records as is
+ if (!isDeduplicationRequired(hoodieConfig) || !isDeduplicationNeeded(operation)) {
+ incomingRecords
+ } else {
+ // Perform deduplication
+ DataSourceUtils.resolveDuplicates(
+ jsc, incomingRecords, parameters.asJava, shouldFailWhenDuplicatesFound(hoodieConfig))
+ }
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala
index 7656e611463..a6008cbf523 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala
@@ -43,7 +43,7 @@ import org.apache.spark.sql.hive.HiveExternalCatalog
import org.apache.spark.sql.hudi.HoodieOptionConfig.mapSqlOptionsToDataSourceWriteConfigs
import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.{filterHoodieConfigs, isUsingHiveCatalog}
import org.apache.spark.sql.hudi.ProvidesHoodieConfig.{combineOptions, getPartitionPathFieldWriteConfig}
-import org.apache.spark.sql.hudi.command.{SqlKeyGenerator, ValidateDuplicateKeyPayload}
+import org.apache.spark.sql.hudi.command.SqlKeyGenerator
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.PARTITION_OVERWRITE_MODE
import org.apache.spark.sql.types.StructType
@@ -96,23 +96,6 @@ trait ProvidesHoodieConfig extends Logging {
defaultOpts = defaultOpts, overridingOpts = overridingOpts)
}
-
- private def deducePayloadClassNameLegacy(operation: String, tableType: String, insertMode: InsertMode): String = {
- if (operation == UPSERT_OPERATION_OPT_VAL &&
- tableType == COW_TABLE_TYPE_OPT_VAL && insertMode == InsertMode.STRICT) {
- // Validate duplicate key for COW, for MOR it will do the merge with the DefaultHoodieRecordPayload
- // on reading.
- // TODO use HoodieSparkValidateDuplicateKeyRecordMerger when SparkRecordMerger is default
- classOf[ValidateDuplicateKeyPayload].getCanonicalName
- } else if (operation == INSERT_OPERATION_OPT_VAL && tableType == COW_TABLE_TYPE_OPT_VAL &&
- insertMode == InsertMode.STRICT) {
- // Validate duplicate key for inserts to COW table when using strict insert mode.
- classOf[ValidateDuplicateKeyPayload].getCanonicalName
- } else {
- classOf[DefaultHoodieRecordPayload].getCanonicalName
- }
- }
-
/**
* Deduce the sql write operation for INSERT_INTO
*/
@@ -274,25 +257,9 @@ trait ProvidesHoodieConfig extends Logging {
Map()
}
- // try to use new insert dup policy instead of legacy insert mode to deduce payload class. If only insert mode is explicitly specified,
- // w/o specifying any value for insert dup policy, legacy configs will be honored. But on all other cases (i.e when neither of the configs is set,
- // or when both configs are set, or when only insert dup policy is set), we honor insert dup policy and ignore the insert mode.
- val useLegacyInsertDropDupFlow = insertModeSet && !insertDupPolicySet
- val deducedPayloadClassName = if (useLegacyInsertDropDupFlow) {
- deducePayloadClassNameLegacy(operation, tableType, insertMode)
- } else {
- if (insertDupPolicy == FAIL_INSERT_DUP_POLICY) {
- classOf[ValidateDuplicateKeyPayload].getCanonicalName
- } else {
- classOf[DefaultHoodieRecordPayload].getCanonicalName
- }
- }
-
- val (recordMergeMode, recordMergeStrategy) = if (deducedPayloadClassName.equals(classOf[ValidateDuplicateKeyPayload].getCanonicalName)) {
- (RecordMergeMode.CUSTOM.name(), HoodieRecordMerger.PAYLOAD_BASED_MERGE_STRATEGY_UUID)
- } else {
- (RecordMergeMode.EVENT_TIME_ORDERING.name(), HoodieRecordMerger.DEFAULT_MERGE_STRATEGY_UUID)
- }
+ val deducedPayloadClassName = classOf[DefaultHoodieRecordPayload].getCanonicalName
+ val recordMergeMode = RecordMergeMode.EVENT_TIME_ORDERING.name
+ val recordMergeStrategy = HoodieRecordMerger.DEFAULT_MERGE_STRATEGY_UUID
if (tableConfig.getPayloadClass.equals(classOf[DefaultHoodieRecordPayload].getCanonicalName) &&
tableConfig.getRecordMergeMode.equals(RecordMergeMode.EVENT_TIME_ORDERING)) {
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/ValidateDuplicateKeyPayload.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/ValidateDuplicateKeyPayload.scala
deleted file mode 100644
index 1a8297e20ce..00000000000
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/ValidateDuplicateKeyPayload.scala
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.hudi.command
-
-import org.apache.avro.Schema
-import org.apache.avro.generic.{GenericRecord, IndexedRecord}
-import org.apache.hudi.common.model.{DefaultHoodieRecordPayload, HoodieRecord}
-import org.apache.hudi.common.util.{Option => HOption}
-import org.apache.hudi.exception.HoodieDuplicateKeyException
-
-import java.util.Properties
-
-/**
- * Validate the duplicate key for insert statement without enable the INSERT_DROP_DUPS_OPT
- * config.
- */
-class ValidateDuplicateKeyPayload(record: GenericRecord, orderingVal: Comparable[_])
- extends DefaultHoodieRecordPayload(record, orderingVal) {
-
- def this(record: HOption[GenericRecord]) {
- this(if (record.isPresent) record.get else null, 0)
- }
-
- override def combineAndGetUpdateValue(currentValue: IndexedRecord,
- schema: Schema, properties: Properties): HOption[IndexedRecord] = {
- val key = currentValue.asInstanceOf[GenericRecord].get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString
- throw new HoodieDuplicateKeyException(key)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/TestDataSourceUtils.java b/hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/TestDataSourceUtils.java
index 375a7e9ded3..b89774c4a97 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/TestDataSourceUtils.java
+++ b/hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/TestDataSourceUtils.java
@@ -52,7 +52,7 @@ class TestDataSourceUtils extends HoodieClientTestBase {
Map<String, String> parameters = config.getProps().entrySet().stream().collect(Collectors.toMap(entry -> entry.getKey().toString(), entry -> entry.getValue().toString()));
List<HoodieRecord> newRecords = dataGen.generateInserts(newCommitTime, 10);
List<HoodieRecord> inputRecords = Stream.concat(records.subList(0, 10).stream(), newRecords.stream()).collect(Collectors.toList());
- List<HoodieRecord> output = DataSourceUtils.dropDuplicates(jsc, jsc.parallelize(inputRecords, 1), parameters).collect();
+ List<HoodieRecord> output = DataSourceUtils.resolveDuplicates(jsc, jsc.parallelize(inputRecords, 1), parameters, false).collect();
Set<String> expectedRecordKeys = newRecords.stream().map(HoodieRecord::getRecordKey).collect(Collectors.toSet());
assertEquals(expectedRecordKeys, output.stream().map(HoodieRecord::getRecordKey).collect(Collectors.toSet()));
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
index 17e2256591a..9cfc01ae6b2 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
@@ -32,7 +32,8 @@ import org.apache.hudi.testutils.{DataSourceTestUtils, HoodieClientTestUtils}
import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
import org.apache.avro.Schema
import org.apache.commons.io.FileUtils
-import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.DataSourceWriteOptions.{DROP_INSERT_DUP_POLICY, FAIL_INSERT_DUP_POLICY, INSERT_DROP_DUPS, INSERT_DUP_POLICY}
+import org.apache.hudi.common.config.{HoodieConfig, HoodieMetadataConfig}
import org.apache.spark.api.java.JavaSparkContext
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
import org.apache.spark.sql.functions.{expr, lit}
@@ -1246,6 +1247,57 @@ def testBulkInsertForDropPartitionColumn(): Unit = {
assert(exc.getMessage.contains("Consistent hashing bucket index does not work with COW table. Use simple bucket index or an MOR table."))
}
+ @Test
+ def testShouldDropDuplicatesForInserts(): Unit = {
+ val hoodieConfig: HoodieConfig = new HoodieConfig()
+ var shouldDrop: Boolean = HoodieSparkSqlWriterInternal.shouldDropDuplicatesForInserts(hoodieConfig)
+ assertFalse(shouldDrop)
+
+ hoodieConfig.setValue(INSERT_DUP_POLICY.key, DROP_INSERT_DUP_POLICY)
+ shouldDrop = HoodieSparkSqlWriterInternal.shouldDropDuplicatesForInserts(hoodieConfig)
+ assertTrue(shouldDrop)
+ }
+
+ @Test
+ def testShouldFailWhenDuplicatesFound(): Unit = {
+ val hoodieConfig: HoodieConfig = new HoodieConfig()
+ var shouldFail: Boolean = HoodieSparkSqlWriterInternal.shouldFailWhenDuplicatesFound(hoodieConfig)
+ assertFalse(shouldFail)
+
+ hoodieConfig.setValue(INSERT_DUP_POLICY.key, FAIL_INSERT_DUP_POLICY)
+ shouldFail = HoodieSparkSqlWriterInternal.shouldFailWhenDuplicatesFound(hoodieConfig)
+ assertTrue(shouldFail)
+ }
+
+ @Test
+ def testIsDeduplicationRequired(): Unit = {
+ val hoodieConfig: HoodieConfig = new HoodieConfig()
+ var isRequired: Boolean = HoodieSparkSqlWriterInternal.isDeduplicationRequired(hoodieConfig)
+ assertFalse(isRequired)
+
+ hoodieConfig.setValue(INSERT_DUP_POLICY.key, FAIL_INSERT_DUP_POLICY)
+ isRequired = HoodieSparkSqlWriterInternal.isDeduplicationRequired(hoodieConfig)
+ assertTrue(isRequired)
+
+ hoodieConfig.setValue(INSERT_DUP_POLICY.key, DROP_INSERT_DUP_POLICY)
+ isRequired = HoodieSparkSqlWriterInternal.isDeduplicationRequired(hoodieConfig)
+ assertTrue(isRequired)
+
+ hoodieConfig.setValue(INSERT_DUP_POLICY.key, "")
+ hoodieConfig.setValue(INSERT_DROP_DUPS.key, "true")
+ isRequired = HoodieSparkSqlWriterInternal.isDeduplicationRequired(hoodieConfig)
+ assertTrue(isRequired)
+ }
+
+ @Test
+ def testIsDeduplicationNeeded(): Unit = {
+ assertFalse(HoodieSparkSqlWriterInternal.isDeduplicationNeeded(WriteOperationType.INSERT_OVERWRITE))
+ assertFalse(HoodieSparkSqlWriterInternal.isDeduplicationNeeded(WriteOperationType.INSERT_OVERWRITE_TABLE))
+ assertFalse(HoodieSparkSqlWriterInternal.isDeduplicationNeeded(WriteOperationType.UPSERT))
+ assertFalse(HoodieSparkSqlWriterInternal.isDeduplicationNeeded(WriteOperationType.INSERT_PREPPED))
+ assertTrue(HoodieSparkSqlWriterInternal.isDeduplicationNeeded(WriteOperationType.INSERT))
+ }
+
private def fetchActualSchema(): Schema = {
val tableMetaClient = createMetaClient(spark, tempBasePath)
new TableSchemaResolver(tableMetaClient).getTableAvroSchema(false)
@@ -1273,5 +1325,4 @@ object TestHoodieSparkSqlWriter {
Arguments.arguments("*5/03/1*", Seq("2016/03/15")),
Arguments.arguments("2016/03/*", Seq("2015/03/16", "2015/03/17")))
}
-
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestInsertDedupPolicy.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestInsertDedupPolicy.scala
new file mode 100644
index 00000000000..be5ed814905
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestInsertDedupPolicy.scala
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi
+
+import org.apache.hudi.DataSourceWriteOptions._
+import org.apache.hudi.common.config.{HoodieReaderConfig, HoodieStorageConfig}
+import org.apache.hudi.config.{HoodieCompactionConfig, HoodieWriteConfig}
+import org.apache.hudi.exception.HoodieUpsertException
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness
+import org.apache.spark.sql.{Dataset, Row, SaveMode}
+import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.{Arguments, MethodSource}
+
+class TestInsertDedupPolicy extends SparkClientFunctionalTestHarness {
+ val firstInsertData: Seq[(Int, String, String, String, Double, Int)] = Seq(
+ (10, "1", "rider-A", "driver-A", 19.10, 7),
+ (10, "2", "rider-B", "driver-B", 27.70, 1),
+ (10, "3", "rider-C", "driver-C", 33.90, 10))
+ val secondInsertData: Seq[(Int, String, String, String, Double, Int)] = Seq(
+ (11, "1", "rider-A", "driver-A", 1.1, 1),
+ (11, "2", "rider-B", "driver-B", 2.2, 2),
+ (11, "5", "rider-C", "driver-C", 3.3, 3))
+ val expectedForDrop: Seq[(Int, String, String, String, Double, Int)] = Seq(
+ (10, "1", "rider-A", "driver-A", 19.10, 7),
+ (10, "2", "rider-B", "driver-B", 27.70, 1),
+ (10, "3", "rider-C", "driver-C", 33.90, 10),
+ (11, "5", "rider-C", "driver-C", 3.3, 3))
+ val expectedForNone: Seq[(Int, String, String, String, Double, Int)] = Seq(
+ (10, "1", "rider-A", "driver-A", 19.10, 7),
+ (11, "1", "rider-A", "driver-A", 1.1, 1),
+ (10, "2", "rider-B", "driver-B", 27.70, 1),
+ (11, "2", "rider-B", "driver-B", 2.2, 2),
+ (10, "3", "rider-C", "driver-C", 33.90, 10),
+ (11, "5", "rider-C", "driver-C", 3.3, 3))
+
+ @ParameterizedTest
+ @MethodSource(Array("provideParams"))
+ def testInsertLogic(tableType: String,
+ dedupPolicy: String): Unit = {
+ val fgReaderOpts: Map[String, String] = Map(
+ HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key -> "true",
+ HoodieReaderConfig.MERGE_USE_RECORD_POSITIONS.key -> "true",
+ HoodieWriteConfig.RECORD_MERGE_MODE.key -> "EVENT_TIME_ORDERING",
+ DataSourceWriteOptions.OPERATION.key -> INSERT_OPERATION_OPT_VAL)
+ val insertDedupPolicy: Map[String, String] = Map(INSERT_DUP_POLICY.key -> dedupPolicy)
+ val opts = fgReaderOpts ++ insertDedupPolicy
+ val columns = Seq("ts", "key", "rider", "driver", "fare", "number")
+
+ // Write the first batch of data.
+ val inserts = spark.createDataFrame(firstInsertData).toDF(columns: _*)
+ inserts.write.format("hudi").
+ option(RECORDKEY_FIELD.key, "key").
+ option(PRECOMBINE_FIELD.key, "ts").
+ option(TABLE_TYPE.key, tableType).
+ option(DataSourceWriteOptions.TABLE_NAME.key, "test_table").
+ option(HoodieCompactionConfig.INLINE_COMPACT.key, "false").
+ options(opts).
+ mode(SaveMode.Overwrite).
+ save(basePath)
+
+ val insertsWithDup = spark.createDataFrame(secondInsertData).toDF(columns: _*)
+ if (dedupPolicy.equals(FAIL_INSERT_DUP_POLICY)) {
+ // Write and check throws.
+ Assertions.assertThrows(
+ classOf[HoodieUpsertException],
+ () => insertsWithDup.write.format("hudi").options(opts).mode(SaveMode.Append).save(basePath)
+ )
+ } else {
+ // Write data.
+ insertsWithDup.write.format("hudi").options(opts).mode(SaveMode.Append).save(basePath)
+ // Validate the data.
+ val df = spark.read.format("hudi").options(opts).load(basePath)
+ val finalDf = df.select("ts", "key", "rider", "driver", "fare", "number").sort("key")
+ val expected = if (dedupPolicy.equals(DROP_INSERT_DUP_POLICY)) expectedForDrop else expectedForNone
+ val expectedDf = spark.createDataFrame(expected).toDF(columns: _*).sort("key")
+ TestInsertDedupPolicy.validate(expectedDf, finalDf)
+ }
+ }
+}
+
+object TestInsertDedupPolicy {
+ def provideParams(): java.util.List[Arguments] = {
+ java.util.Arrays.asList(
+ Arguments.of("MERGE_ON_READ", NONE_INSERT_DUP_POLICY),
+ Arguments.of("MERGE_ON_READ", DROP_INSERT_DUP_POLICY),
+ Arguments.of("MERGE_ON_READ", FAIL_INSERT_DUP_POLICY),
+ Arguments.of("COPY_ON_WRITE", NONE_INSERT_DUP_POLICY),
+ Arguments.of("COPY_ON_WRITE", DROP_INSERT_DUP_POLICY),
+ Arguments.of("COPY_ON_WRITE", FAIL_INSERT_DUP_POLICY)
+ )
+ }
+
+ def validate(expectedDf: Dataset[Row], actualDf: Dataset[Row]): Unit = {
+ val expectedMinusActual = expectedDf.except(actualDf)
+ val actualMinusExpected = actualDf.except(expectedDf)
+ expectedDf.show(false)
+ actualDf.show(false)
+ assertTrue(expectedMinusActual.isEmpty && actualMinusExpected.isEmpty)
+ }
+}
+
+
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestInsertTable.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestInsertTable.scala
index 7814d75a05e..68bbf9e1482 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestInsertTable.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestInsertTable.scala
@@ -352,7 +352,7 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
test("Test Insert Into Non Partitioned Table") {
withRecordType(Seq(HoodieRecordType.AVRO, HoodieRecordType.SPARK))(withTempDir { tmp =>
val tableName = generateTableName
- spark.sql(s"set hoodie.sql.insert.mode=strict")
+ spark.sql(s"set hoodie.datasource.insert.dup.policy=fail")
// Create none partitioned cow table
spark.sql(
s"""
@@ -394,7 +394,7 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
// Create table with dropDup is true
val tableName2 = generateTableName
- spark.sql("set hoodie.datasource.write.insert.drop.duplicates = true")
+ spark.sql("set hoodie.datasource.insert.dup.policy=drop")
spark.sql(
s"""
|create table $tableName2 (
@@ -417,15 +417,15 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
Seq(1, "a1", 10.0, 1000)
)
// disable this config to avoid affect other test in this class.
- spark.sql("set hoodie.datasource.write.insert.drop.duplicates = false")
spark.sql(s"set hoodie.sql.insert.mode=upsert")
})
+ spark.sessionState.conf.unsetConf("hoodie.datasource.insert.dup.policy")
}
test("Test Insert Into None Partitioned Table strict mode with no preCombineField") {
withTempDir { tmp =>
val tableName = generateTableName
- spark.sql(s"set hoodie.sql.insert.mode=strict")
+ spark.sql(s"set hoodie.datasource.insert.dup.policy=fail")
// Create none partitioned cow table
spark.sql(
s"""
@@ -449,7 +449,6 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
Seq(1, "a1", 10.0),
Seq(2, "a2", 12.0)
)
-
spark.sql("set hoodie.merge.allow.duplicate.on.inserts = false")
assertThrows[HoodieDuplicateKeyException] {
try {
@@ -1329,6 +1328,7 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
}
test("Test insert pk-table") {
+ spark.sessionState.conf.unsetConf("hoodie.datasource.insert.dup.policy")
Seq("cow", "mor").foreach { tableType =>
withSQLConf("hoodie.sql.bulk.insert.enable" -> "false") {
withRecordType()(withTempDir { tmp =>
@@ -1529,6 +1529,7 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
}
test("Test enable hoodie.merge.allow.duplicate.on.inserts when write") {
+ spark.sessionState.conf.unsetConf("hoodie.datasource.insert.dup.policy")
spark.sql("set hoodie.datasource.write.operation = insert")
Seq("mor", "cow").foreach { tableType =>
withTempDir { tmp =>
@@ -1572,6 +1573,7 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
}
test("Test Insert Into Bucket Index Table") {
+ spark.sessionState.conf.unsetConf("hoodie.datasource.insert.dup.policy")
withTempDir { tmp =>
val tableName = generateTableName
// Create a partitioned table
@@ -1624,7 +1626,9 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
}
test("Test Bulk Insert Into Bucket Index Table") {
- withSQLConf("hoodie.datasource.write.operation" -> "bulk_insert", "hoodie.bulkinsert.shuffle.parallelism" -> "1") {
+ withSQLConf(
+ "hoodie.datasource.write.operation" -> "bulk_insert",
+ "hoodie.bulkinsert.shuffle.parallelism" -> "1") {
Seq("mor", "cow").foreach { tableType =>
Seq("true", "false").foreach { bulkInsertAsRow =>
withTempDir { tmp =>
@@ -1882,6 +1886,7 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
}
test("Test Insert Overwrite Into Consistent Bucket Index Table") {
+ spark.sessionState.conf.unsetConf("hoodie.datasource.insert.dup.policy")
withSQLConf("hoodie.sql.bulk.insert.enable" -> "false") {
withTempDir { tmp =>
val tableName = generateTableName
@@ -2053,6 +2058,7 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
}
test("Test Bulk Insert Into Consistent Hashing Bucket Index Table") {
+ spark.sessionState.conf.unsetConf("hoodie.datasource.insert.dup.policy")
withSQLConf("hoodie.datasource.write.operation" -> "bulk_insert") {
Seq("false", "true").foreach { bulkInsertAsRow =>
withTempDir { tmp =>
@@ -2317,7 +2323,6 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
countDownLatch.countDown
}
}
-
}
test("Test multiple partition fields pruning") {
@@ -2607,7 +2612,7 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
test("Test FAIL insert dup policy with INSERT_INTO explicit new configs") {
withRecordType(Seq(HoodieRecordType.AVRO))(withTempDir { tmp =>
Seq("cow", "mor").foreach { tableType =>
- val operation = WriteOperationType.UPSERT
+ val operation = WriteOperationType.INSERT
val dupPolicy = FAIL_INSERT_DUP_POLICY
withTable(generateTableName) { tableName =>
ingestAndValidateDataDupPolicy(tableType, tableName, tmp, operation,
@@ -2639,6 +2644,9 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
| PARTITIONED BY (boolean_field, float_field, byte_field, short_field, decimal_field, date_field, string_field, timestamp_field)
|LOCATION '${tmp.getCanonicalPath}'
""".stripMargin)
+ // Avoid operation type modification.
+ spark.sql(s"set ${INSERT_DROP_DUPS.key}=false")
+ spark.sql(s"set ${INSERT_DUP_POLICY.key}=$NONE_INSERT_DUP_POLICY")
// Insert data into partitioned table
spark.sql(
@@ -2851,7 +2859,9 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
spark.sessionState.conf.unsetConf("hoodie.datasource.write.operation")
}
- def ingestAndValidateDropDupPolicyBulkInsert(tableType: String, tableName: String, tmp: File,
+ def ingestAndValidateDropDupPolicyBulkInsert(tableType: String,
+ tableName: String,
+ tmp: File,
setOptions: List[String] = List.empty) : Unit = {
// set additional options
@@ -3087,4 +3097,71 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
})
}
}
+
+ test("Test table with insert dup policy") {
+ withTempDir { tmp =>
+ Seq(
+ NONE_INSERT_DUP_POLICY,
+ FAIL_INSERT_DUP_POLICY,
+ DROP_INSERT_DUP_POLICY).foreach(policy => {
+ val targetTable = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$targetTable"
+
+ spark.sql(s"set hoodie.datasource.insert.dup.policy=$policy")
+ spark.sql(
+ s"""
+ |create table ${targetTable} (
+ | `id` string,
+ | `name` string,
+ | `dt` bigint,
+ | `day` STRING,
+ | `hour` INT
+ |) using hudi
+ |tblproperties (
+ | 'primaryKey' = 'id',
+ | 'type' = 'MOR',
+ | 'preCombineField'= 'dt',
+ | 'hoodie.bucket.index.hash.field' = 'id',
+ | 'hoodie.bucket.index.num.buckets'= 512
+ | )
+ partitioned by (`day`,`hour`)
+ location '${tablePath}'
+ """.stripMargin)
+
+ spark.sql("set spark.sql.shuffle.partitions = 11")
+ spark.sql(
+ s"""
+ |insert into ${targetTable}
+ |select '1' as id, 'aa' as name, 123 as dt, '2024-02-19' as `day`, 10 as `hour`
+ |""".stripMargin)
+ if (policy.equals(FAIL_INSERT_DUP_POLICY)) {
+ checkExceptionContain(
+ () => spark.sql(
+ s"""
+ |insert into ${targetTable}
+ |select '1' as id, 'aa' as name, 1234 as dt, '2024-02-19' as `day`, 10 as `hour`
+ |""".stripMargin))(s"Duplicate key found for insert statement, key is: 1")
+ } else {
+ spark.sql(s"set hoodie.datasource.write.operation=insert")
+ spark.sql(
+ s"""
+ |insert into ${targetTable}
+ |select '1' as id, 'aa' as name, 1234 as dt, '2024-02-19' as `day`, 10 as `hour`
+ |""".stripMargin)
+ if (policy.equals(NONE_INSERT_DUP_POLICY)) {
+ checkAnswer(s"select id, name, dt, day, hour from $targetTable limit 10")(
+ Seq("1", "aa", 123, "2024-02-19", 10),
+ Seq("1", "aa", 1234, "2024-02-19", 10)
+ )
+ } else {
+ checkAnswer(s"select id, name, dt, day, hour from $targetTable limit 10")(
+ Seq("1", "aa", 123, "2024-02-19", 10)
+ )
+ }
+ }
+ })
+ }
+ spark.sessionState.conf.unsetConf("hoodie.datasource.insert.dup.policy")
+ }
}
+
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
index 12c869666e1..bb408753374 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
@@ -926,7 +926,7 @@ public class StreamSync implements Serializable, Closeable {
JavaRDD<HoodieRecord> records = (JavaRDD<HoodieRecord>) inputBatch.getBatch().orElseGet(() -> hoodieSparkContext.emptyRDD());
// filter dupes if needed
if (cfg.filterDupes) {
- records = DataSourceUtils.dropDuplicates(hoodieSparkContext, records, writeClient.getConfig());
+ records = DataSourceUtils.handleDuplicates(hoodieSparkContext, records, writeClient.getConfig(), false);
}
HoodieWriteResult writeResult = null;