wombatu-kun commented on code in PR #19295:
URL: https://github.com/apache/hudi/pull/19295#discussion_r3672912333
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -272,6 +272,12 @@ public static HoodieTableMetaClient buildTableMetaClient(
public static Schema constructSchema(List<String> columnNames,
List<HiveType> columnTypes)
{
+ // A count(*)-style query projects no columns at all;
determineSchemaOrThrowException rejects
+ // an empty column list, so build the empty record schema directly.
+ if (columnNames.isEmpty()) {
+ return SchemaBuilder.record("baseRecord").fields().endRecord();
Review Comment:
The Summary and Changelog names three fixes but not this empty-projection
change or the `requiredSchema`-based `HudiAvroSerializer` rework, which are the
two behaviour changes here a reader would most want called out. `No API or
config changes` also does not hold for a new public constructor plus a changed
contract on a shared static util - can the description cover both?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -715,14 +726,19 @@ public static List<HiveColumnHandle>
appendMissingMergeRequiredColumns(
/**
* 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-insensitive: predicted merge columns carry the
+ * table schema's field casing (e.g. the {@code Op} delete key of DMS
tables) while metastore column names are
+ * lowercased.
*/
private static List<HiveColumnHandle> buildColumnHandles(Table table,
TypeManager typeManager, Set<String> columnNames, HiveTimestampPrecision
timestampPrecision)
{
+ Set<String> lowercaseColumnNames = columnNames.stream()
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toSet());
ImmutableList.Builder<HiveColumnHandle> columns =
ImmutableList.builder();
int hiveColumnIndex = 0;
for (Column field : table.getDataColumns()) {
- if (columnNames.contains(field.getName())) {
+ if
(lowercaseColumnNames.contains(field.getName().toLowerCase(Locale.ROOT))) {
Review Comment:
Once the delete key resolves from the prefixed merge properties,
`appendMissingMergeRequiredColumns` already recovers a merge-required column
the metastore match missed, typed from the table schema and matched
case-insensitively, so reverting this line changes no read outcome and fails no
test. Is the case-folding still needed here, or should it come with a
`getMergeRequiredColumnHandles` test on a mixed-case metastore column?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java:
##########
@@ -112,23 +112,66 @@ public class HudiAvroSerializer
private final List<HiveColumnHandle> columnHandles;
private final List<Type> columnTypes;
private final Schema schema;
+ // serialize() writes page channel i at record position
channelToFieldPosition[i]. -1 marks a
+ // hidden column: a Trino synthesized metadata column ($path, $file_size,
$file_modified_time,
+ // $partition) with no counterpart in the data file or table schema -- its
value is prefilled
+ // per split from PrefilledColumnValues -- so it has no record field and
its channel is skipped.
+ 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
+ // The record schema carries only columns that exist in the data:
hidden (synthesized
+ // metadata) columns are answered from the split by
PrefilledColumnValues, not read from
+ // the file, and cannot be described as Avro fields
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;
+ // With hidden columns excluded from the schema, the remaining
channels map to sequential
+ // field positions; hidden channels map to -1
+ int[] mapping = new int[columnHandles.size()];
Review Comment:
Nothing reads this constructor's `schema` or `channelToFieldPosition` at
head: `serialize` is reached only from `createRecordIterator`, whose
serializers both come from the three-arg constructor, and `HudiPageSource`
calls only `buildRecordInPage`. Is the empty-projection special case in
`constructSchema` still needed, or would dropping the unused schema here fix
the `count(*)` failure at its source?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -651,8 +657,13 @@ static LinkedHashSet<String>
mergeRequiredColumnNames(HoodieTableConfig tableCon
// 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 -- 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();
Review Comment:
`HoodieTableConfig` declares only `getTableMergeProperties(String
payloadClass)`, so this no-arg call does not compile against master. Pass
`tableConfig.getPayloadClass()` - the same value `ConfigUtils.getMergeProps`
resolves here, since `buildReaderProperties` seeds the reader props from the
table config itself.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/MergeModeSemanticsHudiTablesInitializer.java:
##########
@@ -0,0 +1,337 @@
+/*
+ * 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 com.google.common.collect.ImmutableMap;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.metastore.Column;
+import io.trino.metastore.HiveMetastore;
+import io.trino.metastore.HiveMetastoreFactory;
+import io.trino.metastore.PrincipalPrivileges;
+import io.trino.metastore.StorageFormat;
+import io.trino.metastore.Table;
+import io.trino.plugin.hudi.HudiConnector;
+import io.trino.spi.security.ConnectorIdentity;
+import io.trino.testing.QueryRunner;
+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.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.common.HoodieJavaEngineContext;
+import org.apache.hudi.common.bootstrap.index.NoOpBootstrapIndex;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.marker.MarkerType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static com.google.common.io.MoreFiles.deleteRecursively;
+import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
+import static io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_INPUT_FORMAT;
+import static
io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_REALTIME_INPUT_FORMAT;
+import static
io.trino.hive.formats.HiveClassNames.MAPRED_PARQUET_OUTPUT_FORMAT_CLASS;
+import static io.trino.hive.formats.HiveClassNames.PARQUET_HIVE_SERDE_CLASS;
+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 io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
+import static java.nio.file.Files.createTempDirectory;
+import static
org.apache.hudi.common.model.HoodieRecord.HOODIE_IS_DELETED_FIELD;
+
+/**
+ * Creates two non-partitioned Merge-On-Read tables at test runtime that
exercise the read-side
+ * MERGE-MODE dispatch with deletes (issue apache/hudi#18898). Both tables set
ONLY a record merge mode
+ * (no payload class) so table creation persists the mode as-is:
+ * <ul>
+ * <li>{@code mor_deletes} ({@link RecordMergeMode#EVENT_TIME_ORDERING}): a
base commit followed by a
+ * log commit with 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), plus a hard-delete commit
+ * ({@code writeClient.delete}) that produces a native delete log file
read back through the
+ * connector's {@code getFileRecordIterator}.</li>
+ * <li>{@code mor_commit_time} ({@link
RecordMergeMode#COMMIT_TIME_ORDERING}): the same
+ * lower-ordering update shape, where latest-write-wins must KEEP the
update -- the exact mirror
+ * of the event-time case, discriminating the two merger dispatches.</li>
+ * </ul>
+ * 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. Inline
compaction is disabled so log
+ * files survive. Each table registers a read-optimized and a {@code _rt}
metastore table.
+ */
+public class MergeModeSemanticsHudiTablesInitializer
+ implements HudiTablesInitializer
Review Comment:
Both new fixtures implement `HudiTablesInitializer` directly and re-derive
`createTableDefinition`, the write-client setup, the meta-column list and the
`_rt` registration, while the other six MoR merge fixtures get all of that from
`AbstractMergerHudiTablesInitializer` and vary only through its
`configureTableConfig` and `configureWriteConfig` hooks. Can these extend it as
well, composing the multi-table suites with `CompositeHudiTablesInitializer`
the way `TestHudiNonProjectionCompatibleMerger` does?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/reader/HudiTrinoReaderContext.java:
##########
@@ -176,9 +184,25 @@ private List<HiveColumnHandle>
buildRequiredColumnHandles(HoodieSchema requiredS
{
List<HiveColumnHandle> handles = new ArrayList<>();
for (HoodieSchemaField field : requiredSchema.getFields()) {
- handles.add(colNameToHandle.computeIfAbsent(
+ HiveColumnHandle handle = colNameToHandle.computeIfAbsent(
field.name().toLowerCase(Locale.ROOT),
- _ -> HudiUtil.toColumnHandle(field)));
+ _ -> HudiUtil.toColumnHandle(field));
+ if (!handle.getName().equals(field.name())) {
Review Comment:
With the log serializer now stamping `requiredSchema` onto every record,
hudi-common resolves the merge field off the schema's own field rather than the
handle name, and the log parquet source resolves columns case-insensitively
either way. Is the re-label still doing work, or is it now covered by the
serializer change?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/reader/HudiTrinoReaderContext.java:
##########
@@ -243,9 +267,8 @@ protected Option<HoodieRecordMerger>
getRecordMerger(RecordMergeMode mergeMode,
// 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.
+ // and custom-payload tables correct on MoR reads;
TestHudiMorMergeModeSemantics and
+ // TestHudiMorPayloadSemantics cover these branches end-to-end
(apache/hudi#18898).
Review Comment:
`BufferedRecordMergerFactory` picks `CommitTimeRecordMerger` and
`EventTimeRecordMerger` from the merge mode alone and ignores the merger passed
to it, and `FileGroupReaderSchemaHandler` dereferences that merger only under
CUSTOM - so `TestHudiMorMergeModeSemantics` pins the merge modes, not the two
non-CUSTOM arms of this method. Should the comment claim coverage only for the
CUSTOM branch and leave the TODO open for the other two?
--
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]
