wombatu-kun commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3658583976
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -348,6 +365,115 @@ public static Schema.Field getFieldFromSchema(String
columnName, Schema schema)
"Failed to get column " + columnName + " from table schema");
}
+ /**
+ * Builds a {@link HiveColumnHandle} for a table-schema field, typing it
from the field's Avro schema.
+ * Used to resolve columns the file-group reader requires for merging but
the connector projection does
+ * not carry (e.g. {@code _hoodie_commit_time} or a delete-marker column
on a narrow query).
+ * <p>
+ * The handle's {@code hiveColumnIndex} is a placeholder: page sources
built from these handles resolve
+ * parquet columns by NAME (directly when {@code
hudi.parquet.use-column-names=true}, otherwise
+ * {@link HudiPageSourceProvider#remapColumnIndicesToPhysical} rebuilds
every index from the file
+ * schema by name), so the ordinal is never used to locate data.
+ * <p>
+ * Avro types whose Trino mapping has no Hive counterpart (uuid,
time-millis/micros) fail with
+ * NOT_SUPPORTED from {@link HiveTypeTranslator#toHiveType}; such columns
are equally unreadable
+ * through the metastore schema, so this surfaces the same limitation with
a clear error.
+ */
+ public static HiveColumnHandle toColumnHandle(HoodieSchemaField field)
+ {
+ Type trinoType;
+ try {
+ trinoType =
AVRO_TYPE_HANDLER.typeFor(field.schema().toAvroSchema());
+ }
+ catch (AvroTypeException e) {
+ throw new TrinoException(HUDI_SCHEMA_ERROR,
+ "Failed to map Avro type of column " + field.name() + " to
a Trino type", e);
+ }
+ return new HiveColumnHandle(
+ field.name(),
+ 0,
+ HiveTypeTranslator.toHiveType(trinoType),
+ trinoType,
+ Optional.empty(),
+ HiveColumnHandle.ColumnType.REGULAR,
+ Optional.empty());
+ }
+
+ /**
+ * Resolves the merge mode and merge strategy id the file-group reader
will use for this table,
+ * applying hudi-common's version-gated inference: the merge MODE is
inferred for any table below
+ * version 9 ({@code
FileGroupReaderSchemaHandler.generateRequiredSchema}), while the STRATEGY ID
+ * the merger is resolved with is inferred only below version 8
+ * ({@code HoodieReaderContext.initRecordMerger}). The asymmetry is
deliberate and must mirror
+ * hudi-common exactly: pre-v9 tables may persist neither config (a 0.x
table with a custom payload
+ * class infers CUSTOM mode with the payload-based strategy id), so a
connector-side decision built
+ * on the raw configs would disagree with the file-group reader's.
+ */
+ public static Pair<RecordMergeMode, String>
resolveMergeModeAndStrategyId(HoodieTableConfig tableConfig)
+ {
+ RecordMergeMode mergeMode = tableConfig.getRecordMergeMode();
+ String mergeStrategyId = tableConfig.getRecordMergeStrategyId();
+ if (tableConfig.getTableVersion().lesserThan(HoodieTableVersion.NINE))
{
+ Triple<RecordMergeMode, String, String> inferred =
HoodieTableConfig.inferMergingConfigsForPreV9Table(
+ tableConfig.getRecordMergeMode(),
+ tableConfig.getPayloadClass(),
+ tableConfig.getRecordMergeStrategyId(),
+ tableConfig.getOrderingFieldsStr().orElse(null),
+ tableConfig.getTableVersion());
+ mergeMode = inferred.getLeft();
+ if
(tableConfig.getTableVersion().lesserThan(HoodieTableVersion.EIGHT)) {
+ mergeStrategyId = inferred.getRight();
+ }
+ }
+ return Pair.of(mergeMode, mergeStrategyId);
+ }
+
+ /**
+ * Returns whether reads of this table go through a CUSTOM record merger
that is NOT projection
+ * compatible. For such mergers the file-group reader demands the FULL
table schema as its required
+ * schema on any split with log files ({@code
FileGroupReaderSchemaHandler.generateRequiredSchema}),
+ * so the connector must read every table column, not just the query
projection.
+ * <p>
+ * Callers must pass the merge mode and strategy id resolved by {@link
#resolveMergeModeAndStrategyId}
+ * so the {@link HoodieRecordUtils#createValidRecordMerger} call here sees
the same inputs as the
+ * file-group reader's own resolution. A CUSTOM mode without a strategy id
returns false: no merger
+ * can be resolved from it, and the file-group reader itself then fails
loudly.
Review Comment:
For a null strategy id the file-group reader does not fail loudly -
`createValidRecordMerger` dereferences the id on its first line and NPEs, which
is the v8-plus-custom-payload case the new unit test pins. Either drop the
"fails loudly" half or raise the actionable error here.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -425,10 +488,14 @@ public static List<HiveColumnHandle>
remapColumnIndicesToPhysical(
String requestedName = originalHandle.getBaseColumnName();
// Determine the key to use for looking up the physical index
- String lookupKey = caseSensitive ? requestedName :
requestedName.toLowerCase(Locale.getDefault());
+ String lookupKey = caseSensitive ? requestedName :
requestedName.toLowerCase(Locale.ROOT);
- // Find the physical index from the file schema map constructed
from fielSchema
+ // Find the physical index from the file schema map constructed
from fileSchema
Integer physicalIndex = physicalIndexMap.get(lookupKey);
+ if (physicalIndex == null) {
+ throw new TrinoException(HUDI_SCHEMA_ERROR, format(
+ "Column '%s' not found in parquet file schema %s",
requestedName, fileSchema.getName()));
Review Comment:
With the projection now widened to the full table schema, this throw makes
`hudi.parquet.use-column-names=false` fail on any base file written before a
column was added, while the default name-based path null-fills it in
`createParquetPageSource`. Mapping a not-found column to `fileFields.size()`
would fall into the parquet reader's own absent-column handling instead - or is
the hard failure deliberate here?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -260,17 +281,59 @@ public ConnectorPageSource createPageSource(
.withDataSchema(dataSchema)
.withRequestedSchema(HoodieSchema.fromAvroSchema(requestedSchema))
.withLatestCommitTime(hudiTableHandle.getLatestCommitTime())
- .withProps(buildReaderProperties(session, metaClient))
+ .withProps(readerProps)
.withShouldUseRecordPosition(false)
.withStart(start)
.withLength(length)
.build();
return new HudiPageSource(
dataPageSource,
fileGroupReader,
- readerContext,
hiveColumnHandles,
- synthesizedColumnHandler);
+ prefilledColumnValues);
+ }
+
+ private ConnectorPageSource createBaseFilePageSource(
+ ConnectorSession session,
+ List<HiveColumnHandle> columns,
+ HudiSplit hudiSplit,
+ TrinoFileSystem fileSystem,
+ ParquetReaderOptions sessionOptions,
+ long start,
+ long length,
+ DynamicFilter dynamicFilter,
+ boolean enablePredicatePushDown)
+ {
+ HudiBaseFile baseFile = hudiSplit.getBaseFile().orElseThrow();
+ return createPageSource(
+ session,
+ columns,
+ hudiSplit,
+ fileSystem.newInputFile(Location.of(baseFile.getPath()),
baseFile.getFileSize()),
+ baseFile.getPath(),
+ start,
+ length,
+ OptionalLong.of(baseFile.getFileSize()),
+ dataSourceStats,
+ sessionOptions,
+ timeZone,
+ dynamicFilter,
+ enablePredicatePushDown);
+ }
+
+ /**
+ * Mirrors {@code FileGroupReaderSchemaHandler.generateRequiredSchema}'s
decision for CUSTOM merge
+ * mode: resolves the merge mode and strategy id with the version-gated
inference the file-group
+ * reader applies ({@link HudiUtil#resolveMergeModeAndStrategyId}) and
asks the same resolved merger
+ * whether it is projection compatible, so the connector's read projection
and the file-group
+ * reader's required schema cannot disagree.
Review Comment:
This says the two "cannot disagree", but only the CUSTOM and
`isProjectionCompatible` half is mirrored exactly - the mandatory-fields half
is a superset, and the guard this PR adds to `getFileRecordIterator` exists for
exactly that drift. Worth softening to what the method actually guarantees.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiNonProjectionCompatibleMerger.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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 com.google.common.collect.ImmutableMap;
+import
io.trino.plugin.hudi.testing.NonProjectionCompatibleMergerHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.NonProjectionCompatibleRankMerger;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.QueryRunner;
+import org.junit.jupiter.api.Test;
+
+import static
io.trino.plugin.hudi.testing.NonProjectionCompatibleMergerHudiTablesInitializer.RT_TABLE_NAME;
+import static
io.trino.plugin.hudi.testing.NonProjectionCompatibleMergerHudiTablesInitializer.TABLE_NAME;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Acceptance test for full-table-schema merge reads (apache/hudi#19249, issue
comment on scope):
+ * {@link NonProjectionCompatibleRankMerger} does NOT override {@code
isProjectionCompatible()} (default
+ * {@code false}) and does NOT declare {@code merge_rank} mandatory, so the
file-group reader demands the
+ * FULL table schema as its required schema for base and log reads alike, and
nothing prepends
+ * {@code merge_rank} into the connector's read projection.
+ * <p>
+ * The queries below never project {@code merge_rank}. The data is laid out so
each merge direction is
+ * proven independently: {@code k1}'s winning rank is on the LOG record
(update wins, value 99) and
+ * {@code k2}'s winning rank is on the BASE record (base wins, value 100). A
correct result therefore
+ * requires the un-projected rank column to be read on BOTH sides of the
merge. {@code sum(value)} is a
+ * three-way discriminator: merged = 199, base-only = 110, built-in
newest-wins = 103.
+ */
+public class TestHudiNonProjectionCompatibleMerger
Review Comment:
Every pre-v9 payload-class table now takes the full-table-schema read path,
but nothing exercises it end to end - the v6 and v8 MOR fixtures all infer
COMMIT_TIME or EVENT_TIME, and both CUSTOM initializers set an explicit
strategy id and merger impl. Worth one table configured with only a payload
class, so the payload-based strategy actually gets read through.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/NonProjectionCompatibleMergerHudiTablesInitializer.java:
##########
@@ -0,0 +1,242 @@
+/*
+ * 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.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_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;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table at test runtime configured
with the
+ * NON-projection-compatible {@link NonProjectionCompatibleRankMerger} via
{@link RecordMergeMode#CUSTOM}.
+ * One {@code bulkInsert} (base files) is followed by one {@code upsert} of
the same keys (log files), with
+ * inline compaction disabled so the merger runs at read time over a
full-table-schema read.
+ * <p>
+ * Data is laid out so each merge direction is distinguishable: one key's
winning rank comes from the LOG
+ * record and the other's from the BASE record, so a correct result proves
both sides of the merge supplied
+ * the (never projected) {@code merge_rank} column. See {@code
TestHudiNonProjectionCompatibleMerger}.
+ */
+public class NonProjectionCompatibleMergerHudiTablesInitializer
Review Comment:
This is the third near-identical MoR initializer - `initializeTables`,
`createTableDefinition`, the `_hoodie_*` column list and `createWriteClient`
are verbatim copies of `CustomMergerHudiTablesInitializer` apart from names and
the merger. Worth a shared base taking (tableName, mergerClass, strategyId,
fields, records); follow-up, not a blocker.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -471,12 +602,38 @@ public static List<HiveColumnHandle>
getMergeRequiredColumnHandles(
}
}
- if (requiredColumnNames.isEmpty()) {
- return Collections.emptyList();
- }
return buildColumnHandles(table, typeManager, requiredColumnNames,
timestampPrecision);
}
+ /**
+ * The merge-mandatory column names that do not depend on a custom merger,
mirroring the non-CUSTOM
+ * branch of {@code
FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging}. The delete-marker
and
+ * operation fields are added unconditionally: {@code buildColumnHandles}
drops any name that is not a
+ * data column of the table, so tables without them read nothing extra.
+ */
+ @VisibleForTesting
+ static LinkedHashSet<String> mergeRequiredColumnNames(HoodieTableConfig
tableConfig, RecordMergeMode recordMergeMode)
+ {
+ LinkedHashSet<String> requiredColumnNames = new LinkedHashSet<>();
+ if (recordMergeMode != null && recordMergeMode !=
RecordMergeMode.COMMIT_TIME_ORDERING) {
+ requiredColumnNames.addAll(tableConfig.getOrderingFields());
+ }
+ // Without populated meta fields the file-group reader keys records on
the record-key data columns
+ if (!tableConfig.populateMetaFields()) {
+ tableConfig.getRecordKeyFields().ifPresent(fields ->
requiredColumnNames.addAll(Arrays.asList(fields)));
+ }
+ // Delete markers and the operation field decide record deletion at
merge time
+ requiredColumnNames.add(HOODIE_IS_DELETED_FIELD);
Review Comment:
These two are filtered against the metastore data columns by
`buildColumnHandles`, while `getMandatoryFieldsForMerging` and `DeleteContext`
gate them on the Avro table schema - a table whose metastore omits a field the
schema has (hive-sync with `omit_metadata_fields=true` drops
`_hoodie_operation`, a hand-written external DDL drops `_hoodie_is_deleted`)
loses it from the base projection and the new guard then fails the read. Could
these resolve off the table schema via `toColumnHandle`, the way the CUSTOM
branch already does?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/util/PrefilledColumnValues.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * 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.util;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.HivePartitionKey;
+import io.trino.plugin.hudi.HudiSplit;
+import io.trino.plugin.hudi.file.HudiFile;
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import io.trino.spi.block.RunLengthEncodedBlock;
+
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalInt;
+
+import static io.trino.metastore.Partitions.makePartName;
+import static
io.trino.plugin.hive.HiveColumnHandle.isFileModifiedTimeColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isFileSizeColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isPartitionColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isPathColumnHandle;
+import static io.trino.plugin.hive.util.HiveUtil.getPrefilledColumnValue;
+import static io.trino.spi.type.TypeUtils.writeNativeValue;
+
+/**
+ * Per-split constant values for the output columns that are not stored in the
data file: Hive-style
+ * partition columns and Trino's hidden metadata columns ({@code $path},
{@code $file_size},
+ * {@code $file_modified_time}, {@code $partition}). Value computation
delegates to
+ * {@link io.trino.plugin.hive.util.HiveUtil#getPrefilledColumnValue}, the
same implementation the Hive
+ * connector uses for these columns (including the {@code "\N"} hive-null
partition-value convention).
+ */
+public class PrefilledColumnValues
+{
+ private final Map<String, HivePartitionKey> partitionKeysByName;
+ private final String partitionName;
+ private final String filePath;
+ private final long fileSize;
+ private final long fileModifiedTime;
+
+ public static PrefilledColumnValues create(HudiSplit hudiSplit)
+ {
+ return new PrefilledColumnValues(hudiSplit);
+ }
+
+ private PrefilledColumnValues(HudiSplit hudiSplit)
+ {
+ List<HivePartitionKey> partitionKeys = hudiSplit.getPartitionKeys();
+ // ImmutableMap preserves insertion order, so $partition renders the
keys in the split's
+ // partition-column order.
+ ImmutableMap.Builder<String, HivePartitionKey> byName =
ImmutableMap.builder();
+ partitionKeys.forEach(partitionKey -> byName.put(partitionKey.name(),
partitionKey));
+ this.partitionKeysByName = byName.buildOrThrow();
+ this.partitionName = makePartName(
+ partitionKeys.stream().map(HivePartitionKey::name).toList(),
+ partitionKeys.stream().map(HivePartitionKey::value).toList());
+ // Parquet files will be prioritised over log files
+ HudiFile hudiFile = hudiSplit.getBaseFile().isPresent()
+ ? hudiSplit.getBaseFile().get()
+ : hudiSplit.getLogFiles().getFirst();
+ this.filePath = hudiFile.getPath();
+ this.fileSize = hudiFile.getFileSize();
+ this.fileModifiedTime = hudiFile.getModificationTime();
+ }
+
+ /**
+ * Returns whether this split can provide a value for the column, i.e. it
is a partition column of
+ * the split or a hidden metadata column Hudi populates.
+ */
+ public boolean isPrefilled(HiveColumnHandle columnHandle)
+ {
+ return partitionKeysByName.containsKey(columnHandle.getName())
+ || isPathColumnHandle(columnHandle)
+ || isFileSizeColumnHandle(columnHandle)
+ || isFileModifiedTimeColumnHandle(columnHandle)
+ || isPartitionColumnHandle(columnHandle);
+ }
+
+ /**
+ * Appends the column's value for this split to the builder; appends null
for a column this split
+ * cannot provide.
+ */
+ public void appendTo(HiveColumnHandle columnHandle, BlockBuilder
blockBuilder)
+ {
+ writeNativeValue(columnHandle.getType(), blockBuilder,
nativeValueOf(columnHandle));
+ }
+
+ /**
+ * Builds a run-length-encoded {@link Block} repeating the column's
constant value for
+ * {@code positionCount} positions (null-filled for a column this split
cannot provide).
+ */
+ public Block toRleBlock(HiveColumnHandle columnHandle, int positionCount)
+ {
+ return RunLengthEncodedBlock.create(columnHandle.getType(),
nativeValueOf(columnHandle), positionCount);
+ }
+
+ private Object nativeValueOf(HiveColumnHandle columnHandle)
+ {
+ if (!isPrefilled(columnHandle)) {
+ // Lenient null fill, e.g. for a hidden column Trino defines but
Hudi does not populate.
+ return null;
+ }
+ HivePartitionKey partitionKey =
partitionKeysByName.get(columnHandle.getName());
+ return getPrefilledColumnValue(
Review Comment:
The swap also moves `$file_modified_time` from a UTC-pinned zone to the
worker's default zone - `getPrefilledColumnValue` packs with
`DateTimeZone.getDefault()` where the deleted code used `UTC_KEY`. Same
instant, but the rendered zone and zone-sensitive functions change, so it
belongs in the delta list with the other three - or was UTC the accident?
--
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]
