voonhous commented on code in PR #19713:
URL: https://github.com/apache/hudi/pull/19713#discussion_r3852169992
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java:
##########
@@ -377,6 +378,9 @@ private static <R> Option<HoodieRecord<R>>
mergeIncomingWithExistingRecordWithEx
HoodieRecord<R> existing,
HoodieSchema writeSchema,
Review Comment:
**nit, feel free to ignore** -- `writeSchema` is now unused in this method:
both former uses (`prependMetaFields` and `Option.of(writeSchema)`) became
`mergedSchema`. Could we drop the parameter and the argument at line 482?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java:
##########
@@ -398,27 +402,62 @@ private static <R> Option<HoodieRecord<R>>
mergeIncomingWithExistingRecordWithEx
}
//record is inserted or updated
- String partitionPath = inferPartitionPath(incoming, existing,
writeSchemaWithMetaFields, keyGenerator, existingRecordContext, mergeResult);
+ String partitionPath = inferPartitionPath(incoming, existing,
mergedSchemaWithMetaFields, keyGenerator,
+ existingRecordContext, mergeResult, partitionResolvableFromRecord);
HoodieRecord<R> result =
existingRecordContext.constructHoodieRecord(mergeResult, partitionPath);
- HoodieRecord<R> withMeta = result.prependMetaFields(writeSchema,
writeSchemaWithMetaFields,
+ HoodieRecord<R> withMeta = result.prependMetaFields(mergedSchema,
mergedSchemaWithMetaFields,
Review Comment:
**major** -- generic guard behind this fix.
`HoodieAvroIndexedRecord.prependMetaFields:206` infers `metaFieldSize =
target.size - record.size` (positional since #13860, in 1.1.0 through 1.2.0)
and nothing bounds it; #19712's shape gives 8 and `JoinedGenericRecord` writes
the shifted slots silently. Seven of the eight callers are pinned to 0/5/6;
`HoodieBootstrapRecordIterator.java:67` is the exception and is already
malformed when a bootstrap writer schema gained more than one column.
Could we add `checkArgument(0 <= metaFieldSize && metaFieldSize <=
HOODIE_META_COLUMNS.size() + 1)` there, so the next mismatched caller fails
loudly instead of writing shifted values?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java:
##########
@@ -398,27 +402,62 @@ private static <R> Option<HoodieRecord<R>>
mergeIncomingWithExistingRecordWithEx
}
//record is inserted or updated
- String partitionPath = inferPartitionPath(incoming, existing,
writeSchemaWithMetaFields, keyGenerator, existingRecordContext, mergeResult);
+ String partitionPath = inferPartitionPath(incoming, existing,
mergedSchemaWithMetaFields, keyGenerator,
+ existingRecordContext, mergeResult, partitionResolvableFromRecord);
HoodieRecord<R> result =
existingRecordContext.constructHoodieRecord(mergeResult, partitionPath);
- HoodieRecord<R> withMeta = result.prependMetaFields(writeSchema,
writeSchemaWithMetaFields,
+ HoodieRecord<R> withMeta = result.prependMetaFields(mergedSchema,
mergedSchemaWithMetaFields,
new
MetadataValues().setRecordKey(incoming.getRecordKey()).setPartitionPath(partitionPath),
properties);
- return
Option.of(withMeta.wrapIntoHoodieRecordPayloadWithParams(writeSchemaWithMetaFields,
properties, Option.empty(),
- config.allowOperationMetadataField(), Option.empty(), false,
Option.of(writeSchema)));
+ return
Option.of(withMeta.wrapIntoHoodieRecordPayloadWithParams(mergedSchemaWithMetaFields,
properties, Option.empty(),
+ config.allowOperationMetadataField(), Option.empty(), false,
Option.of(mergedSchema)));
}
private static <R> String inferPartitionPath(HoodieRecord<R> incoming,
HoodieRecord<R> existing, HoodieSchema recordSchema, BaseKeyGenerator
keyGenerator,
- RecordContext<R> recordContext,
BufferedRecord<R> resultingBufferedRecord) {
+ RecordContext<R> recordContext,
BufferedRecord<R> resultingBufferedRecord,
+ boolean
partitionResolvableFromRecord) {
R record = resultingBufferedRecord.getRecord();
if (record == incoming.getData()) {
return incoming.getPartitionPath();
} else if (record == existing.getData()) {
return existing.getPartitionPath();
+ } else if (!partitionResolvableFromRecord) {
+ // A partial update carries only the fields named in the assignments, so
the partition fields are
+ // absent. The merge cannot have changed a field the record does not
carry, so the partition is the
+ // existing record's. Deriving it from the record instead resolves to
the default partition,
+ // because KeyGenUtils#getPartitionPath substitutes it for a value it
cannot find.
+ return existing.getPartitionPath();
} else {
// the merged record is not the same as either incoming or existing, so
we need to compute the partition path
return
keyGenerator.getPartitionPath(recordContext.convertToAvroRecord(record,
recordSchema));
}
}
+ /**
+ * Whether the merged record can yield a partition path on its own. False
only for a partial update whose
+ * schema omits a partition field; a full merge result always can, so those
paths keep resolving through
+ * the key generator. Evaluated once per stage, not per record.
+ */
+ private static boolean isPartitionResolvableFromRecord(HoodieSchema
recordSchema, BaseKeyGenerator keyGenerator) {
+ List<String> partitionPathFields = keyGenerator.getPartitionPathFields();
+ if (partitionPathFields == null || partitionPathFields.isEmpty()) {
+ // A non-partitioned table has nothing to resolve, so let the key
generator answer as before.
+ return true;
+ }
+ return partitionPathFields.stream()
+ .map(HoodieIndexUtils::partitionFieldRootName)
+ .allMatch(rootField -> rootField.isEmpty() ||
recordSchema.getField(rootField).isPresent());
+ }
+
+ /**
+ * The schema field name a partition-path config entry refers to. The custom
key generators keep the
+ * mandatory "field:TYPE" spec verbatim in getPartitionPathFields, so the
type suffix is stripped before
+ * the nested path is reduced to its root.
+ */
+ private static String partitionFieldRootName(String partitionPathField) {
Review Comment:
**nit, feel free to ignore** --
`CustomAvroKeyGenerator.getPartitionFieldAndKeyType(field).getLeft()` already
strips the `:TYPE` suffix, and the root split is `split("\\.")[0]`, which would
avoid re-adding the `HoodieAvroUtils` import that #17599 removed from this
file. Separately, `inferPartitionPath` (405) and this guard (565) are handed
`mergedSchemaWithMetaFields` while the record carries `mergedSchema`; inert on
the Avro path, but could we pass `mergedSchema` so the argument matches the
record?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestPartialUpdateForMergeInto.scala:
##########
@@ -732,12 +732,144 @@ class TestPartialUpdateForMergeInto extends
HoodieSparkSqlTestBase {
}
}
+ // A partial update names only the columns being changed, so the record key
is normally absent
+ // from the assignments. On MOR the global-index tagging stage merges the
incoming record with its
+ // existing version and then asks the key generator for the merged record's
partition path - and
+ // that merged record is materialised against WRITE_PARTIAL_UPDATE_SCHEMA,
which carries only the
+ // assigned columns. Resolving the partition path must therefore not also
require the record key.
+ //
+ // The index types split on whether that tagging stage runs at all:
GLOBAL_BLOOM and GLOBAL_SIMPLE
+ // set mayContainDuplicateLookup on MOR and so reach it, while the
record-index spellings pass
+ // false and short-circuit. Both cells are covered, so the fix is pinned
where it applies and the
+ // already-working path is guarded against regression. The source projects
the partition column in
+ // every case, to keep this independent of partition-column resolution
(ENG-46864).
+ Seq(
+ ("GLOBAL_BLOOM re-keying", Map(
+ "hoodie.index.type" -> "GLOBAL_BLOOM",
+ "hoodie.bloom.index.update.partition.path" -> "false")),
+ ("GLOBAL_SIMPLE re-keying", Map(
+ "hoodie.index.type" -> "GLOBAL_SIMPLE",
+ "hoodie.simple.index.update.partition.path" -> "false")),
+ ("RECORD_INDEX", Map(
+ "hoodie.index.type" -> "RECORD_INDEX",
+ "hoodie.record.index.update.partition.path" -> "false",
+ "hoodie.metadata.enable" -> "true",
+ "hoodie.metadata.record.index.enable" -> "true")),
+ ("GLOBAL_RECORD_LEVEL_INDEX", Map(
+ "hoodie.index.type" -> "GLOBAL_RECORD_LEVEL_INDEX",
+ "hoodie.record.index.update.partition.path" -> "false",
+ "hoodie.metadata.enable" -> "true",
+ "hoodie.metadata.record.index.enable" -> "true"))
+ ).foreach { case (label, indexConfs) =>
+ test(s"Test MOR partial update on a global index without assigning the
record key ($label)") {
+ withTempDir { tmp =>
+ withSQLConf(indexConfs.toSeq: _*) {
+ val tableName = generateTableName
+ val basePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ | create table $tableName (
+ | id bigint,
+ | name string,
+ | amount double,
+ | ts bigint,
+ | dt string
+ | ) using hudi
+ | partitioned by (dt)
+ | location '$basePath'
+ | tblproperties (
+ | type = 'mor',
+ | primaryKey = 'id',
+ | preCombineField = 'ts')
+ """.stripMargin)
+ spark.sql(s"insert into $tableName values (1, 'a', 10.0, 1,
'2026-08-11')")
+
+ // `id` is deliberately absent from the assignments - that is what
makes it partial.
+ spark.sql(
+ s"""
+ | merge into $tableName as t
+ | using (
+ | select 1L as id, 15.0 as amount, 200L as ts, '2026-08-11'
as dt
+ | ) as s
+ | on t.id = s.id
+ | when matched then update set t.amount = s.amount, t.ts = s.ts
+ """.stripMargin)
+
+ // The assigned columns change, the unassigned ones keep their
existing values, and the
+ // record stays in its own partition - asserted rather than merely
checking row count,
+ // since a dropped record leaves a table that also "looks unchanged".
+ checkAnswer(s"select id, name, amount, ts, dt,
_hoodie_partition_path from $tableName")(
+ Seq(1L, "a", 15.0, 200L, "2026-08-11", "dt=2026-08-11")
+ )
+
+ // The rows alone cannot distinguish a correct partial log block
from a full-schema one, and
+ // this fix is precisely about which schema the block's meta fields
are laid out against. So
+ // assert the block: IS_PARTIAL set, and the schema equal to the
meta fields prepended onto
+ // the assigned columns only. Without the fix the meta count is
inferred as
+ // targetSchema.size - record.size and the values land in the wrong
slots.
+ validateLogBlock(basePath, 1, Seq(Seq("amount", "ts")), isPartial =
true, "dt=2026-08-11")
+ }
+ }
+ }
+ }
+
+
+ // update.partition.path enabled is deliberately NOT in the matrix above:
for GLOBAL_BLOOM and
+ // GLOBAL_SIMPLE it makes useGlobalIndex true, which disables MOR partial
updates altogether
+ // (isPartialUpdateActionForMOR requires !useGlobalIndex), and MOR then
requires the record key to be
+ // assigned. So the statement is rejected at analysis time rather than
reaching the writer. Pinned
+ // here so the boundary is explicit and nobody mistakes this rejection for
the partial-update defect.
+ //
+ // NOTE the exemption is not universal: isGlobalIndexEnabled maps only
GLOBAL_SIMPLE, GLOBAL_BLOOM
+ // and RECORD_INDEX, so GLOBAL_RECORD_LEVEL_INDEX returns false there
regardless of its
+ // update.partition.path value, while
SparkMetadataTableGlobalRecordLevelIndex still honours that
+ // config. That combination therefore reaches the writer WITH partial
updates enabled. The
+ // partition-resolution guard in HoodieIndexUtils covers it, and the missing
enum mapping is being
+ // added separately, so this test asserts the rejection only for the
spellings that actually reject.
+ test("Test MOR merge without assigning the record key is rejected when the
global index updates the partition path") {
Review Comment:
**minor** -- this is a near-clone of the test at line 686: same table shape,
same statement, same `checkExceptionContain` on the message from
`MergeIntoHoodieTableCommand.scala:1051`, differing only in which
`isPartialUpdateActionForMOR` conjunct fails (custom merge mode there,
`useGlobalIndex` here). The `insert into` is dead, since `validate` runs at the
top of `run()` before any read, and the rejection is in `run()`, not analysis.
Could we fold this into 686 as a two-element `Seq(...).foreach`, or drop it?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestPartialUpdateForMergeInto.scala:
##########
@@ -732,12 +732,144 @@ class TestPartialUpdateForMergeInto extends
HoodieSparkSqlTestBase {
}
}
+ // A partial update names only the columns being changed, so the record key
is normally absent
+ // from the assignments. On MOR the global-index tagging stage merges the
incoming record with its
+ // existing version and then asks the key generator for the merged record's
partition path - and
+ // that merged record is materialised against WRITE_PARTIAL_UPDATE_SCHEMA,
which carries only the
+ // assigned columns. Resolving the partition path must therefore not also
require the record key.
+ //
+ // The index types split on whether that tagging stage runs at all:
GLOBAL_BLOOM and GLOBAL_SIMPLE
+ // set mayContainDuplicateLookup on MOR and so reach it, while the
record-index spellings pass
+ // false and short-circuit. Both cells are covered, so the fix is pinned
where it applies and the
+ // already-working path is guarded against regression. The source projects
the partition column in
+ // every case, to keep this independent of partition-column resolution
(ENG-46864).
+ Seq(
+ ("GLOBAL_BLOOM re-keying", Map(
Review Comment:
**minor** -- the four cells are two behaviours. `GLOBAL_BLOOM` and
`GLOBAL_SIMPLE` call `tagGlobalLocationBackToRecords` with identical flags
(`HoodieGlobalBloomIndex.java:106-109`, `HoodieGlobalSimpleIndex.java:77-80`),
and both record-index spellings map to
`SparkMetadataTableGlobalRecordLevelIndex`
(`SparkHoodieIndexFactory.java:65-67`). Also
`hoodie.metadata.record.index.enable` is a deprecated alias of
`hoodie.metadata.global.record.level.index.enable`; if it stops resolving, the
index falls back to `GLOBAL_SIMPLE` on a warning and the cell stays green.
Could we collapse to two cells and use the current key (or assert the
`record_index` partition exists)?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestPartialUpdateForMergeInto.scala:
##########
@@ -732,12 +732,144 @@ class TestPartialUpdateForMergeInto extends
HoodieSparkSqlTestBase {
}
}
+ // A partial update names only the columns being changed, so the record key
is normally absent
+ // from the assignments. On MOR the global-index tagging stage merges the
incoming record with its
+ // existing version and then asks the key generator for the merged record's
partition path - and
+ // that merged record is materialised against WRITE_PARTIAL_UPDATE_SCHEMA,
which carries only the
+ // assigned columns. Resolving the partition path must therefore not also
require the record key.
+ //
+ // The index types split on whether that tagging stage runs at all:
GLOBAL_BLOOM and GLOBAL_SIMPLE
+ // set mayContainDuplicateLookup on MOR and so reach it, while the
record-index spellings pass
+ // false and short-circuit. Both cells are covered, so the fix is pinned
where it applies and the
+ // already-working path is guarded against regression. The source projects
the partition column in
+ // every case, to keep this independent of partition-column resolution
(ENG-46864).
Review Comment:
**nit, feel free to ignore** -- comment accuracy: this paragraph describes
the #19708 key-generator problem rather than the meta-field layout this PR
fixes; `ENG-46864` is an internal tracker id; the `validateLogBlock` note at
805-809 overclaims, since it reads only the block header whose schema comes
from config on both sides of the fix (the discriminator is the merge no longer
throwing); the "re-keying" labels say the opposite of
`update.partition.path=false`; "short-circuit" should read "skip the merged
lookup" (those cells do enter the stage); double blank line at 815; commit
message and body say five cases for four.
Could we tidy these in one pass?
--
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]
