wombatu-kun commented on code in PR #18837:
URL: https://github.com/apache/hudi/pull/18837#discussion_r3619021810
##########
.github/workflows/hudi_trino_ci.yml:
##########
@@ -0,0 +1,67 @@
+name: Hudi Trino Connector CI
+
+on:
+ push:
+ branches:
+ - master
+ - 'release-*'
+ paths:
+ - 'hudi-trino/**'
+ - '.github/workflows/hudi_trino_ci.yml'
+ pull_request:
+ branches:
+ - master
+ - 'release-*'
+ paths:
+ - 'hudi-trino/**'
Review Comment:
test-hudi-trino-plugin is a required status check in .asf.yaml, but gating
this workflow at on: pull_request: paths means it never runs on PRs that do not
touch hudi-trino, leaving the required check permanently pending and blocking
merges. Run the workflow on every PR and guard the expensive build/test steps
with a path-based `if:` (as bot.yml does via needs.changes) so the required
context still reports.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiTableHandle.java:
##########
@@ -248,6 +261,18 @@ public TupleDomain<HiveColumnHandle> getRegularPredicates()
return regularPredicates;
}
+ @JsonProperty
+ public OptionalLong getLimit()
+ {
+ return limit;
+ }
+
+ @JsonProperty
+ public List<HiveColumnHandle> getOrderingColumns()
Review Comment:
getOrderingColumns (field lazyOrderingColumns, JSON property
orderingColumns) is populated by getMergeRequiredColumnHandles, so for a CUSTOM
merger it also carries the merger's mandatory fields, not just ordering
columns. Rename it to mergeRequiredColumns / getMergeRequiredColumns to match
what it actually holds.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/reader/HudiTrinoReaderContext.java:
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.reader;
+
+import io.trino.metastore.HiveType;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hudi.util.HudiAvroSerializer;
+import io.trino.plugin.hudi.util.SynthesizedColumnHandler;
+import io.trino.spi.Page;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.SourcePage;
+import io.trino.spi.type.VarcharType;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hudi.common.avro.AvroRecordContext;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.engine.EngineType;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieAvroRecordMerger;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordMerger;
+import org.apache.hudi.common.model.OverwriteWithLatestMerger;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.util.HoodieRecordUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.storage.inline.InLineFSUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
+import static java.lang.String.format;
+import static
org.apache.hudi.common.config.HoodieReaderConfig.RECORD_MERGE_IMPL_CLASSES_WRITE_CONFIG_KEY;
+
+public class HudiTrinoReaderContext
+ extends HoodieReaderContext<IndexedRecord>
+{
+ ConnectorPageSource pageSource;
+ private final HudiAvroSerializer avroSerializer;
+ private final SynthesizedColumnHandler synthesizedColumnHandler;
+ private final LogFileParquetPageSourceFactory logPageSourceFactory;
+ Map<String, Integer> colToPosMap;
+ Map<String, HiveColumnHandle> colNameToHandle;
+ List<HiveColumnHandle> dataHandles;
+ List<HiveColumnHandle> columnHandles;
+
+ /**
+ * Factory for building a Trino parquet page source over a native
(RFC-103) delta-log file on demand.
+ * Kept as an injected interface so this reader context stays free of
Trino-parquet/session types.
+ */
+ @FunctionalInterface
+ public interface LogFileParquetPageSourceFactory
+ {
+ ConnectorPageSource create(String path, long start, long length,
List<HiveColumnHandle> projection);
+ }
+
+ public HudiTrinoReaderContext(
+ StorageConfiguration storageConfiguration,
+ HoodieTableConfig tableConfig,
+ ConnectorPageSource pageSource,
+ List<HiveColumnHandle> dataHandles,
+ List<HiveColumnHandle> columnHandles,
+ SynthesizedColumnHandler synthesizedColumnHandler,
+ LogFileParquetPageSourceFactory logPageSourceFactory)
+ {
+ super(storageConfiguration, tableConfig, Option.empty(),
Option.empty(), new AvroRecordContext(tableConfig,
tableConfig.getPayloadClass()));
+ this.pageSource = pageSource;
+ this.synthesizedColumnHandler = synthesizedColumnHandler;
+ this.avroSerializer = new HudiAvroSerializer(columnHandles,
synthesizedColumnHandler);
+ this.dataHandles = dataHandles;
+ this.columnHandles = columnHandles;
+ this.logPageSourceFactory = logPageSourceFactory;
+ this.colToPosMap = new HashMap<>();
+ this.colNameToHandle = new HashMap<>();
+ for (int i = 0; i < columnHandles.size(); i++) {
+ HiveColumnHandle handle = columnHandles.get(i);
+ colToPosMap.put(handle.getBaseColumnName(), i);
+
colNameToHandle.put(handle.getBaseColumnName().toLowerCase(Locale.ROOT),
handle);
+ }
+ }
+
+ @Override
+ public ClosableIterator<IndexedRecord> getFileRecordIterator(
+ StoragePath storagePath,
+ long start,
+ long length,
+ HoodieSchema dataSchema,
+ HoodieSchema requiredSchema,
+ HoodieStorage storage)
+ {
+ return getFileRecordIterator(storagePath, start, length,
requiredSchema);
+ }
+
+ @Override
+ public ClosableIterator<IndexedRecord> getFileRecordIterator(
+ StoragePathInfo storagePathInfo,
+ long start,
+ long length,
+ HoodieSchema dataSchema,
+ HoodieSchema requiredSchema,
+ HoodieStorage storage)
+ {
+ return getFileRecordIterator(storagePathInfo.getPath(), start, length,
requiredSchema);
+ }
+
+ /**
+ * Reads the given file and projects {@code requiredSchema}. For a native
(RFC-103) delta-log parquet file a
+ * fresh page source is built on demand with predicate pushdown disabled
so every log record is read and
+ * merged; for the base file the pre-built base page source is reused.
Classic Avro log blocks never reach
+ * here (they deserialize inline).
+ */
+ private ClosableIterator<IndexedRecord> getFileRecordIterator(
+ StoragePath path,
+ long start,
+ long length,
+ HoodieSchema requiredSchema)
+ {
+ if (FSUtils.isLogFile(path)) {
+ // Inline parquet log blocks (inlinefs:// scheme) need a separate
inline-aware reader; out of scope here.
+ if (InLineFSUtils.SCHEME.equals(path.toUri().getScheme())) {
+ throw new UnsupportedOperationException("Inline log blocks are
not supported by the Hudi Trino connector: " + path);
+ }
+ List<HiveColumnHandle> logProjection =
buildRequiredColumnHandles(requiredSchema);
+ ConnectorPageSource logSource =
logPageSourceFactory.create(path.toString(), start, length, logProjection);
+ HudiAvroSerializer logSerializer = new
HudiAvroSerializer(logProjection, synthesizedColumnHandler);
+ return createRecordIterator(logSource, logSerializer);
+ }
+ return createRecordIterator(pageSource, avroSerializer);
+ }
+
+ /**
+ * Resolves the {@link HiveColumnHandle} for each field of {@code
requiredSchema} against the reader's column
+ * handles (keyed by lowercased base column name) so the on-demand log
page source reads exactly the columns
+ * the file-group reader needs to merge. The file-group reader can add
meta fields to {@code requiredSchema}
+ * that the connector projection does not carry -- the projection holds
only the query columns plus
+ * {@code HUDI_REQUIRED_META_COLUMNS} (record key + partition path),
whereas the merge may also need e.g.
+ * {@code _hoodie_commit_time}; for such a Hudi meta column a handle is
synthesized (see the inline note)
+ * rather than failing the read.
+ */
+ private List<HiveColumnHandle> buildRequiredColumnHandles(HoodieSchema
requiredSchema)
+ {
+ List<HiveColumnHandle> handles = new ArrayList<>();
+ for (HoodieSchemaField field : requiredSchema.getFields()) {
+ String name = field.name();
+ HiveColumnHandle handle =
colNameToHandle.get(name.toLowerCase(Locale.ROOT));
+ if (handle == null) {
+ if (!HoodieRecord.HOODIE_META_COLUMNS.contains(name)) {
+ // A data column outside the projection cannot be typed at
this layer. The file-group reader
+ // only asks for one when the table's custom merger is not
projection compatible (it then
+ // reads the FULL table schema), which this connector does
not support.
+ throw new TrinoException(NOT_SUPPORTED, format(
Review Comment:
This NOT_SUPPORTED branch for a non-projection-compatible custom merger is
never exercised, since both test mergers return isProjectionCompatible()=true.
Add a negative test that reads a table whose merger reports false and asserts
this error - follow-up, not a blocker.
--
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]
