voonhous commented on code in PR #19791:
URL: https://github.com/apache/hudi/pull/19791#discussion_r3886680534
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -617,8 +618,29 @@ class HoodieSparkSqlWriterInternal {
}
/**
- * Resolve wildcards in partitions
+ * Reject insert overwrite combined with non-blocking concurrency control.
*
+ * Insert overwrite reuses the deterministic bucket file id under
non-blocking concurrency
+ * control, but the replace commit records that same file id as replaced.
The file system view
+ * hides a replaced file group by file id (ignoring the replace instant), so
the freshly
+ * overwritten data would become invisible. Reject the combination to avoid
data loss.
+ */
+ private def validateNonBlockingConcurrencyControl(hoodieConfig:
HoodieConfig, operation: WriteOperationType): Unit = {
+ val isNonBlockingConcurrencyControl =
WriteConcurrencyMode.isNonBlockingConcurrencyControl(
+
hoodieConfig.getStringOrDefault(HoodieWriteConfig.WRITE_CONCURRENCY_MODE))
+ val rowWriterOverwriteType =
Option(hoodieConfig.getString(HoodieInternalConfig.BULKINSERT_OVERWRITE_OPERATION_TYPE))
+ .map(WriteOperationType.fromValue)
+ .orNull
+ val isInsertOverwrite = operation == WriteOperationType.INSERT_OVERWRITE ||
+ operation == WriteOperationType.INSERT_OVERWRITE_TABLE ||
+ rowWriterOverwriteType == WriteOperationType.INSERT_OVERWRITE ||
+ rowWriterOverwriteType == WriteOperationType.INSERT_OVERWRITE_TABLE
+ if (isNonBlockingConcurrencyControl && isInsertOverwrite) {
+ throw new HoodieException("Insert overwrite is not supported with
non-blocking concurrency control")
Review Comment:
Verified at 7c36e9ba: `WriteConcurrencyMode.checkInsertOverwriteSupported` +
`INSERT_OVERWRITE_NOT_SUPPORTED_ERROR` used at all four sites, single
`HoodieException`. Resolving.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -617,8 +618,29 @@ class HoodieSparkSqlWriterInternal {
}
/**
- * Resolve wildcards in partitions
+ * Reject insert overwrite combined with non-blocking concurrency control.
*
+ * Insert overwrite reuses the deterministic bucket file id under
non-blocking concurrency
+ * control, but the replace commit records that same file id as replaced.
The file system view
+ * hides a replaced file group by file id (ignoring the replace instant), so
the freshly
+ * overwritten data would become invisible. Reject the combination to avoid
data loss.
+ */
+ private def validateNonBlockingConcurrencyControl(hoodieConfig:
HoodieConfig, operation: WriteOperationType): Unit = {
+ val isNonBlockingConcurrencyControl =
WriteConcurrencyMode.isNonBlockingConcurrencyControl(
+
hoodieConfig.getStringOrDefault(HoodieWriteConfig.WRITE_CONCURRENCY_MODE))
+ val rowWriterOverwriteType =
Option(hoodieConfig.getString(HoodieInternalConfig.BULKINSERT_OVERWRITE_OPERATION_TYPE))
+ .map(WriteOperationType.fromValue)
+ .orNull
+ val isInsertOverwrite = operation == WriteOperationType.INSERT_OVERWRITE ||
+ operation == WriteOperationType.INSERT_OVERWRITE_TABLE ||
+ rowWriterOverwriteType == WriteOperationType.INSERT_OVERWRITE ||
+ rowWriterOverwriteType == WriteOperationType.INSERT_OVERWRITE_TABLE
+ if (isNonBlockingConcurrencyControl && isInsertOverwrite) {
+ throw new HoodieException("Insert overwrite is not supported with
non-blocking concurrency control")
+ }
+ }
+
+ /**
* @param partitions list of partitions that may contain wildcards
Review Comment:
Verified at 7c36e9ba: summary line and indent restored. Resolving.
--
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]
