wombatu-kun commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3662219964
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -192,33 +201,52 @@ public ConnectorPageSource createPageSource(
.withBloomFilter(useParquetBloomFilter(session))
.withVectorizedDecodingEnabled(isParquetVectorizedDecodingEnabled(session))
.build();
- ConnectorPageSource dataPageSource = createPageSource(
- session,
- isBaseFileOnly ? dataColumnHandles :
hudiMetaAndDataColumnHandles,
- hudiSplit,
-
fileSystem.newInputFile(Location.of(hudiBaseFileOpt.get().getPath()),
hudiBaseFileOpt.get().getFileSize()),
- hudiBaseFileOpt.get().getPath(),
- start,
- length,
- OptionalLong.of(hudiBaseFileOpt.get().getFileSize()),
- dataSourceStats,
- sessionOptions,
- timeZone, dynamicFilter, isBaseFileOnly);
-
- SynthesizedColumnHandler synthesizedColumnHandler =
SynthesizedColumnHandler.create(hudiSplit);
+ PrefilledColumnValues prefilledColumnValues =
PrefilledColumnValues.create(hudiSplit);
// Avoid avro serialization if split/filegroup only contains base files
if (isBaseFileOnly) {
return new HudiBaseFileOnlyPageSource(
- dataPageSource,
+ createBaseFilePageSource(session, dataColumnHandles,
hudiSplit, fileSystem, sessionOptions, start, length, dynamicFilter, true),
hiveColumnHandles,
dataColumnHandles,
- synthesizedColumnHandler);
+ prefilledColumnValues);
+ }
+
+ // The merge path below is built around a base-file page source; fail
log-only file slices with a
+ // clear error instead of an opaque NoSuchElementException from
getBaseFile().orElseThrow().
+ // TODO: support log-only file slices by feeding the file-group reader
an empty base page source.
+ if (hudiBaseFileOpt.isEmpty()) {
+ throw new TrinoException(NOT_SUPPORTED, "Hudi splits with log
files but no base file are not supported: "
+ + hudiSplit.getLogFiles().getFirst().getPath());
}
// TODO: Move this into HudiTableHandle
HoodieTableMetaClient metaClient = buildTableMetaClient(
fileSystemFactory.create(session),
hudiTableHandle.getSchemaTableName().toString(), hudiTableHandle.getBasePath());
+ HoodieSchema dataSchema =
+ Optional.ofNullable(hudiTableHandle.getTableSchema())
+ .orElseGet(() -> getLatestTableSchema(metaClient,
hudiTableHandle.getTableName()));
+ TypedProperties readerProps = buildReaderProperties(session,
metaClient);
+
+ // A non-projection-compatible CUSTOM merger makes the file-group
reader demand the FULL table
+ // schema as requiredSchema for this split (it has log files), for the
base and log reads alike.
+ // Expand the read projection to the full schema up front so the base
page source carries every
+ // merge column; the log page sources resolve their columns from the
same expanded handles.
+ List<HiveColumnHandle> readColumnHandles;
+ if (requiresFullSchemaRead(metaClient.getTableConfig(), readerProps)) {
+ log.debug("Expanding the read projection of %s to the full table
schema: the resolved record merger is not projection compatible",
+ hudiTableHandle.getSchemaTableName());
+ readColumnHandles = appendMissingSchemaColumns(dataSchema,
hudiMetaAndDataColumnHandles);
+ }
+ else {
+ // The metastore may lack merge-required columns the table schema
carries (e.g. hive sync with
+ // omit_metadata_fields=true drops _hoodie_operation); recover
them from the already-resolved
+ // schema so the base read is not starved of them.
+ readColumnHandles = appendMissingMergeRequiredColumns(dataSchema,
hudiMetaAndDataColumnHandles, metaClient.getTableConfig());
Review Comment:
Nothing fails if this line goes back to `readColumnHandles =
hudiMetaAndDataColumnHandles` - `TestHudiMergeRequiredColumns` only calls the
helper directly, and every fixture's metastore column list matches its Avro
schema exactly. Worth one fixture whose metastore omits a merge-required column
its schema carries, so the recovery is pinned end to end.
##########
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:
The file-group reader schema-gates only `_hoodie_operation` and
`_hoodie_is_deleted`; ordering fields, record-key fields and the custom delete
key it adds unconditionally. Worth scoping the sentence to those two, since
this method gates all five.
##########
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:
This comment used to sit inside the builder chain where
`withPreCombineField` would have gone; above `configureWriteConfig` it reads as
documentation of a hook that has nothing to do with pre-combine. Worth moving
it back into the chain or dropping it.
##########
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:
Table creation never persists `hoodie.record.merge.mode` for a version 6
table - `HoodieTableConfig.dropInvalidConfigs` drops it because the config's
since-version is 1.0.0 - so only the strategy id is ever written and this
filter plus its `checkAbsent` assert nothing. Worth dropping the merge-mode
half, or rewording the javadoc, which says both are persisted.
##########
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:
A delegate that returns `keepsWriterOpen()` true keeps its write client and
staging directory alive, and the composite has no way to release them since
`HudiTablesInitializer` declares no close. Worth saying so in the javadoc, or
making the interface `AutoCloseable` with a no-op default so the composite can
forward it.
##########
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:
The strategy id is passed through unchanged from table version 8 upward, so
a version 9 or 10 table whose config says CUSTOM without an id lands here too
and gets told this is a version 8 situation. Worth widening the message and the
javadoc to "version 8 or later".
##########
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:
`getMergeRequiredColumnHandles` drops every name the metastore lacks,
including names the Avro table schema does carry - the schema half of the gate
lives in `appendMissingMergeRequiredColumns`, on the page-source path. Worth
splitting the claim across the two, since as written it describes a union gate
that no single function implements.
--
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]
