This is an automated email from the ASF dual-hosted git repository.
voonhous 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 633d1427f35f test(trino): add MoR read tests for delete markers,
custom payloads and commit-time ordering (#19295)
633d1427f35f is described below
commit 633d1427f35fd2d4a21c6e02363dc8dd90239237
Author: voonhous <[email protected]>
AuthorDate: Thu Jul 30 22:46:42 2026 +0800
test(trino): add MoR read tests for delete markers, custom payloads and
commit-time ordering (#19295)
* test(trino): add MoR read tests for delete markers, custom payloads and
commit-time ordering
Fixes apache/hudi#18898 (follow-up from the RFC-105 review of
HudiTrinoReaderContext.getRecordMerger): the merge-mode dispatch had no
end-to-end coverage where the merger choice matters. Adds runtime-written
v10 MOR tables and snapshot-read tests for:
- Deletes under EVENT_TIME_ORDERING (mor_deletes): hard deletes -- which at
v10 are native delete log files read back through the connector's own
getFileRecordIterator with the synthetic delete-log schema, a previously
untested path enabled by the required-column resolution this branch
stacks on -- plus _hoodie_is_deleted soft deletes, an OBSOLETE soft
delete and an OBSOLETE update (lower ordering value) that must LOSE
against the base row.
- COMMIT_TIME_ORDERING (mor_commit_time): a lower-ordering update that
must WIN (latest write), the exact mirror of the event-time case,
discriminating OverwriteWithLatestMerger from event-time merging.
- Payload-driven semantics without any hudi.record-merger-impls property:
AWSDmsAvroPayload (v10-translated to COMMIT_TIME + delete-key props;
Op='D' log records delete rows at merge time),
OverwriteNonDefaultsWithLatestAvroPayload (v10-translated to
PARTIAL_UPDATE_MODE=IGNORE_DEFAULTS; null update columns keep stored
values), and a user-defined SummingTestPayload riding the payload-based
CUSTOM strategy whose merged value (old + new) proves the payload's
combineAndGetUpdateValue executed at read time.
Test rows are written with the pass-through HoodieAvroPayload (not a
BaseAvroPayload) so delete-flagged rows land as data records and every
merge decision happens at read time from the table config.
Bugs found by these tests and fixed here:
- HudiUtil.mergeRequiredColumnNames read the custom delete key/marker from
raw table props, but v9+ table creation persists them PREFIXED
(hoodie.record.merge.property.*). The file-group reader strips the
prefix via getTableMergeProperties before DeleteContext reads the plain
keys, so its required schema included the delete column while the
connector's base-read prediction missed it and the base-read projection
guard failed every narrow projection on such tables. Now resolved the
same way the reader does: table merge properties first, raw props as
fallback. Unit-tested in TestHudiMergeRequiredColumns.
- buildColumnHandles matched predicted merge columns against metastore
data columns case-sensitively; the prediction carries the table schema's
casing (AWSDms hardcodes 'Op') while metastore names are lowercased, so
the handle was never built. Matching is now case-insensitive.
- buildRequiredColumnHandles emitted projection handles with the
lowercased metastore name on the log projection, but hudi-common reads
merge fields off log records by the table schema's exact field name
(Avro lookups are case-sensitive), so e.g. the DMS delete marker was
never seen on log records. Handles are now re-labeled with the
required-schema field's exact name; parquet columns resolve
case-insensitively either way.
- count(*) over a MoR split with log files failed with "Failed to
construct Avro schema": the query projects no columns, and
HudiPageSource's serializer fed the empty projection to
AvroHiveFileUtils.determineSchemaOrThrowException, which rejects an
empty column list. HudiUtil.constructSchema now short-circuits an empty
projection to an empty record schema; the page source already counts
positions fine with zero channels. HudiUtilTest's empty-columns test now
pins the empty-record schema instead of the former exception.
- Payload-based merging read garbage on the summing table ([, -44] instead
of [k1, 109]): records handed to hudi-common carried a schema
reconstructed from Hive metastore types (every column a nullable union,
fields in projection order), while the file-group reader tracks its
required schema for those records. BaseAvroPayload round-trips the newer
record through Avro binary with the tracked schema, so the structural
mismatch misaligned the binary decode. HudiTrinoReaderContext now builds
the log-side AND base-side serializers with requiredSchema itself, via a
new HudiAvroSerializer constructor that takes the record schema
explicitly and maps page channels to record fields BY NAME (the base
projection's order can differ from the required schema's field order;
containment holds in both directions through the existing base-read
guard and generateRequiredSchema always containing the requested
schema). This also aligns record layouts with the positions hudi-common
derives from the tracked schema (partial-update merging,
_hoodie_operation lookup).
- (hudi-common, engine-agnostic) The file-group reader passed its raw
input props to initRecordMerger, the MERGE_TYPE lookup and the schema
handler instead of the ConfigUtils.getMergeProps view, which layers the
table's prefixed merge properties (hoodie.record.merge.property.*) in
as plain keys -- a regression from commit 19961e77d539 replacing the
previous in-place putAll with a copy. DeleteContext therefore missed
the translated DMS delete key/marker, the delete column was not a
mandatory merge field, and Op='D' log records merged as regular data
updates instead of deletes on snapshot reads. Fixed in
HoodieFileGroupReader and HoodieLsmFileGroupReader alike by passing
this.props downstream.
Also re-ports a shim-lineage guard (trino repo commit 78209f83b50) that
the RFC-105 migration squash predates: HudiAvroSerializer's default
constructor now maps channels past hidden (synthesized, split-prefilled)
columns instead of relying on serialize() never being called with them;
with identity positions a hidden column would shift every later value into
the wrong field and overrun the record.
Also replaces the TODO(apache/hudi#18898) in getRecordMerger with a
pointer to the new suites.
* test(trino): address review comments on MoR merge-semantics read tests
- pass the payload class to getTableMergeProperties (the no-arg overload
does not exist; the module profile being off by default hid the break)
- drop the buildColumnHandles case-folding and the log-projection handle
re-label: appendMissingMergeRequiredColumns and the
requiredSchema-stamping
serializer already cover both
- stop building a record schema in the page-building serializer and revert
the constructSchema empty-list special case; the count(*) path stays
pinned by TestHudiMorMergeModeSemantics
- scope the getRecordMerger comment to the CUSTOM arm, the only return
value the file-group reader dereferences for merging
- split the two monolithic fixtures into per-table subclasses of
AbstractMergerHudiTablesInitializer composed with
CompositeHudiTablesInitializer
* test(trino): address round-2 review nits on MoR merge-semantics tests
- qualify every fixture's TABLE_NAME/RT_TABLE_NAME in both suites so each
assertion names the table it checks (no mixed static imports)
- replace the duplicated PARTITION_PATH constants in the delete-writing
fixtures with a hoodieKey(String) helper on
AbstractMergerHudiTablesInitializer
- correct the getRecordMerger comment: the ordering arms ARE reachable via
partialMerge on IS_PARTIAL log blocks, just not covered by these suites;
restore the TODO, now pointing at follow-up issue apache/hudi#19413
* test(trino): address round-3 review comments on MoR merge-semantics tests
- rename testPrefixedDeleteKeyAndMarkerAreRequested to
testPrefixedDeleteKeyIsRequested: only the delete key is requested,
the marker is a precondition
- reuse AWSDmsAvroPayload.OP_FIELD / DELETE_OPERATION_VALUE in the DMS
fixture instead of local Op / D literals
- add a non-marker Op='U' log record for k1 in the DMS fixture so a
broken marker comparison fails the suite (k1 must update, not delete)
- flip the five new table names to trail with _mor (deletes_mor,
commit_time_mor, dms_mor, overwrite_non_defaults_mor, summing_mor)
to match the existing fixtures in the package
- cover the CUSTOM merge arm's delete path: the summing fixture gains a
k2 base row and a hard-delete commit, pinned by
testSummingPayloadHardDeleteRemovesRowOnSnapshotRead
* test(trino): address round-4 review comment on MoR merge-semantics tests
The hard-delete comments claimed the delete reaches the payload arm as
an empty payload whose combineAndGetUpdateValue returns empty. In fact
HoodieAvroRecordMerger.merge returns the delete on its
isCommitTimeOrderingDelete short-circuit (writeClient.delete records
carry the sentinel ordering value) before loadPayload runs, so no
payload is constructed. Say so in the initializer javadoc, the commit
comment and the test method comment.
---
.../main/java/io/trino/plugin/hudi/HudiUtil.java | 15 +-
.../plugin/hudi/reader/HudiTrinoReaderContext.java | 33 +++--
.../trino/plugin/hudi/util/HudiAvroSerializer.java | 41 +++++-
.../plugin/hudi/TestHudiMergeRequiredColumns.java | 15 ++
.../plugin/hudi/TestHudiMorMergeModeSemantics.java | 135 ++++++++++++++++++
.../plugin/hudi/TestHudiMorPayloadSemantics.java | 129 +++++++++++++++++
.../AbstractMergerHudiTablesInitializer.java | 8 +-
.../CommitTimeOrderingHudiTablesInitializer.java | 129 +++++++++++++++++
.../testing/DmsPayloadHudiTablesInitializer.java | 135 ++++++++++++++++++
.../EventTimeDeletesHudiTablesInitializer.java | 156 +++++++++++++++++++++
...iteNonDefaultsPayloadHudiTablesInitializer.java | 123 ++++++++++++++++
.../SummingPayloadHudiTablesInitializer.java | 131 +++++++++++++++++
.../plugin/hudi/testing/SummingTestPayload.java | 72 ++++++++++
13 files changed, 1103 insertions(+), 19 deletions(-)
diff --git a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java
index 4d339e7d56c4..4bcd3a65ec20 100644
--- a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java
+++ b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java
@@ -651,8 +651,14 @@ public final class HudiUtil
// Delete markers and the operation field decide record deletion at
merge time
requiredColumnNames.add(HOODIE_IS_DELETED_FIELD);
requiredColumnNames.add(OPERATION_METADATA_FIELD);
- String deleteKey = tableConfig.getProps().getProperty(DELETE_KEY);
- String deleteMarker =
tableConfig.getProps().getProperty(DELETE_MARKER);
+ // Resolve the delete key/marker the way the file-group reader does
(ConfigUtils.getMergeProps):
+ // table merge properties first -- v9+ tables persist these PREFIXED
(hoodie.record.merge.property.*)
+ // and getTableMergeProperties strips the prefix and bridges legacy
delete payloads (keyed on the
+ // payload class, which for the read path resolves from the table
config) -- falling back to the raw
+ // table props, where pre-prefix tables and reader/write configs carry
the plain keys.
+ Map<String, String> tableMergeProps =
tableConfig.getTableMergeProperties(tableConfig.getPayloadClass());
+ String deleteKey = tableMergeProps.getOrDefault(DELETE_KEY,
tableConfig.getProps().getProperty(DELETE_KEY));
+ String deleteMarker = tableMergeProps.getOrDefault(DELETE_MARKER,
tableConfig.getProps().getProperty(DELETE_MARKER));
// DeleteContext only honors a custom delete key when the marker value
is also set
if (!StringUtils.isNullOrEmpty(deleteKey) &&
!StringUtils.isNullOrEmpty(deleteMarker)) {
requiredColumnNames.add(deleteKey);
@@ -715,7 +721,10 @@ public final class HudiUtil
/**
* Builds {@link HiveColumnHandle}s, preserving physical (data-column)
index, for the data columns whose names
* appear in {@code columnNames}. Names that are not data columns (e.g.
Hudi meta fields) or whose types are not
- * supported by the storage format are skipped.
+ * supported by the storage format are skipped. Matching is
case-sensitive: a predicted merge column that carries
+ * the table schema's field casing (e.g. the {@code Op} delete key of DMS
tables) misses the lowercased metastore
+ * name here and is recovered from the table schema by {@link
#appendMissingMergeRequiredColumns} on the merge
+ * read path.
*/
private static List<HiveColumnHandle> buildColumnHandles(Table table,
TypeManager typeManager, Set<String> columnNames, HiveTimestampPrecision
timestampPrecision)
{
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/reader/HudiTrinoReaderContext.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/reader/HudiTrinoReaderContext.java
index 57e8ad5d288c..cc812a9a466f 100644
---
a/hudi-trino/src/main/java/io/trino/plugin/hudi/reader/HudiTrinoReaderContext.java
+++
b/hudi-trino/src/main/java/io/trino/plugin/hudi/reader/HudiTrinoReaderContext.java
@@ -61,7 +61,7 @@ public class HudiTrinoReaderContext
extends HoodieReaderContext<IndexedRecord>
{
private final ConnectorPageSource pageSource;
- private final HudiAvroSerializer avroSerializer;
+ private final List<HiveColumnHandle> columnHandles;
private final PrefilledColumnValues prefilledColumnValues;
private final LogFileParquetPageSourceFactory logPageSourceFactory;
private final Map<String, HiveColumnHandle> colNameToHandle;
@@ -88,7 +88,7 @@ public class HudiTrinoReaderContext
super(storageConfiguration, tableConfig, Option.empty(),
Option.empty(), new AvroRecordContext(tableConfig,
tableConfig.getPayloadClass()));
this.pageSource = pageSource;
this.prefilledColumnValues = prefilledColumnValues;
- this.avroSerializer = new HudiAvroSerializer(columnHandles,
prefilledColumnValues);
+ this.columnHandles = columnHandles;
this.logPageSourceFactory = logPageSourceFactory;
this.colNameToHandle = new HashMap<>();
for (HiveColumnHandle handle : columnHandles) {
@@ -129,6 +129,11 @@ public class HudiTrinoReaderContext
* fresh page source is built on demand with predicate pushdown disabled
so every log record is read and
* merged; for the base file the pre-built base page source is reused.
Classic Avro log blocks never reach
* here (they deserialize inline).
+ * <p>
+ * Both paths emit records that CARRY {@code requiredSchema}: the
file-group reader tracks that schema for
+ * every buffered record, and payload-based merging round-trips records
through Avro binary with it
+ * ({@code BaseAvroPayload}), so a record whose own schema differs (in
field order or nullability) would
+ * decode into garbage values there.
*/
private ClosableIterator<IndexedRecord> getFileRecordIterator(
StoragePath path,
@@ -143,7 +148,7 @@ public class HudiTrinoReaderContext
}
List<HiveColumnHandle> logProjection =
buildRequiredColumnHandles(requiredSchema);
ConnectorPageSource logSource =
logPageSourceFactory.create(path.toString(), start, length, logProjection);
- HudiAvroSerializer logSerializer = new
HudiAvroSerializer(logProjection, prefilledColumnValues);
+ HudiAvroSerializer logSerializer = new
HudiAvroSerializer(logProjection, prefilledColumnValues,
requiredSchema.toAvroSchema());
return createRecordIterator(logSource, logSerializer);
}
// The base read reuses the pre-built page source, so it can only
satisfy requiredSchema fields the
@@ -161,7 +166,10 @@ public class HudiTrinoReaderContext
+
"FileGroupReaderSchemaHandler.generateRequiredSchema.",
missingColumns));
}
- return createRecordIterator(pageSource, avroSerializer);
+ // Every requiredSchema field is in the base projection (checked
above) and every projected column is
+ // a requiredSchema field (the file-group reader's required schema
always contains the full requested
+ // schema), so the by-name channel mapping resolves in both directions.
+ return createRecordIterator(pageSource, new
HudiAvroSerializer(columnHandles, prefilledColumnValues,
requiredSchema.toAvroSchema()));
}
/**
@@ -176,6 +184,11 @@ public class HudiTrinoReaderContext
{
List<HiveColumnHandle> handles = new ArrayList<>();
for (HoodieSchemaField field : requiredSchema.getFields()) {
+ // A projection handle may carry the lowercased metastore column
name (e.g. 'op' for the
+ // DMS delete key 'Op'); that is fine on this path: parquet
columns resolve
+ // case-insensitively, the serializer maps channels into
requiredSchema positions
+ // case-insensitively, and hudi-common reads merge fields off the
record's own schema,
+ // which serialize() stamps with the table casing.
handles.add(colNameToHandle.computeIfAbsent(
field.name().toLowerCase(Locale.ROOT),
_ -> HudiUtil.toColumnHandle(field)));
@@ -241,11 +254,13 @@ public class HudiTrinoReaderContext
protected Option<HoodieRecordMerger> getRecordMerger(RecordMergeMode
mergeMode, String mergeStrategyId, String mergeImplClasses)
{
// Dispatch on the table's merge mode, mirroring
HoodieAvroReaderContext. The Trino reader
- // operates on IndexedRecord, so the Avro mergers apply directly.
Using the read-time merger
- // (combineAndGetUpdateValue) rather than a fixed preCombine merger
keeps COMMIT_TIME_ORDERING
- // and custom-payload tables correct on MoR reads.
- // TODO(apache/hudi#18898): add MoR read tests for delete markers and
custom payloads to
- // exercise the EVENT_TIME_ORDERING (combineAndGetUpdateValue) and
CUSTOM branches below.
+ // operates on IndexedRecord, so the Avro mergers apply directly. The
CUSTOM arm's return
+ // value drives merging (combineAndGetUpdateValue at read time,
covered end-to-end by
+ // TestHudiMorPayloadSemantics; apache/hudi#18898). The ordering arms'
mergers are reached
+ // through partialMerge when a log block carries IS_PARTIAL
(BufferedRecordMergerFactory),
+ // which these suites do not cover; TestHudiMorMergeModeSemantics pins
the mode semantics
+ // themselves. TODO(apache/hudi#19413): cover the ordering arms'
partialMerge path
+ // (IS_PARTIAL log blocks) once the Avro mergers implement it.
switch (mergeMode) {
case EVENT_TIME_ORDERING:
return Option.of(new HoodieAvroRecordMerger());
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
index 479150748928..0d9bc9f5978b 100644
--- a/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
+++ b/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
@@ -60,9 +60,9 @@ import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;
+import static com.google.common.base.Preconditions.checkState;
import static com.google.common.base.Verify.verify;
import static io.airlift.slice.Slices.utf8Slice;
-import static io.trino.plugin.hudi.HudiUtil.constructSchema;
import static io.trino.plugin.hudi.HudiUtil.getFieldFromSchema;
import static io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
import static io.trino.spi.StandardErrorCode.NUMERIC_VALUE_OUT_OF_RANGE;
@@ -111,24 +111,53 @@ public class HudiAvroSerializer
private final List<HiveColumnHandle> columnHandles;
private final List<Type> columnTypes;
+ // Both are null for a page-building-only serializer (the two-arg
constructor): buildRecordInPage
+ // reads field positions off each record's own schema, so no record schema
is needed there -- and
+ // none could be built for hidden (synthesized) columns, which are
answered from the split by
+ // PrefilledColumnValues rather than read from the file. serialize()
requires the three-arg
+ // constructor, which maps page channel i to record position
channelToFieldPosition[i].
private final Schema schema;
+ private final int[] channelToFieldPosition;
public HudiAvroSerializer(List<HiveColumnHandle> columnHandles,
PrefilledColumnValues prefilledColumnValues)
{
this.columnHandles = columnHandles;
this.columnTypes =
columnHandles.stream().map(HiveColumnHandle::getType).toList();
- // Fetches projected schema
- this.schema = constructSchema(columnHandles.stream().filter(ch ->
!ch.isHidden()).map(HiveColumnHandle::getName).toList(),
- columnHandles.stream().filter(ch ->
!ch.isHidden()).map(HiveColumnHandle::getHiveType).toList());
this.prefilledColumnValues = prefilledColumnValues;
+ this.schema = null;
+ this.channelToFieldPosition = null;
+ }
+
+ /**
+ * Builds a serializer whose {@link #serialize} records carry {@code
recordSchema} -- the exact
+ * schema hudi-common tracks for the records of this read (the file-group
reader's required
+ * schema) -- instead of a schema reconstructed from the projection's Hive
types. The
+ * reconstruction differs from the table's real schema (every Hive column
becomes a nullable
+ * union, fields follow projection order), and payload-based merging
round-trips the record
+ * through Avro BINARY with the tracked schema ({@code BaseAvroPayload}),
where any structural
+ * difference misaligns the decode and yields garbage values. Page
channels are matched to
+ * record fields BY NAME, so the projection may order columns differently
from the schema;
+ * every projected column must be a field of {@code recordSchema}.
+ */
+ public HudiAvroSerializer(List<HiveColumnHandle> columnHandles,
PrefilledColumnValues prefilledColumnValues, Schema recordSchema)
+ {
+ this.columnHandles = columnHandles;
+ this.columnTypes =
columnHandles.stream().map(HiveColumnHandle::getType).toList();
+ this.schema = recordSchema;
+ this.prefilledColumnValues = prefilledColumnValues;
+ int[] mapping = new int[columnHandles.size()];
+ for (int i = 0; i < columnHandles.size(); i++) {
+ mapping[i] = getFieldFromSchema(columnHandles.get(i).getName(),
recordSchema).pos();
+ }
+ this.channelToFieldPosition = mapping;
}
public IndexedRecord serialize(Page sourcePage, int position)
{
+ checkState(schema != null, "serialize() requires a serializer built
with a record schema");
IndexedRecord record = new GenericData.Record(schema);
for (int i = 0; i < columnTypes.size(); i++) {
- Object value = getValue(sourcePage, i, position);
- record.put(i, value);
+ record.put(channelToFieldPosition[i], getValue(sourcePage, i,
position));
}
return record;
}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMergeRequiredColumns.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMergeRequiredColumns.java
index f6a4ffb874d0..2abc2c680b08 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMergeRequiredColumns.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMergeRequiredColumns.java
@@ -89,6 +89,21 @@ class TestHudiMergeRequiredColumns
.doesNotContain("op");
}
+ @Test
+ public void testPrefixedDeleteKeyIsRequested()
+ {
+ // v9+ table creation persists the delete key/marker under the
hoodie.record.merge.property.
+ // prefix (e.g. for AWSDmsAvroPayload tables); the file-group reader
strips the prefix via
+ // getTableMergeProperties() before DeleteContext reads the plain
keys, and the connector's
+ // prediction must see the same values or the base-read projection
guard fires on narrow queries
+ HoodieTableConfig tableConfig = new HoodieTableConfig();
+ tableConfig.setValue(HoodieTableConfig.RECORD_MERGE_PROPERTY_PREFIX +
DELETE_KEY, "Op");
+ tableConfig.setValue(HoodieTableConfig.RECORD_MERGE_PROPERTY_PREFIX +
DELETE_MARKER, "D");
+
+ assertThat(mergeRequiredColumnNames(tableConfig,
RecordMergeMode.COMMIT_TIME_ORDERING))
+ .contains("Op");
+ }
+
@Test
public void testRecordKeyFieldsRequestedWithoutPopulatedMetaFields()
{
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMorMergeModeSemantics.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMorMergeModeSemantics.java
new file mode 100644
index 000000000000..ba21f2b34a67
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMorMergeModeSemantics.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi;
+
+import io.trino.plugin.hudi.testing.CommitTimeOrderingHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.CompositeHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.EventTimeDeletesHudiTablesInitializer;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.QueryRunner;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * End-to-end MoR snapshot-read tests for the merge-mode dispatch in
+ * {@code HudiTrinoReaderContext.getRecordMerger} with deletes (issue
apache/hudi#18898), on tables
+ * written by {@link EventTimeDeletesHudiTablesInitializer} and
+ * {@link CommitTimeOrderingHudiTablesInitializer}:
+ * <ul>
+ * <li>EVENT_TIME_ORDERING: updates and soft deletes apply only when their
ordering value wins;
+ * obsolete (lower-ordering) updates and soft deletes must LOSE against
the base row.</li>
+ * <li>Hard deletes (native delete log files, read back through the
connector's own
+ * {@code getFileRecordIterator}) always win.</li>
+ * <li>COMMIT_TIME_ORDERING: the latest write wins even with a LOWER
ordering value -- the exact
+ * mirror of the event-time obsolete-update case, discriminating the two
merger dispatches.</li>
+ * </ul>
+ */
+public class TestHudiMorMergeModeSemantics
+ extends AbstractTestQueryFramework
+{
+ @Override
+ protected QueryRunner createQueryRunner()
+ throws Exception
+ {
+ return HudiQueryRunner.builder()
+ .setDataLoader(new CompositeHudiTablesInitializer(
+ new EventTimeDeletesHudiTablesInitializer(),
+ new CommitTimeOrderingHudiTablesInitializer()))
+ .build();
+ }
+
+ @Test
+ public void testReadOptimizedShowsAllBaseRows()
+ {
+ // Deletes and updates live in log files only; the read-optimized
tables reflect the base commit
+ assertQuery(
+ "SELECT key, name, value FROM " +
EventTimeDeletesHudiTablesInitializer.TABLE_NAME + " ORDER BY key",
+ "VALUES ('k1', 'k1_base', CAST(10 AS BIGINT)), ('k2',
'k2_base', 20), ('k3', 'k3_base', 30),"
+ + " ('k4', 'k4_base', 40), ('k5', 'k5_base', 50),
('k6', 'k6_base', 60)");
+ assertQuery(
+ "SELECT key, name, value FROM " +
CommitTimeOrderingHudiTablesInitializer.TABLE_NAME + " ORDER BY key",
+ "VALUES ('k1', 'k1_base', CAST(10 AS BIGINT)), ('k2',
'k2_base', 20), ('k3', 'k3_base', 30)");
+ }
+
+ @Test
+ public void testEventTimeMergeWithDeletes()
+ {
+ // k1: higher-ts update wins; k2: hard-deleted; k3: soft-deleted
(higher ts);
+ // k4: OBSOLETE soft delete (lower ts) -> base row survives;
+ // k5: untouched; k6: OBSOLETE update (lower ts) -> base row survives
+ assertQuery(
+ "SELECT key, name, value FROM " +
EventTimeDeletesHudiTablesInitializer.RT_TABLE_NAME + " ORDER BY key",
+ "VALUES ('k1', 'k1_updated', CAST(11 AS BIGINT)), ('k4',
'k4_base', 40),"
+ + " ('k5', 'k5_base', 50), ('k6', 'k6_base', 60)");
+ }
+
+ @Test
+ public void testHardDeleteRemovesRowOnSnapshotRead()
+ {
+ // The hard delete is a native delete log file, resolved through the
connector's
+ // getFileRecordIterator with the synthetic delete-log schema (record
key + ordering field)
+ assertThat(computeScalar("SELECT count(*) FROM " +
EventTimeDeletesHudiTablesInitializer.RT_TABLE_NAME + " WHERE key = 'k2'"))
+ .isEqualTo(0L);
+ assertThat(computeScalar("SELECT count(*) FROM " +
EventTimeDeletesHudiTablesInitializer.TABLE_NAME + " WHERE key = 'k2'"))
+ .isEqualTo(1L);
+ }
+
+ @Test
+ public void testSoftDeleteRemovesRowOnSnapshotRead()
+ {
+ // _hoodie_is_deleted=true log record with a winning (higher) ordering
value
+ assertThat(computeScalar("SELECT count(*) FROM " +
EventTimeDeletesHudiTablesInitializer.RT_TABLE_NAME + " WHERE key = 'k3'"))
+ .isEqualTo(0L);
+ assertThat(computeScalar("SELECT count(*) FROM " +
EventTimeDeletesHudiTablesInitializer.TABLE_NAME + " WHERE key = 'k3'"))
+ .isEqualTo(1L);
+ }
+
+ @Test
+ public void testObsoleteSoftDeleteLosesUnderEventTimeOrdering()
+ {
+ // The k4 soft delete carries ts=50 < base ts=100: event-time merging
must keep the base row
+ assertQuery(
+ "SELECT key, name, value FROM " +
EventTimeDeletesHudiTablesInitializer.RT_TABLE_NAME + " WHERE key = 'k4'",
+ "VALUES ('k4', 'k4_base', CAST(40 AS BIGINT))");
+ }
+
+ @Test
+ public void testCommitTimeOrderingKeepsLatestWrite()
+ {
+ // k1's update carries ts=50 < base ts=100. Under COMMIT_TIME_ORDERING
the LATEST WRITE wins
+ // regardless of the ordering value -- the mirror of the event-time k6
case, where the same
+ // shape keeps the BASE row. Together they discriminate the two merger
dispatches.
+ assertQuery(
+ "SELECT key, name, value FROM " +
CommitTimeOrderingHudiTablesInitializer.RT_TABLE_NAME + " ORDER BY key",
+ "VALUES ('k1', 'k1_updated', CAST(11 AS BIGINT)), ('k3',
'k3_base', 30)");
+ }
+
+ @Test
+ public void testCountAfterDeletes()
+ {
+ assertThat(computeScalar("SELECT count(*) FROM " +
EventTimeDeletesHudiTablesInitializer.RT_TABLE_NAME)).isEqualTo(4L);
+ assertThat(computeScalar("SELECT count(*) FROM " +
CommitTimeOrderingHudiTablesInitializer.RT_TABLE_NAME)).isEqualTo(2L);
+ }
+
+ @Test
+ public void testNarrowProjectionMergesCorrectly()
+ {
+ // Neither the ordering field nor _hoodie_is_deleted is projected; the
connector must still read
+ // them on both the base and log sides for the merge to resolve
updates and deletes correctly
+ assertQuery(
+ "SELECT key, value FROM " +
EventTimeDeletesHudiTablesInitializer.RT_TABLE_NAME + " ORDER BY key",
+ "VALUES ('k1', CAST(11 AS BIGINT)), ('k4', 40), ('k5', 50),
('k6', 60)");
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMorPayloadSemantics.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMorPayloadSemantics.java
new file mode 100644
index 000000000000..2183d80fe686
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMorPayloadSemantics.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi;
+
+import io.trino.plugin.hudi.testing.CompositeHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.DmsPayloadHudiTablesInitializer;
+import
io.trino.plugin.hudi.testing.OverwriteNonDefaultsPayloadHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.SummingPayloadHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.SummingTestPayload;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.QueryRunner;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * End-to-end MoR snapshot-read tests for PAYLOAD-driven merge semantics
(issue apache/hudi#18898), on
+ * tables written by {@link DmsPayloadHudiTablesInitializer},
+ * {@link OverwriteNonDefaultsPayloadHudiTablesInitializer} and {@link
SummingPayloadHudiTablesInitializer}.
+ * No {@code hudi.record-merger-impls} connector property is set anywhere --
every behavior below must
+ * resolve purely from the table config:
+ * <ul>
+ * <li>AWSDms: a log record with {@code Op='D'} deletes the row at merge
time via the translated
+ * delete-key/marker table properties, while a log record with the
non-marker {@code Op='U'} must
+ * apply as an update (the narrow-projection case pins the fix that
reads those properties with
+ * their {@code hoodie.record.merge.property.} prefix).</li>
+ * <li>OverwriteNonDefaults: IGNORE_DEFAULTS partial merging keeps the
stored value for update columns
+ * equal to the schema default (null).</li>
+ * <li>{@link SummingTestPayload}: a user-defined payload rides the
payload-based CUSTOM merge
+ * strategy; the merged value is the SUM of stored and incoming values,
which proves the payload's
+ * {@code combineAndGetUpdateValue} executed (overwrite would yield the
incoming value), and a hard
+ * delete routed through the same arm must remove its row.</li>
+ * </ul>
+ */
+public class TestHudiMorPayloadSemantics
+ extends AbstractTestQueryFramework
+{
+ @Override
+ protected QueryRunner createQueryRunner()
+ throws Exception
+ {
+ return HudiQueryRunner.builder()
+ .setDataLoader(new CompositeHudiTablesInitializer(
+ new DmsPayloadHudiTablesInitializer(),
+ new OverwriteNonDefaultsPayloadHudiTablesInitializer(),
+ new SummingPayloadHudiTablesInitializer()))
+ .build();
+ }
+
+ @Test
+ public void testDmsDeleteMarkerRemovesRowOnSnapshotRead()
+ {
+ // Read-optimized: both rows (the log records are not merged)
+ assertQuery(
+ "SELECT key, name, value, Op FROM " +
DmsPayloadHudiTablesInitializer.TABLE_NAME + " ORDER BY key",
+ "VALUES ('k1', 'k1_base', CAST(10 AS BIGINT), 'I'), ('k2',
'k2_base', 20, 'I')");
+ // Snapshot: k2 is deleted by the Op='D' log record via the
delete-key/marker table properties,
+ // while k1's NON-marker Op='U' log record must apply as an update --
a marker comparison that
+ // fires on any non-null Op would wrongly delete k1 too
+ assertQuery(
+ "SELECT key, name, value, Op FROM " +
DmsPayloadHudiTablesInitializer.RT_TABLE_NAME + " ORDER BY key",
+ "VALUES ('k1', 'k1_updated', CAST(11 AS BIGINT), 'U')");
+ }
+
+ @Test
+ public void testDmsNarrowProjectionMergesCorrectly()
+ {
+ // The Op column is NOT projected, so the connector must predict it as
a merge-required column
+ // from the PREFIXED table properties
(hoodie.record.merge.property.hoodie.payload.delete.field)
+ // for the base read -- the regression this suite pins for
HudiUtil.mergeRequiredColumnNames
+ assertQuery(
+ "SELECT key, value FROM " +
DmsPayloadHudiTablesInitializer.RT_TABLE_NAME + " ORDER BY key",
+ "VALUES ('k1', CAST(11 AS BIGINT))");
+ assertThat(computeScalar("SELECT count(*) FROM " +
DmsPayloadHudiTablesInitializer.RT_TABLE_NAME)).isEqualTo(1L);
+ }
+
+ @Test
+ public void testOverwriteNonDefaultsKeepsStoredValueForDefaultColumns()
+ {
+ // Read-optimized: base values
+ assertQuery(
+ "SELECT key, a, b FROM " +
OverwriteNonDefaultsPayloadHudiTablesInitializer.TABLE_NAME,
+ "VALUES ('k1', 'base_a', 'base_b')");
+ // Snapshot: the update carried a='new_a' and b=null (the schema
default); IGNORE_DEFAULTS
+ // partial merging takes the update's a but keeps the STORED b
+ assertQuery(
+ "SELECT key, a, b FROM " +
OverwriteNonDefaultsPayloadHudiTablesInitializer.RT_TABLE_NAME,
+ "VALUES ('k1', 'new_a', 'base_b')");
+ }
+
+ @Test
+ public void testSummingPayloadRunsCombineAndGetUpdateValueOnRead()
+ {
+ // Read-optimized: the base values
+ assertQuery(
+ "SELECT key, value FROM " +
SummingPayloadHudiTablesInitializer.TABLE_NAME + " ORDER BY key",
+ "VALUES ('k1', CAST(10 AS BIGINT)), ('k2', 20)");
+ // Snapshot: 10 + 99 = 109 -- only the payload's
combineAndGetUpdateValue can produce this
+ // (newest-wins would yield 99, base-only 10), proving the CUSTOM
payload-strategy branch ran;
+ // k2 is hard-deleted
+ assertQuery(
+ "SELECT key, value FROM " +
SummingPayloadHudiTablesInitializer.RT_TABLE_NAME + " ORDER BY key",
+ "VALUES ('k1', CAST(109 AS BIGINT))");
+ }
+
+ @Test
+ public void testSummingPayloadHardDeleteRemovesRowOnSnapshotRead()
+ {
+ // The hard delete is a native delete log record routed to the
payload-based CUSTOM merge arm,
+ // where it wins on HoodieAvroRecordMerger's
isCommitTimeOrderingDelete short-circuit (the
+ // delete carries the sentinel ordering value) -- the delete path of
the user-merger dispatch,
+ // which both ordering arms already cover
+ assertThat(computeScalar("SELECT count(*) FROM " +
SummingPayloadHudiTablesInitializer.RT_TABLE_NAME + " WHERE key = 'k2'"))
+ .isEqualTo(0L);
+ assertThat(computeScalar("SELECT count(*) FROM " +
SummingPayloadHudiTablesInitializer.TABLE_NAME + " WHERE key = 'k2'"))
+ .isEqualTo(1L);
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java
index 045e8a3386ea..2aa39bf1cffa 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java
@@ -206,7 +206,13 @@ public abstract class AbstractMergerHudiTablesInitializer
protected static HoodieRecord<HoodieAvroPayload> avroRecord(GenericRecord
record, String key)
{
- return new HoodieAvroRecord<>(new HoodieKey(key, PARTITION_PATH), new
HoodieAvroPayload(Option.of(record)), null);
+ return new HoodieAvroRecord<>(hoodieKey(key), new
HoodieAvroPayload(Option.of(record)), null);
+ }
+
+ /** Addresses a record in the single unnamed partition, e.g. for hard
deletes via {@code writeClient.delete}. */
+ protected static HoodieKey hoodieKey(String key)
+ {
+ return new HoodieKey(key, PARTITION_PATH);
}
/** Mirrors the staged table into the Trino filesystem so the connector
observes the commits written so far. */
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/CommitTimeOrderingHudiTablesInitializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/CommitTimeOrderingHudiTablesInitializer.java
new file mode 100644
index 000000000000..8fa5c7fb3d79
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/CommitTimeOrderingHudiTablesInitializer.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.metastore.Column;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table in {@link
RecordMergeMode#COMMIT_TIME_ORDERING} that
+ * exercises the read-side merge-mode dispatch (issue apache/hudi#18898). ONLY
a record merge mode is set
+ * (no payload class), so table creation persists the mode as-is, which is
exactly the dispatch input
+ * {@code HudiTrinoReaderContext.getRecordMerger} switches on.
+ * <p>
+ * A base commit is followed by a log commit whose update carries an ordering
value LOWER than the base
+ * row's: latest-write-wins must KEEP the update, the exact mirror of the
event-time obsolete-update case
+ * in {@link EventTimeDeletesHudiTablesInitializer}, which is what
discriminates the two merger dispatches.
+ * A final commit hard-deletes a key ({@code writeClient.delete}); commit-time
deletes always win. See
+ * {@code TestHudiMorMergeModeSemantics}.
+ */
+public class CommitTimeOrderingHudiTablesInitializer
+ extends AbstractMergerHudiTablesInitializer
+{
+ public static final String TABLE_NAME = "commit_time_mor";
+ public static final String RT_TABLE_NAME = TABLE_NAME + "_rt";
+
+ public CommitTimeOrderingHudiTablesInitializer()
+ {
+ super(TABLE_NAME);
+ }
+
+ @Override
+ protected List<Column> dataColumns()
+ {
+ return ImmutableList.of(
+ new Column(RECORD_KEY_FIELD, HIVE_STRING, Optional.empty(),
Map.of()),
+ new Column("name", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("value", HIVE_LONG, Optional.empty(), Map.of()),
+ new Column(ORDERING_FIELD, HIVE_LONG, Optional.empty(),
Map.of()));
+ }
+
+ @Override
+ protected Schema avroSchema()
+ {
+ List<Schema.Field> fields = ImmutableList.of(
+ new Schema.Field(RECORD_KEY_FIELD,
Schema.create(Schema.Type.STRING)),
+ new Schema.Field("name", Schema.create(Schema.Type.STRING)),
+ new Schema.Field("value", Schema.create(Schema.Type.LONG)),
+ new Schema.Field(ORDERING_FIELD,
Schema.create(Schema.Type.LONG)));
+ return Schema.createRecord(TABLE_NAME, null, null, false, new
ArrayList<>(fields));
+ }
+
+ @Override
+ protected void configureTableConfig(HoodieTableMetaClient.TableBuilder
tableBuilder)
+ {
+ tableBuilder.setRecordMergeMode(RecordMergeMode.COMMIT_TIME_ORDERING);
+ }
+
+ @Override
+ protected void configureWriteConfig(HoodieWriteConfig.Builder
writeConfigBuilder)
+ {
+
writeConfigBuilder.withRecordMergeMode(RecordMergeMode.COMMIT_TIME_ORDERING);
+ }
+
+ @Override
+ protected void
writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ {
+ Schema schema = avroSchema();
+ // First commit: base parquet file with 3 keys at ts 100.
+ String firstCommit = client.startCommit();
+ List<WriteStatus> firstStatuses = client.bulkInsert(ImmutableList.of(
+ record(schema, "k1", "k1_base", 10L, 100L),
+ record(schema, "k2", "k2_base", 20L, 100L),
+ record(schema, "k3", "k3_base", 30L, 100L)), firstCommit);
+ client.commit(firstCommit, firstStatuses);
+
+ // Second commit (log file): update k1 with a LOWER ts (50).
Commit-time ordering keeps the
+ // LATEST WRITE regardless of the ordering value -- the exact mirror
of the event-time k6
+ // case, which discriminates OverwriteWithLatestMerger from event-time
merging.
+ String secondCommit = client.startCommit();
+ List<WriteStatus> secondStatuses = client.upsert(ImmutableList.of(
+ record(schema, "k1", "k1_updated", 11L, 50L)), secondCommit);
+ client.commit(secondCommit, secondStatuses);
+
+ // Third commit: hard delete of k2 (commit-time deletes always win).
+ String deleteCommit = client.startCommit();
+ List<WriteStatus> deleteStatuses = client.delete(
+ ImmutableList.of(hoodieKey("k2")), deleteCommit);
+ client.commit(deleteCommit, deleteStatuses);
+ }
+
+ private static HoodieRecord<HoodieAvroPayload> record(Schema schema,
String key, String name, long value, long ts)
+ {
+ GenericRecord record = new GenericData.Record(schema);
+ record.put(RECORD_KEY_FIELD, key);
+ record.put("name", name);
+ record.put("value", value);
+ record.put(ORDERING_FIELD, ts);
+ return avroRecord(record, key);
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/DmsPayloadHudiTablesInitializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/DmsPayloadHudiTablesInitializer.java
new file mode 100644
index 000000000000..0d1de49d5a3a
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/DmsPayloadHudiTablesInitializer.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.metastore.Column;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.AWSDmsAvroPayload;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+import static
org.apache.hudi.common.model.AWSDmsAvroPayload.DELETE_OPERATION_VALUE;
+import static org.apache.hudi.common.model.AWSDmsAvroPayload.OP_FIELD;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table whose merge semantics come
from the
+ * {@link AWSDmsAvroPayload} class persisted in the table config (issue
apache/hudi#18898). ONLY the payload
+ * class is set (no merge mode / strategy id), so table creation translates it
exactly as a real writer
+ * would: at the current table version this "deprecated" payload becomes
COMMIT_TIME_ORDERING plus PREFIXED
+ * delete-key props ({@code
hoodie.record.merge.property.hoodie.payload.delete.field=Op}, marker {@code D}).
+ * <p>
+ * A base commit is followed by a log record with {@code Op='D'}, which
deletes the row at merge time via
+ * {@code DeleteContext}, with the payload never executing at read, plus a log
record with the non-marker
+ * {@code Op='U'} whose update must APPLY rather than delete -- pinning the
marker-value comparison itself.
+ * <p>
+ * Records are wrapped in {@link HoodieAvroPayload} (a pass-through that is
NOT a {@code BaseAvroPayload}),
+ * so rows a semantic payload would drop at write time land as DATA records
and every merge decision happens
+ * at read time. See {@code TestHudiMorPayloadSemantics}.
+ */
+public class DmsPayloadHudiTablesInitializer
+ extends AbstractMergerHudiTablesInitializer
+{
+ public static final String TABLE_NAME = "dms_mor";
+ public static final String RT_TABLE_NAME = TABLE_NAME + "_rt";
+
+ public DmsPayloadHudiTablesInitializer()
+ {
+ super(TABLE_NAME);
+ }
+
+ @Override
+ protected List<Column> dataColumns()
+ {
+ return ImmutableList.of(
+ new Column(RECORD_KEY_FIELD, HIVE_STRING, Optional.empty(),
Map.of()),
+ new Column("name", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("value", HIVE_LONG, Optional.empty(), Map.of()),
+ // The Avro/parquet field is 'Op' (AWSDms hardcodes that
casing), but a real Hive
+ // metastore lowercases column names on DDL -- exactly the
case mismatch the connector's
+ // merge-column matching must bridge
+ new Column(OP_FIELD.toLowerCase(Locale.ROOT), HIVE_STRING,
Optional.empty(), Map.of()),
+ new Column(ORDERING_FIELD, HIVE_LONG, Optional.empty(),
Map.of()));
+ }
+
+ @Override
+ protected Schema avroSchema()
+ {
+ List<Schema.Field> fields = ImmutableList.of(
+ new Schema.Field(RECORD_KEY_FIELD,
Schema.create(Schema.Type.STRING)),
+ new Schema.Field("name", Schema.create(Schema.Type.STRING)),
+ new Schema.Field("value", Schema.create(Schema.Type.LONG)),
+ new Schema.Field(OP_FIELD, Schema.create(Schema.Type.STRING)),
+ new Schema.Field(ORDERING_FIELD,
Schema.create(Schema.Type.LONG)));
+ return Schema.createRecord(TABLE_NAME, null, null, false, new
ArrayList<>(fields));
+ }
+
+ @Override
+ protected void configureTableConfig(HoodieTableMetaClient.TableBuilder
tableBuilder)
+ {
+ tableBuilder.setPayloadClassName(AWSDmsAvroPayload.class.getName());
+ }
+
+ @Override
+ protected void configureWriteConfig(HoodieWriteConfig.Builder
writeConfigBuilder)
+ {
+ writeConfigBuilder.withWritePayLoad(AWSDmsAvroPayload.class.getName());
+ }
+
+ @Override
+ protected void
writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ {
+ Schema schema = avroSchema();
+ String firstCommit = client.startCommit();
+ List<WriteStatus> firstStatuses = client.bulkInsert(ImmutableList.of(
+ record(schema, "k1", "k1_base", 10L, "I", 100L),
+ record(schema, "k2", "k2_base", 20L, "I", 100L)), firstCommit);
+ client.commit(firstCommit, firstStatuses);
+
+ // Log records, both written as DATA records by the pass-through
HoodieAvroPayload. Only k2's
+ // marker value deletes at merge time via DeleteContext (delete key
Op, marker D from the
+ // translated table config); k1 carries the NON-marker Op='U' and its
update must apply, so a
+ // marker comparison that fires on any non-null Op fails the suite.
+ String secondCommit = client.startCommit();
+ List<WriteStatus> secondStatuses = client.upsert(ImmutableList.of(
+ record(schema, "k1", "k1_updated", 11L, "U", 200L),
+ record(schema, "k2", "k2_deleted", 22L,
DELETE_OPERATION_VALUE, 200L)), secondCommit);
+ client.commit(secondCommit, secondStatuses);
+ }
+
+ private static HoodieRecord<HoodieAvroPayload> record(Schema schema,
String key, String name, long value, String op, long ts)
+ {
+ GenericRecord record = new GenericData.Record(schema);
+ record.put(RECORD_KEY_FIELD, key);
+ record.put("name", name);
+ record.put("value", value);
+ record.put(OP_FIELD, op);
+ record.put(ORDERING_FIELD, ts);
+ return avroRecord(record, key);
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/EventTimeDeletesHudiTablesInitializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/EventTimeDeletesHudiTablesInitializer.java
new file mode 100644
index 000000000000..12cb59371b99
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/EventTimeDeletesHudiTablesInitializer.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.metastore.Column;
+import org.apache.avro.JsonProperties;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.trino.metastore.HiveType.HIVE_BOOLEAN;
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+import static
org.apache.hudi.common.model.HoodieRecord.HOODIE_IS_DELETED_FIELD;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table in {@link
RecordMergeMode#EVENT_TIME_ORDERING} that
+ * exercises the read-side merge-mode dispatch with deletes (issue
apache/hudi#18898). ONLY a record merge
+ * mode is set (no payload class), so table creation persists the mode as-is,
which is exactly the dispatch
+ * input {@code HudiTrinoReaderContext.getRecordMerger} switches on.
+ * <p>
+ * A base commit is followed by a log commit carrying an update, a soft delete
+ * ({@code _hoodie_is_deleted=true}), an OBSOLETE soft delete and an OBSOLETE
update (both with an ordering
+ * value LOWER than the base row's, so event-time merging must keep the base
row), and then a hard-delete
+ * commit ({@code writeClient.delete}) that produces a native delete log file
read back through the
+ * connector's {@code getFileRecordIterator}.
+ * <p>
+ * Records are wrapped in {@link HoodieAvroPayload}, which implements {@code
HoodieRecordPayload} directly
+ * (NOT {@code BaseAvroPayload}), so rows with {@code _hoodie_is_deleted=true}
are written as DATA records
+ * and delete semantics are evaluated at READ time. See {@code
TestHudiMorMergeModeSemantics}.
+ */
+public class EventTimeDeletesHudiTablesInitializer
+ extends AbstractMergerHudiTablesInitializer
+{
+ public static final String TABLE_NAME = "deletes_mor";
+ public static final String RT_TABLE_NAME = TABLE_NAME + "_rt";
+
+ public EventTimeDeletesHudiTablesInitializer()
+ {
+ super(TABLE_NAME);
+ }
+
+ @Override
+ protected List<Column> dataColumns()
+ {
+ return ImmutableList.of(
+ new Column(RECORD_KEY_FIELD, HIVE_STRING, Optional.empty(),
Map.of()),
+ new Column("name", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("value", HIVE_LONG, Optional.empty(), Map.of()),
+ new Column(ORDERING_FIELD, HIVE_LONG, Optional.empty(),
Map.of()),
+ new Column(HOODIE_IS_DELETED_FIELD, HIVE_BOOLEAN,
Optional.empty(), Map.of()));
+ }
+
+ @Override
+ protected Schema avroSchema()
+ {
+ List<Schema.Field> fields = ImmutableList.of(
+ new Schema.Field(RECORD_KEY_FIELD,
Schema.create(Schema.Type.STRING)),
+ new Schema.Field("name", Schema.create(Schema.Type.STRING)),
+ new Schema.Field("value", Schema.create(Schema.Type.LONG)),
+ new Schema.Field(ORDERING_FIELD,
Schema.create(Schema.Type.LONG)),
+ new Schema.Field(
+ HOODIE_IS_DELETED_FIELD,
+ Schema.createUnion(Schema.create(Schema.Type.NULL),
Schema.create(Schema.Type.BOOLEAN)),
+ null,
+ JsonProperties.NULL_VALUE));
+ return Schema.createRecord(TABLE_NAME, null, null, false, new
ArrayList<>(fields));
+ }
+
+ @Override
+ protected void configureTableConfig(HoodieTableMetaClient.TableBuilder
tableBuilder)
+ {
+ tableBuilder.setRecordMergeMode(RecordMergeMode.EVENT_TIME_ORDERING);
+ }
+
+ @Override
+ protected void configureWriteConfig(HoodieWriteConfig.Builder
writeConfigBuilder)
+ {
+
writeConfigBuilder.withRecordMergeMode(RecordMergeMode.EVENT_TIME_ORDERING);
+ }
+
+ @Override
+ protected void
writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ {
+ Schema schema = avroSchema();
+ // First commit: base parquet file with 6 keys, all at ordering value
(ts) 100.
+ String firstCommit = client.startCommit();
+ List<WriteStatus> firstStatuses = client.bulkInsert(ImmutableList.of(
+ record(schema, "k1", "k1_base", 10L, 100L, false),
+ record(schema, "k2", "k2_base", 20L, 100L, false),
+ record(schema, "k3", "k3_base", 30L, 100L, false),
+ record(schema, "k4", "k4_base", 40L, 100L, false),
+ record(schema, "k5", "k5_base", 50L, 100L, false),
+ record(schema, "k6", "k6_base", 60L, 100L, false)),
firstCommit);
+ client.commit(firstCommit, firstStatuses);
+
+ // Second commit (log file). Event-time merging must resolve each key
by ordering value:
+ // - k1: update with HIGHER ts (200) -> update wins
+ // - k3: soft delete with HIGHER ts (200) -> row deleted at read time
+ // - k4: soft delete with LOWER ts (50) -> OBSOLETE delete, base row
survives
+ // - k6: update with LOWER ts (50) -> OBSOLETE update, base row
survives
+ String secondCommit = client.startCommit();
+ List<WriteStatus> secondStatuses = client.upsert(ImmutableList.of(
+ record(schema, "k1", "k1_updated", 11L, 200L, false),
+ record(schema, "k3", "k3_deleted", 33L, 200L, true),
+ record(schema, "k4", "k4_deleted", 44L, 50L, true),
+ record(schema, "k6", "k6_updated", 66L, 50L, false)),
secondCommit);
+ client.commit(secondCommit, secondStatuses);
+
+ // Third commit: HARD delete of k2. At the current table version this
produces a native
+ // delete log file, which the file-group reader reads back through the
connector's
+ // getFileRecordIterator with the synthetic delete-log schema (record
key + ordering).
+ // Hard deletes carry the sentinel ordering value and win regardless
of merge mode.
+ String deleteCommit = client.startCommit();
+ List<WriteStatus> deleteStatuses = client.delete(
+ ImmutableList.of(hoodieKey("k2")), deleteCommit);
+ client.commit(deleteCommit, deleteStatuses);
+ }
+
+ private static HoodieRecord<HoodieAvroPayload> record(Schema schema,
String key, String name, long value, long ts, boolean deleted)
+ {
+ GenericRecord record = new GenericData.Record(schema);
+ record.put(RECORD_KEY_FIELD, key);
+ record.put("name", name);
+ record.put("value", value);
+ record.put(ORDERING_FIELD, ts);
+ record.put(HOODIE_IS_DELETED_FIELD, deleted);
+ // HoodieAvroPayload passes the record through untouched (it is not a
BaseAvroPayload), so a row
+ // with _hoodie_is_deleted=true is WRITTEN as a data record and only
deleted at merge/read time.
+ return avroRecord(record, key);
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/OverwriteNonDefaultsPayloadHudiTablesInitializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/OverwriteNonDefaultsPayloadHudiTablesInitializer.java
new file mode 100644
index 000000000000..ba460111d5d3
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/OverwriteNonDefaultsPayloadHudiTablesInitializer.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.metastore.Column;
+import org.apache.avro.JsonProperties;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.OverwriteNonDefaultsWithLatestAvroPayload;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table whose merge semantics come
from the
+ * {@link OverwriteNonDefaultsWithLatestAvroPayload} class persisted in the
table config (issue
+ * apache/hudi#18898). ONLY the payload class is set (no merge mode / strategy
id), so table creation
+ * translates it exactly as a real writer would: into COMMIT_TIME_ORDERING plus
+ * {@code PARTIAL_UPDATE_MODE=IGNORE_DEFAULTS}.
+ * <p>
+ * A base commit is followed by an update whose column is null (the schema
default), which must keep the
+ * STORED value for that column at merge time.
+ * <p>
+ * Records are wrapped in {@link HoodieAvroPayload} (a pass-through that is
NOT a {@code BaseAvroPayload}),
+ * so every merge decision happens at read time from the table config. See
+ * {@code TestHudiMorPayloadSemantics}.
+ */
+public class OverwriteNonDefaultsPayloadHudiTablesInitializer
+ extends AbstractMergerHudiTablesInitializer
+{
+ public static final String TABLE_NAME = "overwrite_non_defaults_mor";
+ public static final String RT_TABLE_NAME = TABLE_NAME + "_rt";
+
+ public OverwriteNonDefaultsPayloadHudiTablesInitializer()
+ {
+ super(TABLE_NAME);
+ }
+
+ @Override
+ protected List<Column> dataColumns()
+ {
+ return ImmutableList.of(
+ new Column(RECORD_KEY_FIELD, HIVE_STRING, Optional.empty(),
Map.of()),
+ new Column("a", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("b", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column(ORDERING_FIELD, HIVE_LONG, Optional.empty(),
Map.of()));
+ }
+
+ @Override
+ protected Schema avroSchema()
+ {
+ Schema nullableString =
Schema.createUnion(Schema.create(Schema.Type.NULL),
Schema.create(Schema.Type.STRING));
+ List<Schema.Field> fields = ImmutableList.of(
+ new Schema.Field(RECORD_KEY_FIELD,
Schema.create(Schema.Type.STRING)),
+ new Schema.Field("a", nullableString, null,
JsonProperties.NULL_VALUE),
+ new Schema.Field("b", nullableString, null,
JsonProperties.NULL_VALUE),
+ new Schema.Field(ORDERING_FIELD,
Schema.create(Schema.Type.LONG)));
+ return Schema.createRecord(TABLE_NAME, null, null, false, new
ArrayList<>(fields));
+ }
+
+ @Override
+ protected void configureTableConfig(HoodieTableMetaClient.TableBuilder
tableBuilder)
+ {
+
tableBuilder.setPayloadClassName(OverwriteNonDefaultsWithLatestAvroPayload.class.getName());
+ }
+
+ @Override
+ protected void configureWriteConfig(HoodieWriteConfig.Builder
writeConfigBuilder)
+ {
+
writeConfigBuilder.withWritePayLoad(OverwriteNonDefaultsWithLatestAvroPayload.class.getName());
+ }
+
+ @Override
+ protected void
writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ {
+ Schema schema = avroSchema();
+ String firstCommit = client.startCommit();
+ List<WriteStatus> firstStatuses = client.bulkInsert(ImmutableList.of(
+ record(schema, "k1", "base_a", "base_b", 100L)), firstCommit);
+ client.commit(firstCommit, firstStatuses);
+
+ // Update with b=null (the schema default): IGNORE_DEFAULTS partial
merging must keep the
+ // stored 'base_b' while taking the updated 'new_a'.
+ String secondCommit = client.startCommit();
+ List<WriteStatus> secondStatuses = client.upsert(ImmutableList.of(
+ record(schema, "k1", "new_a", null, 200L)), secondCommit);
+ client.commit(secondCommit, secondStatuses);
+ }
+
+ private static HoodieRecord<HoodieAvroPayload> record(Schema schema,
String key, String a, String b, long ts)
+ {
+ GenericRecord record = new GenericData.Record(schema);
+ record.put(RECORD_KEY_FIELD, key);
+ record.put("a", a);
+ record.put("b", b);
+ record.put(ORDERING_FIELD, ts);
+ return avroRecord(record, key);
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SummingPayloadHudiTablesInitializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SummingPayloadHudiTablesInitializer.java
new file mode 100644
index 000000000000..07d437decb13
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SummingPayloadHudiTablesInitializer.java
@@ -0,0 +1,131 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.metastore.Column;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table whose merge semantics come
from the {@link SummingTestPayload}
+ * class persisted in the table config (issue apache/hudi#18898). ONLY the
payload class is set (no merge mode
+ * / strategy id), so table creation translates it exactly as a real writer
would: this user-defined payload is
+ * NOT in the deprecation set, so it is persisted as RECORD_MERGE_MODE=CUSTOM
with the payload-based merge
+ * strategy id. Reads resolve {@code HoodieAvroRecordMerger} (no {@code
hudi.record-merger-impls} needed) and
+ * run the payload's {@code combineAndGetUpdateValue}, observable as SUMMED
values.
+ * <p>
+ * A final commit hard-deletes a key ({@code writeClient.delete}): the native
delete log record routes to
+ * {@code HoodieAvroRecordMerger} but wins on its {@code
isCommitTimeOrderingDelete} short-circuit (the
+ * delete carries the sentinel ordering value), before any payload is
constructed -- the delete coverage
+ * both ordering arms already have.
+ * <p>
+ * Records are wrapped in {@link HoodieAvroPayload} (a pass-through that is
NOT a {@code BaseAvroPayload}), so
+ * every merge decision happens at read time from the table config. See {@code
TestHudiMorPayloadSemantics}.
+ */
+public class SummingPayloadHudiTablesInitializer
+ extends AbstractMergerHudiTablesInitializer
+{
+ public static final String TABLE_NAME = "summing_mor";
+ public static final String RT_TABLE_NAME = TABLE_NAME + "_rt";
+
+ private static final String SUM_FIELD = SummingTestPayload.SUM_COLUMN;
+
+ public SummingPayloadHudiTablesInitializer()
+ {
+ super(TABLE_NAME);
+ }
+
+ @Override
+ protected List<Column> dataColumns()
+ {
+ return ImmutableList.of(
+ new Column(RECORD_KEY_FIELD, HIVE_STRING, Optional.empty(),
Map.of()),
+ new Column(SUM_FIELD, HIVE_LONG, Optional.empty(), Map.of()),
+ new Column(ORDERING_FIELD, HIVE_LONG, Optional.empty(),
Map.of()));
+ }
+
+ @Override
+ protected Schema avroSchema()
+ {
+ List<Schema.Field> fields = ImmutableList.of(
+ new Schema.Field(RECORD_KEY_FIELD,
Schema.create(Schema.Type.STRING)),
+ new Schema.Field(SUM_FIELD, Schema.create(Schema.Type.LONG)),
+ new Schema.Field(ORDERING_FIELD,
Schema.create(Schema.Type.LONG)));
+ return Schema.createRecord(TABLE_NAME, null, null, false, new
ArrayList<>(fields));
+ }
+
+ @Override
+ protected void configureTableConfig(HoodieTableMetaClient.TableBuilder
tableBuilder)
+ {
+ tableBuilder.setPayloadClassName(SummingTestPayload.class.getName());
+ }
+
+ @Override
+ protected void configureWriteConfig(HoodieWriteConfig.Builder
writeConfigBuilder)
+ {
+
writeConfigBuilder.withWritePayLoad(SummingTestPayload.class.getName());
+ }
+
+ @Override
+ protected void
writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ {
+ Schema schema = avroSchema();
+ String firstCommit = client.startCommit();
+ List<WriteStatus> firstStatuses = client.bulkInsert(ImmutableList.of(
+ record(schema, "k1", 10L, 100L),
+ record(schema, "k2", 20L, 100L)), firstCommit);
+ client.commit(firstCommit, firstStatuses);
+
+ // The payload's combineAndGetUpdateValue SUMS stored and incoming
values: 10 + 99 = 109 --
+ // a result neither overwrite (99) nor base-only (10) can produce.
+ String secondCommit = client.startCommit();
+ List<WriteStatus> secondStatuses = client.upsert(ImmutableList.of(
+ record(schema, "k1", 99L, 200L)), secondCommit);
+ client.commit(secondCommit, secondStatuses);
+
+ // Third commit: hard delete of k2. The native delete log record
reaches the payload-based
+ // CUSTOM merge arm, where it wins on HoodieAvroRecordMerger's
isCommitTimeOrderingDelete
+ // short-circuit (writeClient.delete records carry the sentinel
ordering value), before any
+ // payload is constructed -- the delete path of the user-merger
dispatch.
+ String deleteCommit = client.startCommit();
+ List<WriteStatus> deleteStatuses = client.delete(
+ ImmutableList.of(hoodieKey("k2")), deleteCommit);
+ client.commit(deleteCommit, deleteStatuses);
+ }
+
+ private static HoodieRecord<HoodieAvroPayload> record(Schema schema,
String key, long value, long ts)
+ {
+ GenericRecord record = new GenericData.Record(schema);
+ record.put(RECORD_KEY_FIELD, key);
+ record.put(SUM_FIELD, value);
+ record.put(ORDERING_FIELD, ts);
+ return avroRecord(record, key);
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SummingTestPayload.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SummingTestPayload.java
new file mode 100644
index 000000000000..7a009fbeab22
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SummingTestPayload.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
+import org.apache.hudi.common.util.Option;
+
+import java.io.IOException;
+
+/**
+ * Test-only user-defined {@link
org.apache.hudi.common.model.HoodieRecordPayload} whose read-side merge
+ * SUMS the {@code value} column of the stored and incoming records. A merged
row therefore carries a
+ * value no built-in merge policy can produce (overwrite yields the incoming
value, base-only the stored
+ * value), which proves end-to-end that the connector's CUSTOM merge branch
executed this payload's
+ * {@code combineAndGetUpdateValue} (issue apache/hudi#18898).
+ * <p>
+ * Unlike the built-in payloads named in the issue, this class is NOT in
hudi's payloads-under-deprecation
+ * set, so v9+ table creation persists it as {@code RECORD_MERGE_MODE=CUSTOM}
with the payload-based merge
+ * strategy id -- the configuration that routes reads through {@code
HoodieAvroRecordMerger} and this
+ * payload, with no {@code hudi.record-merger-impls} connector property
involved.
+ */
+public class SummingTestPayload
+ extends OverwriteWithLatestAvroPayload
+{
+ /** Name of the column whose stored and incoming values are summed at
merge time. */
+ public static final String SUM_COLUMN = "value";
+
+ public SummingTestPayload(GenericRecord record, Comparable orderingVal)
+ {
+ super(record, orderingVal);
+ }
+
+ public SummingTestPayload(Option<GenericRecord> record)
+ {
+ super(record);
+ }
+
+ @Override
+ public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord
currentValue, Schema schema)
+ throws IOException
+ {
+ Option<IndexedRecord> incoming = getInsertValue(schema);
+ if (incoming.isEmpty()) {
+ return Option.empty();
+ }
+ GenericRecord newer = (GenericRecord) incoming.get();
+ GenericRecord older = (GenericRecord) currentValue;
+
+ long sum = ((Number) older.get(SUM_COLUMN)).longValue() + ((Number)
newer.get(SUM_COLUMN)).longValue();
+ GenericRecord merged = new GenericData.Record(newer.getSchema());
+ for (Schema.Field field : newer.getSchema().getFields()) {
+ merged.put(field.pos(), newer.get(field.pos()));
+ }
+ merged.put(SUM_COLUMN, sum);
+ return Option.of(merged);
+ }
+}
