linliu-code commented on code in PR #19610:
URL: https://github.com/apache/hudi/pull/19610#discussion_r3788854529
##########
hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala:
##########
@@ -266,6 +266,74 @@ case class MergeIntoHoodieTableCommand(mergeInto:
MergeIntoTable) extends Hoodie
"ordering fields",
updatingActions.flatMap(_.assignments))
+ /**
+ * Mapping of the target table's partition columns onto the [[sourceTable]]
expression that
+ * supplies each one, resolved the same way as the record-key and ordering
fields.
+ *
+ * This is required, not optional. For a target table bearing a record key
the incoming batch is
+ * written from the [[sourceTable]] alone (see [[getProcessedInputDf]]) and
Hudi's index tagging
+ * is what identifies updates - tagging keys off `(recordKey,
partitionPath)`. When the partition
+ * column reaches the writer unset, the key generator resolves it to the
default partition, so
+ * tagging looks in the wrong partition, finds nothing, and treats every
incoming record as
+ * not-matched. That is silently destructive rather than merely wrong:
+ *
+ * - with only `WHEN MATCHED` clauses the record falls through to
[[HoodieRecord.SENTINEL]] in
+ * `ExpressionPayload#processNotMatchedRecord` and is dropped - the
statement reports success
+ * and writes an *empty commit*;
+ * - with a `WHEN NOT MATCHED ... INSERT` clause the record is instead
inserted into the default
+ * partition, **duplicating the primary key across two partitions** while
the intended update
+ * is lost.
+ *
+ * Resolving the column here means the merge places records in the right
partition whenever the
+ * value is derivable. Three sources are consulted, in order:
+ *
+ * 1. the `ON` condition, via [[recordKeyAttributeToConditionExpression]],
which already
+ * enumerates partition fields but leaves them optional (see the "allow
partition path to be
+ * part of the merge condition but not required" note there) - reused
as-is so a column
+ * matched under a different source name (`ON t.dt = s.event_dt`) keeps
working, and so the
+ * same target attribute is never contributed twice;
+ * 2. the source-table output and the MERGE assignments, via
+ * [[resolveFieldAssociationsBetweenSourceAndTarget]];
+ * 3. otherwise the statement is rejected, rather than letting the write
silently corrupt the
+ * table.
+ *
+ * NOTE: this applies only to tables bearing a record key. A primary-keyless
target takes the
+ * other branch of [[getProcessedInputDf]], where the source is
left-outer-joined with the
+ * target and the meta columns are projected, and
[[MergeIntoKeyGenerator.getPartitionPath]]
+ * reads `_hoodie_partition_path` off that meta - so a matched row is
already placed in the
+ * right partition without the source carrying the column at all.
+ */
+ private lazy val partitionFieldsAssociatedExpressions: Seq[(Attribute,
Expression)] =
+ if (!hasPrimaryKey() || hoodieCatalogTable.partitionFields.isEmpty) {
+ Seq.empty
+ } else {
+ val resolver = sparkSession.sessionState.conf.resolver
+ // Associations the ON-condition path has already produced; anything
covered there needs no
+ // further resolution, and re-adding it would put two aliases for one
column in the projection.
+ val resolvedFromCondition = recordKeyAttributeToConditionExpression
+ hoodieCatalogTable.partitionFields.toSeq.flatMap { partitionField =>
+ if (resolvedFromCondition.exists { case (attr, _) =>
resolver(attr.name, partitionField) }) {
+ Seq.empty
+ } else {
+ try {
+ resolveFieldAssociationsBetweenSourceAndTarget(
+ resolver,
+ mergeInto.targetTable,
+ mergeInto.sourceTable,
+ Seq(partitionField),
+ "partition fields",
+ updatingActions.flatMap(_.assignments) ++
insertingActions.flatMap(_.assignments))
Review Comment:
An assignment gives the record's *new* partition, not its current one — so
for a partition-changing update on a non-global index (`update set t.dt =
s.new_dt` where `new_dt` differs from the row's partition), tagging looks in
`new_dt`, misses, and the record is dropped as `SENTINEL`: the empty commit
this PR is fixing. The assignment test passes because `src_dt` equals the
existing partition value; could we add a case with a differing value to pin
down what actually happens there?
Relatedly, `orderingFieldsAssociatedExpressions` (`:267`) uses only
`updatingActions` — with `insertingActions` in the union here, a merge whose
INSERT clause alone names the partition column will tag *matched* records from
it. Same applies to the spark4-common copy.
##########
hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala:
##########
@@ -266,6 +266,74 @@ case class MergeIntoHoodieTableCommand(mergeInto:
MergeIntoTable) extends Hoodie
"ordering fields",
updatingActions.flatMap(_.assignments))
+ /**
+ * Mapping of the target table's partition columns onto the [[sourceTable]]
expression that
+ * supplies each one, resolved the same way as the record-key and ordering
fields.
+ *
+ * This is required, not optional. For a target table bearing a record key
the incoming batch is
+ * written from the [[sourceTable]] alone (see [[getProcessedInputDf]]) and
Hudi's index tagging
+ * is what identifies updates - tagging keys off `(recordKey,
partitionPath)`. When the partition
+ * column reaches the writer unset, the key generator resolves it to the
default partition, so
+ * tagging looks in the wrong partition, finds nothing, and treats every
incoming record as
+ * not-matched. That is silently destructive rather than merely wrong:
+ *
+ * - with only `WHEN MATCHED` clauses the record falls through to
[[HoodieRecord.SENTINEL]] in
+ * `ExpressionPayload#processNotMatchedRecord` and is dropped - the
statement reports success
+ * and writes an *empty commit*;
+ * - with a `WHEN NOT MATCHED ... INSERT` clause the record is instead
inserted into the default
+ * partition, **duplicating the primary key across two partitions** while
the intended update
+ * is lost.
+ *
+ * Resolving the column here means the merge places records in the right
partition whenever the
+ * value is derivable. Three sources are consulted, in order:
+ *
+ * 1. the `ON` condition, via [[recordKeyAttributeToConditionExpression]],
which already
+ * enumerates partition fields but leaves them optional (see the "allow
partition path to be
+ * part of the merge condition but not required" note there) - reused
as-is so a column
+ * matched under a different source name (`ON t.dt = s.event_dt`) keeps
working, and so the
+ * same target attribute is never contributed twice;
+ * 2. the source-table output and the MERGE assignments, via
+ * [[resolveFieldAssociationsBetweenSourceAndTarget]];
+ * 3. otherwise the statement is rejected, rather than letting the write
silently corrupt the
+ * table.
+ *
+ * NOTE: this applies only to tables bearing a record key. A primary-keyless
target takes the
+ * other branch of [[getProcessedInputDf]], where the source is
left-outer-joined with the
+ * target and the meta columns are projected, and
[[MergeIntoKeyGenerator.getPartitionPath]]
+ * reads `_hoodie_partition_path` off that meta - so a matched row is
already placed in the
+ * right partition without the source carrying the column at all.
+ */
+ private lazy val partitionFieldsAssociatedExpressions: Seq[(Attribute,
Expression)] =
+ if (!hasPrimaryKey() || hoodieCatalogTable.partitionFields.isEmpty) {
Review Comment:
Should this be gated on the index being global? For `hoodie.index.type =
RECORD_INDEX` (global, and `hoodie.record.index.update.partition.path` defaults
to false), `tagGlobalLocationBackToRecords` takes the branch at
`HoodieIndexUtils:656-663` and re-keys the incoming record to the existing
location's partition, so
```sql
MERGE INTO t USING (select 1L as id, 15.0 as amount) s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.amount = s.amount
```
on a partitioned table updates correctly today and would now be rejected.
Same for `GLOBAL_SIMPLE`/`GLOBAL_BLOOM` with `update.partition.path=false`.
Worth noting `useGlobalIndex` (`:1199`) returns the value of
`update.partition.path`, so it's true precisely when the guard *isn't* needed —
the check would need "is global AND !updatePartitionPath". The existing
global-index merge test (`TestMergeIntoTable.scala:242`) projects the partition
column, so CI wouldn't flag this. Same applies to the spark4-common copy.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeIntoPartitionFieldResolution.scala:
##########
@@ -0,0 +1,387 @@
+/*
+ * 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.dml.others
+
+import org.apache.hudi.common.table.timeline.TimelineUtils
+import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
+
+import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
+
+/**
+ * Regression coverage for MERGE INTO partition-column resolution.
+ *
+ * For a target table bearing a record key, the incoming batch is written from
the source alone
+ * and Hudi's index tagging identifies updates, keying off `(recordKey,
partitionPath)`. Before the
+ * fix, a partition column the source did not carry was never back-filled into
the source
+ * projection (only record-key and ordering columns were), so the key
generator resolved it to the
+ * default partition and tagging looked in the wrong place. Every incoming
record came back
+ * not-matched, which is silently destructive in two different ways depending
on the clauses:
+ *
+ * - `WHEN MATCHED` only: the record falls through to HoodieRecord.SENTINEL in
+ * `ExpressionPayload#processNotMatchedRecord` and is dropped - the
statement reports success and
+ * writes an EMPTY COMMIT, leaving the target untouched.
+ * - `WHEN NOT MATCHED ... INSERT` present: the record is inserted into the
default partition
+ * instead, DUPLICATING THE PRIMARY KEY across two partitions while the
intended update is lost.
+ *
+ * The assertions below deliberately check the resulting rows AND their
`_hoodie_partition_path`,
+ * not merely that the data is unchanged. "Unchanged" is exactly what the
dropped-record bug
+ * produces, so a test asserting only that passes vacuously in the presence of
this defect.
+ */
+class TestMergeIntoPartitionFieldResolution extends HoodieSparkSqlTestBase {
+
+ private val expectedError = "Failed to resolve partition fields"
+
+ private def createPartitionedTable(tableName: String, tableType: String,
location: String): Unit =
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id bigint,
+ | name string,
+ | amount double,
+ | ts bigint,
+ | dt string
+ |) using hudi
+ | partitioned by (dt)
+ | tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | preCombineField = 'ts'
+ | )
+ | location '$location'
+ """.stripMargin)
+
+ Seq("cow", "mor").foreach { tableType =>
+ test(s"Test MergeInto rejects an update whose source omits the partition
column ($tableType)") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ createPartitionedTable(tableName, tableType,
s"${tmp.getCanonicalPath}/$tableName")
+ spark.sql(s"insert into $tableName values (1, 'a', 10.0, 1,
'2026-08-11')")
+
+ // `dt` appears in neither the source output, the ON condition, nor
the assignments.
+ // Before the fix this reported success and wrote an empty commit.
+ checkExceptionContain(
+ s"""
+ |merge into $tableName as t
+ |using (
+ | select 1L as id, 15.0 as amount, 200L as ts
+ |) as s
+ |on t.id = s.id
+ |when matched then update set t.amount = s.amount, t.ts = s.ts
+ """.stripMargin)(expectedError)
+
+ // Target untouched, and still a single row in its original partition.
+ checkAnswer(s"select id, amount, ts, dt, _hoodie_partition_path from
$tableName")(
+ Seq(1L, 10.0, 1L, "2026-08-11", "dt=2026-08-11")
+ )
+ }
+ }
+ }
+
+ Seq("cow", "mor").foreach { tableType =>
+ test(s"Test MergeInto with an INSERT clause cannot duplicate the key into
the default partition ($tableType)") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ createPartitionedTable(tableName, tableType,
s"${tmp.getCanonicalPath}/$tableName")
+ spark.sql(s"insert into $tableName values (1, 'a', 10.0, 1,
'2026-08-11')")
+
+ // The most damaging shape: with a NOT MATCHED clause the
mis-partitioned record used to be
+ // INSERTED into the default partition, yielding two rows with id=1 in
different partitions
+ // (name/dt null on the new one) while the intended update never
landed.
+ checkExceptionContain(
+ s"""
+ |merge into $tableName as t
+ |using (
+ | select 1L as id, 99.0 as amount, 9L as ts
+ |) as s
+ |on t.id = s.id
+ |when matched then update set t.amount = s.amount, t.ts = s.ts
+ |when not matched then insert (id, amount, ts) values (s.id,
s.amount, s.ts)
+ """.stripMargin)(expectedError)
+
+ // Exactly one row, in the right partition - no duplicate key, no
default-partition row.
+ checkAnswer(s"select count(*) from $tableName")(Seq(1L))
+ checkAnswer(s"select id, amount, dt, _hoodie_partition_path from
$tableName")(
+ Seq(1L, 10.0, "2026-08-11", "dt=2026-08-11")
+ )
+ }
+ }
+ }
+
+ Seq("cow", "mor").foreach { tableType =>
+ test(s"Test MergeInto applies when the source projects the partition
column ($tableType)") {
Review Comment:
Could we add a `when matched then delete` case? `DeleteAction` carries no
assignments, so a delete-only merge whose source omits the partition column now
hits the new throw via a path none of the nine cases exercise. Previously it
wrote an empty commit and deleted nothing, so rejecting looks right — just
untested, and it's the common CDC shape.
--
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]
