voonhous commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3662903760
##########
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:
Not out of scope -- done. `appendMissingMergeRequiredColumns` now takes
`readerProps` and, under CUSTOM mode, asks the same resolved merger for
`getMandatoryFieldsForMerging` against the resolved schema, so merger-declared
columns are recovered too. Pinned both ways: a unit test with
`MaxRankRecordMerger`, and a fourth fixture (`omitted_rank_field_mor`) whose
metastore omits `merge_rank` -- its tests fail with the guard error if the
merger-ask is removed.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -348,6 +365,135 @@ 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 projection
+ * expansion is possible without a merger to ask, and {@code
HudiTrinoReaderContext.getRecordMerger}
+ * rejects the read with an actionable error when the merger is actually
needed -- hudi-common's
+ * {@link HoodieRecordUtils#createValidRecordMerger} would otherwise NPE
on the null id.
+ * <p>
+ * Note the payload-based strategy id resolves {@code
HoodieAvroRecordMerger}, which keeps the
+ * interface default {@code isProjectionCompatible() == false} -- so EVERY
custom-payload table takes
+ * the full-schema read path, not just tables with exotic custom mergers.
That matches the file-group
+ * reader (and Spark); the extra I/O is inherent to payload-based merging.
+ */
+ public static boolean usesNonProjectionCompatibleMerger(RecordMergeMode
mergeMode, String mergeStrategyId, String mergeImplClasses)
+ {
+ if (mergeMode != RecordMergeMode.CUSTOM ||
StringUtils.isNullOrEmpty(mergeStrategyId)) {
+ return false;
+ }
+ Option<HoodieRecordMerger> merger =
HoodieRecordUtils.createValidRecordMerger(EngineType.JAVA, mergeImplClasses,
mergeStrategyId);
+ return merger.isPresent() && !merger.get().isProjectionCompatible();
+ }
+
+ /**
+ * Rejects a CUSTOM-mode read that has no merge strategy id to resolve a
record merger with. The
+ * strategy id is only inferred below table version 8 ({@link
#resolveMergeModeAndStrategyId}), so a
+ * version 8 table written without a persisted id reaches merger
resolution with a null id, which
+ * {@link HoodieRecordUtils#createValidRecordMerger} dereferences straight
away.
+ */
+ public static void validateCustomMergeStrategyId(String mergeStrategyId)
+ {
+ if (StringUtils.isNullOrEmpty(mergeStrategyId)) {
+ String strategyIdKey =
HoodieTableConfig.RECORD_MERGE_STRATEGY_ID.key();
+ throw new TrinoException(HUDI_BAD_DATA,
+ ("Table resolved to %s merge mode but persists no `%s`, so
no record merger can be resolved for merging log files. "
+ + "Table version 8 tables written without a
persisted strategy id resolve this way; set `%s` on the table to the "
Review Comment:
Widened both to "version 8 or later".
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -471,12 +622,73 @@ 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}
keeps only metastore data
+ * columns, and the merge read path recovers any name the metastore does
not carry from the resolved
+ * table schema ({@link #appendMissingMergeRequiredColumns}) -- the gate
the file-group reader itself
+ * applies -- so tables whose schema has neither field 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);
+ requiredColumnNames.add(OPERATION_METADATA_FIELD);
+ String deleteKey = tableConfig.getProps().getProperty(DELETE_KEY);
+ String deleteMarker =
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);
+ }
+ return requiredColumnNames;
+ }
+
+ /**
+ * Returns {@code projection} extended with handles for the merge-required
column names it does not
+ * already carry (matched case-insensitively), typed from {@code
dataSchema} via {@link #toColumnHandle};
+ * names absent from the schema as well are dropped. The Avro table schema
is the gate the file-group
Review Comment:
Scoped it: the javadoc now says the schema gate is the reader's own only for
`_hoodie_is_deleted`/`_hoodie_operation`, and that the
ordering/record-key/delete-key names are added without a schema check but are
table columns whenever the config is coherent.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMergeRequiredColumns.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.metastore.HiveType;
+import io.trino.plugin.hive.HiveColumnHandle;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Optional;
+
+import static io.trino.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
+import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static io.trino.plugin.hudi.HudiUtil.appendMissingMergeRequiredColumns;
+import static io.trino.plugin.hudi.HudiUtil.mergeRequiredColumnNames;
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+import static
org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_KEY;
+import static
org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_MARKER;
+import static
org.apache.hudi.common.model.HoodieRecord.HOODIE_IS_DELETED_FIELD;
+import static
org.apache.hudi.common.model.HoodieRecord.OPERATION_METADATA_FIELD;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests {@link HudiUtil#mergeRequiredColumnNames}, which mirrors the
non-CUSTOM branch of the file-group
+ * reader's {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging}
so the base-file read
+ * projection carries every column the merge may consult (ordering fields,
delete markers, the operation
+ * field, record-key data columns) even when the query does not project them,
plus the merge-path recovery
+ * of names the metastore could not resolve ({@link
HudiUtil#appendMissingMergeRequiredColumns}).
+ */
+class TestHudiMergeRequiredColumns
+{
+ @Test
+ public void testOrderingFieldsFollowMergeMode()
+ {
+ HoodieTableConfig tableConfig = new HoodieTableConfig();
+ tableConfig.setValue(HoodieTableConfig.ORDERING_FIELDS, "ts,seq");
+
+ assertThat(mergeRequiredColumnNames(tableConfig,
RecordMergeMode.EVENT_TIME_ORDERING))
+ .contains("ts", "seq");
+ // Commit-time merging ignores ordering fields
+ assertThat(mergeRequiredColumnNames(tableConfig,
RecordMergeMode.COMMIT_TIME_ORDERING))
+ .doesNotContain("ts", "seq");
+ assertThat(mergeRequiredColumnNames(tableConfig, null))
+ .doesNotContain("ts", "seq");
+ }
+
+ @Test
+ public void testDeleteAndOperationColumnsAlwaysRequested()
+ {
+ // Requested unconditionally: getMergeRequiredColumnHandles drops
names that are in neither the
Review Comment:
Reworded to split the claim: the metastore half in
`getMergeRequiredColumnHandles`, the schema half in
`appendMissingMergeRequiredColumns` on the page-source path.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java:
##########
@@ -0,0 +1,291 @@
+/*
+ * 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.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+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.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.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_STRING;
+import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
+import static java.nio.file.Files.createTempDirectory;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Shared machinery for the non-partitioned Merge-On-Read fixtures that exist
to exercise record merging at
+ * read time. The table is written by the Hudi Java write client into a local
staging directory and then
+ * mirrored into the Trino filesystem the connector reads from; two metastore
tables are registered against
+ * that location, a read-optimized one (base files only) and a real-time one
(suffix {@code _rt}) that merges
+ * base + log files through the file group reader.
+ * <p>
+ * Inline compaction is off for every fixture so the log files written by the
delta commits survive and must
+ * be merged at read time, and the metadata table is off because MDT writes
need hbase dependencies that are
+ * not on the Trino test classpath.
+ * <p>
+ * Subclasses supply the schema, the merge-related table and write
configuration, and the commits; everything
+ * a fixture does not vary lives here.
+ */
+public abstract class AbstractMergerHudiTablesInitializer
+ implements HudiTablesInitializer
+{
+ /** Every fixture built on this base is keyed on {@code key} and ordered
by {@code ts}, in a single unnamed partition. */
+ protected static final String RECORD_KEY_FIELD = "key";
+ protected static final String ORDERING_FIELD = "ts";
+
+ private static final String PARTITION_PATH = "";
+
+ /** Hudi metadata columns, prepended to every table's data columns in the
metastore definition. */
+ private static final List<Column> HUDI_META_COLUMNS = ImmutableList.of(
+ new Column("_hoodie_commit_time", HIVE_STRING, Optional.empty(),
Map.of()),
+ new Column("_hoodie_commit_seqno", HIVE_STRING, Optional.empty(),
Map.of()),
+ new Column("_hoodie_record_key", HIVE_STRING, Optional.empty(),
Map.of()),
+ new Column("_hoodie_partition_path", HIVE_STRING,
Optional.empty(), Map.of()),
+ new Column("_hoodie_file_name", HIVE_STRING, Optional.empty(),
Map.of()));
+
+ private final String tableName;
+
+ private TrinoFileSystem fileSystem;
+ private Location tableLocation;
+ private java.nio.file.Path stagingDir;
+ private Path stagingTablePath;
+ private HoodieJavaWriteClient<HoodieAvroPayload> writeClient;
+
+ protected AbstractMergerHudiTablesInitializer(String tableName)
+ {
+ this.tableName = requireNonNull(tableName, "tableName is null");
+ }
+
+ @Override
+ public final void initializeTables(QueryRunner queryRunner, Location
externalLocation, String schemaName)
+ throws Exception
+ {
+ fileSystem = ((HudiConnector)
queryRunner.getCoordinator().getConnector("hudi")).getInjector()
+ .getInstance(TrinoFileSystemFactory.class)
+ .create(ConnectorIdentity.ofUser("test"));
+ HiveMetastore metastore = ((HudiConnector)
queryRunner.getCoordinator().getConnector("hudi")).getInjector()
+ .getInstance(HiveMetastoreFactory.class)
+ .createMetastore(Optional.empty());
+
+ tableLocation = externalLocation.appendPath(tableName);
+ stagingDir = createTempDirectory(tableName.replace('_', '-'));
+ stagingTablePath = new Path(stagingDir.resolve(tableName).toUri());
+
+ boolean initialized = false;
+ try {
+ initTable();
+ afterTableInit();
+ writeClient = createWriteClient();
+ writeInitialCommits(writeClient);
+ syncToTrino();
+
+ metastore.createTable(createTableDefinition(schemaName, tableName,
false), PrincipalPrivileges.NO_PRIVILEGES);
+ metastore.createTable(createTableDefinition(schemaName, tableName
+ "_rt", true), PrincipalPrivileges.NO_PRIVILEGES);
+ initialized = true;
+ }
+ finally {
+ // Only fixtures that keep writing commits after initialization
need the staging directory and the
+ // write client to outlive this call; they are responsible for
calling close().
+ if (!initialized || !keepsWriterOpen()) {
+ close();
+ }
+ }
+ }
+
+ /** The table's data columns, in schema order; the Hudi metadata columns
are prepended by this class. */
+ protected abstract List<Column> dataColumns();
+
+ /** The Avro schema the write client writes, matching {@link
#dataColumns()}. */
+ protected abstract Schema avroSchema();
+
+ /**
+ * Applies the fixture's merge-related table configuration (merge mode,
merge strategy id, payload class).
+ * The table type, record key fields and ordering fields are set by this
class.
+ */
+ protected abstract void
configureTableConfig(HoodieTableMetaClient.TableBuilder tableBuilder);
+
+ /** Applies the fixture's merge-related write configuration (merge mode,
merge strategy id, merger impl classes). */
+ protected abstract void configureWriteConfig(HoodieWriteConfig.Builder
writeConfigBuilder);
+
+ /** Writes the commits that make up the fixture's initial state. */
+ protected abstract void
writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ throws IOException;
+
+ /** Runs on the freshly created table, before the write client exists and
before any data is written. */
+ protected void afterTableInit()
+ throws IOException {}
+
+ /** Inline compaction is disabled, so this only has to stay above the
number of delta commits a fixture writes. */
+ protected int maxDeltaCommitsBeforeCompaction()
+ {
+ return 100;
+ }
+
+ /** Whether the staging directory and write client survive {@link
#initializeTables}; such fixtures must call {@link #close()}. */
+ protected boolean keepsWriterOpen()
+ {
+ return false;
+ }
+
+ public void close()
+ throws IOException
+ {
+ if (writeClient != null) {
+ writeClient.close();
+ writeClient = null;
+ }
+ if (stagingDir != null) {
+ deleteRecursively(stagingDir, ALLOW_INSECURE);
+ stagingDir = null;
+ }
+ }
+
+ /** Local directory the write client writes into, before {@link
#syncToTrino()} mirrors it to the connector. */
+ protected java.nio.file.Path stagingTableDirectory()
+ {
+ return stagingDir.resolve(tableName);
+ }
+
+ protected HoodieJavaWriteClient<HoodieAvroPayload> writeClient()
+ {
+ return writeClient;
+ }
+
+ protected static HoodieRecord<HoodieAvroPayload> avroRecord(GenericRecord
record, String key)
+ {
+ return new HoodieAvroRecord<>(new HoodieKey(key, PARTITION_PATH), new
HoodieAvroPayload(Option.of(record)), null);
+ }
+
+ /** Mirrors the staged table into the Trino filesystem so the connector
observes the commits written so far. */
+ protected void syncToTrino()
+ {
+ try {
+ if (fileSystem.directoryExists(tableLocation).orElse(false)) {
+ fileSystem.deleteDirectory(tableLocation);
+ }
+ ResourceHudiTablesInitializer.copyDir(stagingTableDirectory(),
fileSystem, tableLocation);
+ }
+ catch (IOException e) {
+ throw new RuntimeException("Failed to sync staged Hudi table to
Trino filesystem", e);
+ }
+ }
+
+ private void initTable()
+ {
+ HoodieTableMetaClient.TableBuilder tableBuilder =
HoodieTableMetaClient.newTableBuilder()
+ .setTableType(HoodieTableType.MERGE_ON_READ)
+ .setTableName(tableName)
+ .setTimelineLayoutVersion(1)
+ .setBootstrapIndexClass(NoOpBootstrapIndex.class.getName())
+ .setRecordKeyFields(RECORD_KEY_FIELD)
+ .setOrderingFields(ORDERING_FIELD);
+ configureTableConfig(tableBuilder);
+ try {
+ tableBuilder.initTable(new HadoopStorageConfiguration(new
Configuration()), stagingTablePath.toString());
+ }
+ catch (IOException e) {
+ throw new RuntimeException("Could not init table " + tableName, e);
+ }
+ }
+
+ private HoodieJavaWriteClient<HoodieAvroPayload> createWriteClient()
+ {
+ Configuration conf = new Configuration();
+ HoodieWriteConfig.Builder writeConfigBuilder =
HoodieWriteConfig.newBuilder()
+ .withPath(stagingTablePath.toString())
+ .withSchema(avroSchema().toString())
+ .withParallelism(2, 2)
+ .withDeleteParallelism(2)
+ .forTable(tableName)
+
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build())
+ // Keep log files around so merging runs at read time.
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withInlineCompaction(false)
+
.withMaxNumDeltaCommitsBeforeCompaction(maxDeltaCommitsBeforeCompaction())
+ .build())
+ .withEmbeddedTimelineServerEnabled(false)
+ .withMarkersType(MarkerType.DIRECT.name())
+ // MDT writes require hbase deps not present in the Trino
runtime.
+
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build());
+ // The ordering field is carried by the table config
(setOrderingFields); the deprecated
Review Comment:
Moved it back into the chain, at the spot `withPreCombineField` would go.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/CompositeHudiTablesInitializer.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.filesystem.Location;
+import io.trino.testing.QueryRunner;
+
+import java.util.List;
+
+/**
+ * Runs several {@link HudiTablesInitializer}s against one query runner, in
order, so a single test class can
+ * load more than one table fixture. Each delegate must create its own tables
under a distinct name.
+ */
+public class CompositeHudiTablesInitializer
+ implements HudiTablesInitializer
+{
+ private final List<HudiTablesInitializer> delegates;
+
+ public CompositeHudiTablesInitializer(HudiTablesInitializer... delegates)
+ {
+ this.delegates = ImmutableList.copyOf(delegates);
+ }
+
+ @Override
+ public void initializeTables(QueryRunner queryRunner, Location
externalLocation, String schemaName)
+ throws Exception
+ {
+ for (HudiTablesInitializer delegate : delegates) {
+ delegate.initializeTables(queryRunner, externalLocation,
schemaName);
Review Comment:
Went with the javadoc: the composite now says delegates must finish their
writes within `initializeTables`, and that a `keepsWriterOpen()` fixture can't
be composed since the interface has no close to forward. Didn't want to grow
the interface for a case nothing needs yet.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/PayloadOnlyMergerHudiTablesInitializer.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table at TABLE VERSION 6 whose only
merge configuration is a
+ * {@link org.apache.hudi.common.model.HoodieRecordPayload} class ({@link
RankBasedTestPayload}), the way a
+ * genuine pre-1.0 table looks on storage.
+ * <p>
+ * Such a table resolves at read time to {@code CUSTOM} merge mode with the
payload-based merge strategy, which
+ * in turn resolves {@code HoodieAvroRecordMerger}. That merger is not
projection compatible, so the file group
+ * reader demands the FULL table schema for base and log reads alike and
nothing prepends the payload's decision
+ * column into the connector's read projection.
+ * <p>
+ * One {@code bulkInsert} (base files) is followed by one {@code upsert} of
the same keys (log files), with data
+ * laid out so each merge direction is distinguishable: {@code k1}'s winning
rank is on the LOG record and
+ * {@code k2}'s on the BASE record. See {@code
TestHudiNonProjectionCompatibleMerger}.
+ */
+public class PayloadOnlyMergerHudiTablesInitializer
+ extends AbstractMergerHudiTablesInitializer
+{
+ public static final String TABLE_NAME = "payload_only_mor";
+ public static final String RT_TABLE_NAME = TABLE_NAME + "_rt";
+
+ private static final String RANK_FIELD = RankBasedTestPayload.RANK_COLUMN;
+ private static final HoodieTableVersion TABLE_VERSION =
HoodieTableVersion.SIX;
+
+ public PayloadOnlyMergerHudiTablesInitializer()
+ {
+ 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(RANK_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("name", Schema.create(Schema.Type.STRING)),
+ new Schema.Field("value", Schema.create(Schema.Type.LONG)),
+ new Schema.Field(RANK_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)
+ {
+ // Only the payload class: the merge mode and the merge strategy id
must stay out of hoodie.properties
+ // so the reader has to infer them, which is what makes this a pre-1.0
payload-only table.
+ tableBuilder
+ .setTableVersion(TABLE_VERSION)
+ .setPayloadClassName(RankBasedTestPayload.class.getName());
+ }
+
+ @Override
+ protected void configureWriteConfig(HoodieWriteConfig.Builder
writeConfigBuilder)
+ {
+ writeConfigBuilder
+ .withWriteTableVersion(TABLE_VERSION.versionCode())
+ // Without this the write client silently upgrades the table
to the current version on the first commit.
+ .withAutoUpgradeVersion(false)
+ .withWritePayLoad(RankBasedTestPayload.class.getName());
+ }
+
+ @Override
+ protected void afterTableInit()
+ throws IOException
+ {
+ stripInferredMergeConfigs();
+ }
+
+ @Override
+ protected void
writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ throws IOException
+ {
+ Schema schema = avroSchema();
+ // First commit: bulk insert base records (produces base parquet
files).
+ String firstCommit = client.startCommit();
+ List<WriteStatus> firstStatuses = client.bulkInsert(ImmutableList.of(
+ record(schema, "k1", "k1_base", 10L, 5L, 1L),
+ record(schema, "k2", "k2_base", 100L, 9L, 1L)), firstCommit);
+ client.commit(firstCommit, firstStatuses);
+
+ // Second commit: upserts the same keys (produces log files since
inline compaction is disabled).
+ // k1 update has a HIGHER rank (7 > 5) -> the payload keeps the update
(99): the LOG record's rank decides.
+ // k2 update has a LOWER rank (1 < 9) -> the payload keeps the base
record (100): the BASE record's rank
+ // decides, which only works when the base read carries merge_rank
despite it never being projected.
+ String secondCommit = client.startCommit();
+ List<WriteStatus> secondStatuses = client.upsert(ImmutableList.of(
+ record(schema, "k1", "k1_updated", 99L, 7L, 2L),
+ record(schema, "k2", "k2_updated", 4L, 1L, 2L)), secondCommit);
+ client.commit(secondCommit, secondStatuses);
+
+ // The commits go through the table config; re-check that the writer
left the fixture payload-only.
+ stripInferredMergeConfigs();
+ }
+
+ /**
+ * Removes the merge mode and merge strategy id that table creation infers
and persists for every pre-v9
+ * table, then verifies what is left. The fixture has to mimic a genuine
pre-1.0 table, which persists its
+ * payload class and nothing else about merging; leaving the inferred keys
in would hand the reader the
+ * answer this fixture exists to make it derive.
+ */
+ private void stripInferredMergeConfigs()
+ throws IOException
+ {
+ Path metaDirectory = stagingTableDirectory().resolve(".hoodie");
+ Path propertiesFile = metaDirectory.resolve("hoodie.properties");
+ List<String> retainedLines = Files.readAllLines(propertiesFile,
UTF_8).stream()
+ .filter(line -> !isEntryFor(line,
HoodieTableConfig.RECORD_MERGE_MODE.key())
Review Comment:
Checked `dropInvalidConfigs` and confirmed: the mode's config is
since-version 1.0.0, so it never reaches disk for a v6 table. The filter now
strips just the strategy id, the javadoc explains why, and the `checkAbsent` on
the mode stays as a pin of creation's own behavior.
##########
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:
Folded them all into the Summary and Changelog: the `$file_modified_time`
zone change joins the `PrefilledColumnValues` delta list, and the null-fill
under `use-column-names=false`, the `HUDI_BAD_DATA` guard, and the
metastore-omitted recovery each have their own bullet now. Also refreshed the
stale "stacked on #18837" opener while in there.
--
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]
