wombatu-kun commented on code in PR #19295:
URL: https://github.com/apache/hudi/pull/19295#discussion_r3680358889
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMergeRequiredColumns.java:
##########
@@ -89,6 +89,21 @@ public void testCustomDeleteKeyRequiresMarkerToo()
.doesNotContain("op");
}
+ @Test
+ public void testPrefixedDeleteKeyAndMarkerAreRequested()
Review Comment:
`mergeRequiredColumnNames` only ever adds the delete key, so the marker is a
precondition rather than a requested column and the assertion checks just `Op`.
Rename to something like `testPrefixedDeleteKeyIsRequested` to match what it
pins.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/DmsPayloadHudiTablesInitializer.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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;
+
+/**
+ * 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.
+ * <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 = "mor_dms";
+ public static final String RT_TABLE_NAME = TABLE_NAME + "_rt";
+
+ private static final String OP_FIELD = "Op";
Review Comment:
`AWSDmsAvroPayload` is already imported here and exposes `OP_FIELD` and
`DELETE_OPERATION_VALUE`, the same two values the table-config translation this
fixture exercises writes into the merge properties. Reuse them instead of the
local `Op` and `D` literals so the fixture cannot drift from the payload it
pins.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SummingPayloadHudiTablesInitializer.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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>
+ * 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 = "mor_summing";
+ 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)), 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);
Review Comment:
No fixture on the CUSTOM merge arm writes a delete, so the path that routes
a delete record through the user merger stays uncovered while both ordering
arms have it. Is a hard delete on this table in scope, or a follow-up?
##########
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 = "mor_deletes";
Review Comment:
The five new table names lead with `mor_` while every existing MoR fixture
in this package trails it (`custom_merger_mor`, `payload_only_mor`,
`omitted_ordering_field_mor`). Worth flipping to `deletes_mor` and friends for
consistency; nit, not a blocker.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/DmsPayloadHudiTablesInitializer.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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;
+
+/**
+ * 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.
+ * <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 = "mor_dms";
+ public static final String RT_TABLE_NAME = TABLE_NAME + "_rt";
+
+ private static final String OP_FIELD = "Op";
+
+ 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 record with Op='D'. The pass-through HoodieAvroPayload writes
it as a DATA record;
+ // DeleteContext (delete key Op, marker D from the translated table
config) deletes the row
+ // at merge time.
+ String secondCommit = client.startCommit();
+ List<WriteStatus> secondStatuses = client.upsert(ImmutableList.of(
+ record(schema, "k2", "k2_deleted", 22L, "D", 200L)),
secondCommit);
Review Comment:
The second commit's only log record is the `Op='D'` delete, so `k1` survives
by having no log record at all and nothing pins the marker value itself. Could
the commit also carry a non-marker `Op` record for an existing key, so a broken
marker comparison fails the suite?
--
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]
