Skip to content

Support MOR snapshot query and MDT-based file listing in Trino Hudi connector #25169

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 2 commits into
base: master
Choose a base branch
from
+1,740 −2,871
Draft

Support MOR snapshot query and MDT-based file listing in Trino Hudi connector #25169

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5375fb3
Add MOR snapshot query and MDT-based file listing in Trino Hudi conne…
yihua Feb 26, 2025
86b1681
Fix dependency and style checks
yihua Mar 6, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions plugin/trino-hive/src/main/java/io/trino/plugin/hive/avro/AvroHiveFileUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ public final class AvroHiveFileUtils
{
private final AtomicInteger recordNameSuffix = new AtomicInteger(0);

private AvroHiveFileUtils() {}
public AvroHiveFileUtils() {}

// Lifted and shifted from org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils.determineSchemaOrThrowException
public static Schema determineSchemaOrThrowException(TrinoFileSystem fileSystem, Map<String, String> properties)
Expand Down Expand Up @@ -146,7 +146,7 @@ private static Schema constructSchemaFromParts(List<String> columnNames, List<Hi
return fieldBuilder.endRecord();
}

private Schema avroSchemaForHiveType(HiveType hiveType)
public Schema avroSchemaForHiveType(HiveType hiveType)
{
Schema schema = switch (hiveType.getCategory()) {
case PRIMITIVE -> createAvroPrimitive(hiveType);
Expand Down
77 changes: 43 additions & 34 deletions plugin/trino-hudi/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,18 @@

<properties>
<air.compiler.fail-warnings>true</air.compiler.fail-warnings>
<dep.hudi.version>1.0.1</dep.hudi.version>
<dep.hudi.version>0.15.0</dep.hudi.version>
<dep.hbase.version>2.4.9</dep.hbase.version>
</properties>

<dependencies>
<dependency>
<!--Used to test execution in task executor after de-serializing-->
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo</artifactId>
<version>4.0.3</version>
</dependency>

<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
Expand Down Expand Up @@ -120,6 +128,40 @@
<artifactId>avro</artifactId>
</dependency>

<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-common</artifactId>
<version>${dep.hudi.version}</version>
<exclusions>
<exclusion>
<groupId>com.fasterxml.jackson.module</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>io.dropwizard.metrics</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.hbase</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.orc</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>

<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-io</artifactId>
<version>${dep.hudi.version}</version>
</dependency>

<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-column</artifactId>
Expand Down Expand Up @@ -202,13 +244,6 @@
<scope>runtime</scope>
</dependency>

<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo-shaded</artifactId>
<version>4.0.3</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>io.airlift</groupId>
<artifactId>junit-extensions</artifactId>
Expand Down Expand Up @@ -309,19 +344,6 @@
</exclusions>
</dependency>

<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-common</artifactId>
<version>${dep.hudi.version}</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>*</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>

<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-hadoop-common</artifactId>
Expand All @@ -335,19 +357,6 @@
</exclusions>
</dependency>

<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-io</artifactId>
<version>${dep.hudi.version}</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>*</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>

<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-java-client</artifactId>
Expand Down
17 changes: 15 additions & 2 deletions plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/HudiConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,12 +31,12 @@

@DefunctConfig({
"hudi.min-partition-batch-size",
"hudi.max-partition-batch-size",
"hudi.metadata-enabled",
"hudi.max-partition-batch-size"
})
public class HudiConfig
{
private List<String> columnsToHide = ImmutableList.of();
private boolean metadataEnabled = true;
private boolean shouldUseParquetColumnNames = true;
private boolean sizeBasedSplitWeightsEnabled = true;
private DataSize standardSplitWeightSize = DataSize.of(128, MEGABYTE);
Expand Down Expand Up @@ -64,6 +64,19 @@ public HudiConfig setColumnsToHide(List<String> columnsToHide)
return this;
}

@Config("hudi.metadata-enabled")
@ConfigDescription("If enabled, Hudi's metadata table is used for file listing and data skipping.")
public HudiConfig setMetadataEnabled(boolean metadataEnabled)
{
this.metadataEnabled = metadataEnabled;
return this;
}

public boolean isMetadataEnabled()
{
return this.metadataEnabled;
}

@Config("hudi.parquet.use-column-names")
@ConfigDescription("Access Parquet columns using names from the file. If disabled, then columns are accessed using index."
+ "Only applicable to Parquet file format.")
Expand Down
4 changes: 2 additions & 2 deletions plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/HudiErrorCode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,8 @@ public enum HudiErrorCode
HUDI_CURSOR_ERROR(6, EXTERNAL),
HUDI_FILESYSTEM_ERROR(7, EXTERNAL),
HUDI_PARTITION_NOT_FOUND(8, EXTERNAL),
// HUDI_UNSUPPORTED_TABLE_TYPE(9, EXTERNAL), // Unused. Could be mistaken with HUDI_UNKNOWN_TABLE_TYPE.

HUDI_UNSUPPORTED_TABLE_TYPE(9, EXTERNAL),
HUDI_NO_VALID_COMMIT(10, EXTERNAL)
/**/;

private final ErrorCode errorCode;
Expand Down
17 changes: 15 additions & 2 deletions plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/HudiMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,16 @@
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.Table;
import io.trino.metastore.TableInfo;
import io.trino.plugin.base.classloader.ClassLoaderSafeSystemTable;
import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hudi.storage.HudiTrinoStorage;
import io.trino.plugin.hudi.storage.TrinoStorageConfiguration;
import io.trino.spi.TrinoException;
import io.trino.spi.connector.ColumnHandle;
import io.trino.spi.connector.ColumnMetadata;
Expand All @@ -40,6 +43,9 @@
import io.trino.spi.connector.TableNotFoundException;
import io.trino.spi.predicate.TupleDomain;
import io.trino.spi.type.TypeManager;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.storage.StoragePath;

import java.util.Collection;
import java.util.Collections;
Expand Down Expand Up @@ -69,7 +75,6 @@
import static io.trino.plugin.hudi.HudiTableProperties.LOCATION_PROPERTY;
import static io.trino.plugin.hudi.HudiTableProperties.PARTITIONED_BY_PROPERTY;
import static io.trino.plugin.hudi.HudiUtil.hudiMetadataExists;
import static io.trino.plugin.hudi.model.HudiTableType.COPY_ON_WRITE;
import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
import static io.trino.spi.StandardErrorCode.QUERY_REJECTED;
import static io.trino.spi.StandardErrorCode.UNSUPPORTED_TABLE_TYPE;
Expand All @@ -78,6 +83,7 @@
import static java.util.Collections.singletonList;
import static java.util.Objects.requireNonNull;
import static java.util.function.Function.identity;
import static org.apache.hudi.common.model.HoodieTableType.COPY_ON_WRITE;

public class HudiMetadata
implements ConnectorMetadata
Expand Down Expand Up @@ -119,15 +125,22 @@ public HudiTableHandle getTableHandle(ConnectorSession session, SchemaTableName
throw new TrinoException(UNSUPPORTED_TABLE_TYPE, format("Not a Hudi table: %s", tableName));
}
Location location = Location.of(table.get().getStorage().getLocation());
if (!hudiMetadataExists(fileSystemFactory.create(session), location)) {
TrinoFileSystem fileSystem = fileSystemFactory.create(session);
if (!hudiMetadataExists(fileSystem, location)) {
throw new TrinoException(HUDI_BAD_DATA, "Location of table %s does not contain Hudi table metadata: %s".formatted(tableName, location));
}
StoragePath metaLocation = new StoragePath(
table.get().getStorage().getLocation(), HoodieTableMetaClient.METAFOLDER_NAME);
HoodieTableConfig tableConfig = new HoodieTableConfig(
new HudiTrinoStorage(fileSystem, new TrinoStorageConfiguration()), metaLocation, null, null);
String preCombineField = tableConfig.getPreCombineField();

return new HudiTableHandle(
tableName.getSchemaName(),
tableName.getTableName(),
table.get().getStorage().getLocation(),
COPY_ON_WRITE,
preCombineField,
getPartitionKeyColumnHandles(table.get(), typeManager),
TupleDomain.all(),
TupleDomain.all());
Expand Down
95 changes: 76 additions & 19 deletions plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,9 @@
import io.trino.plugin.hive.HivePartitionKey;
import io.trino.plugin.hive.ReaderColumns;
import io.trino.plugin.hive.parquet.ParquetReaderConfig;
import io.trino.plugin.hudi.model.HudiFileFormat;
import io.trino.plugin.hudi.file.HudiBaseFile;
import io.trino.plugin.hudi.storage.HudiTrinoStorage;
import io.trino.plugin.hudi.storage.TrinoStorageConfiguration;
import io.trino.spi.TrinoException;
import io.trino.spi.block.Block;
import io.trino.spi.connector.ColumnHandle;
Expand All @@ -46,9 +48,12 @@
import io.trino.spi.connector.ConnectorTableHandle;
import io.trino.spi.connector.ConnectorTransactionHandle;
import io.trino.spi.connector.DynamicFilter;
import io.trino.spi.connector.EmptyPageSource;
import io.trino.spi.predicate.TupleDomain;
import io.trino.spi.type.Decimals;
import io.trino.spi.type.TypeSignature;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.storage.StoragePath;
import org.apache.parquet.column.ColumnDescriptor;
import org.apache.parquet.io.MessageColumnIO;
import org.apache.parquet.schema.MessageType;
Expand Down Expand Up @@ -90,6 +95,7 @@
import static io.trino.plugin.hudi.HudiSessionProperties.isParquetVectorizedDecodingEnabled;
import static io.trino.plugin.hudi.HudiSessionProperties.shouldUseParquetColumnNames;
import static io.trino.plugin.hudi.HudiUtil.getHudiFileFormat;
import static io.trino.plugin.hudi.HudiUtil.prependHudiMetaColumns;
import static io.trino.spi.predicate.Utils.nativeValueToBlock;
import static io.trino.spi.type.StandardTypes.BIGINT;
import static io.trino.spi.type.StandardTypes.BOOLEAN;
Expand Down Expand Up @@ -145,9 +151,57 @@ public ConnectorPageSource createPageSource(
DynamicFilter dynamicFilter)
{
HudiSplit split = (HudiSplit) connectorSplit;
String path = split.location();
HudiFileFormat hudiFileFormat = getHudiFileFormat(path);
if (!HudiFileFormat.PARQUET.equals(hudiFileFormat)) {
String dataFilePath = split.baseFile().isPresent()
? split.baseFile().get().getPath()
: split.logFiles().getFirst();
// Filter out metadata table splits
if (dataFilePath.contains(new StoragePath(
((HudiTableHandle) connectorTable).getBasePath()).toUri().getPath() + "/.hoodie/metadata")) {
return new EmptyPageSource();
}
if (split.logFiles().isEmpty()) {
HudiBaseFile baseFile = split.baseFile().get();
String path = baseFile.getPath();
HoodieFileFormat hudiFileFormat = getHudiFileFormat(path);
if (!HoodieFileFormat.PARQUET.equals(hudiFileFormat)) {
throw new TrinoException(HUDI_UNSUPPORTED_FILE_FORMAT, format("File format %s not supported", hudiFileFormat));
}

List<HiveColumnHandle> hiveColumns = columns.stream()
.map(HiveColumnHandle.class::cast)
.collect(toList());
// just send regular columns to create parquet page source
// for partition columns, separate blocks will be created
List<HiveColumnHandle> regularColumns = hiveColumns.stream()
.filter(columnHandle -> !columnHandle.isPartitionKey() && !columnHandle.isHidden())
.collect(Collectors.toList());
TrinoFileSystem fileSystem = fileSystemFactory.create(session);
TrinoInputFile inputFile = fileSystem.newInputFile(Location.of(path), baseFile.getFileSize());
ConnectorPageSource dataPageSource = createPageSource(
session,
regularColumns,
split,
inputFile,
dataSourceStats,
options.withSmallFileThreshold(getParquetSmallFileThreshold(session))
.withVectorizedDecodingEnabled(isParquetVectorizedDecodingEnabled(session)),
timeZone);

return new HudiReadOptimizedPageSource(
toPartitionName(split.partitionKeys()),
hiveColumns,
convertPartitionValues(hiveColumns, split.partitionKeys()), // create blocks for partition values
dataPageSource,
path,
baseFile.getFileSize(),
baseFile.getModificationTime());
}

HudiTableHandle hudiTableHandle = (HudiTableHandle) connectorTable;
HudiBaseFile baseFile = split.baseFile().get();
String path = baseFile.getPath();
HoodieFileFormat hudiFileFormat = getHudiFileFormat(path);
if (!HoodieFileFormat.PARQUET.equals(hudiFileFormat)) {
throw new TrinoException(HUDI_UNSUPPORTED_FILE_FORMAT, format("File format %s not supported", hudiFileFormat));
}

Expand All @@ -160,25 +214,27 @@ public ConnectorPageSource createPageSource(
.filter(columnHandle -> !columnHandle.isPartitionKey() && !columnHandle.isHidden())
.collect(Collectors.toList());
TrinoFileSystem fileSystem = fileSystemFactory.create(session);
TrinoInputFile inputFile = fileSystem.newInputFile(Location.of(path), split.fileSize());
TrinoInputFile inputFile = fileSystem.newInputFile(Location.of(path), baseFile.getFileSize());
ConnectorPageSource dataPageSource = createPageSource(
session,
regularColumns,
prependHudiMetaColumns(regularColumns),
split,
inputFile,
dataSourceStats,
options.withSmallFileThreshold(getParquetSmallFileThreshold(session))
.withVectorizedDecodingEnabled(isParquetVectorizedDecodingEnabled(session)),
timeZone);

return new HudiPageSource(
toPartitionName(split.partitionKeys()),
hiveColumns,
convertPartitionValues(hiveColumns, split.partitionKeys()), // create blocks for partition values
dataPageSource,
path,
split.fileSize(),
split.fileModifiedTime());
return new HudiSnapshotPageSource(
split.partitionKeys(),
new HudiTrinoStorage(fileSystemFactory.create(session), new TrinoStorageConfiguration()),
hudiTableHandle.getBasePath(),
split,
Optional.of(dataPageSource),
hiveColumns.stream()
.filter(columnHandle -> !columnHandle.isHidden())
.collect(Collectors.toList()),
prependHudiMetaColumns(regularColumns),
hudiTableHandle.getPreCombineField());
}

private static ConnectorPageSource createPageSource(
Expand All @@ -192,12 +248,13 @@ private static ConnectorPageSource createPageSource(
{
ParquetDataSource dataSource = null;
boolean useColumnNames = shouldUseParquetColumnNames(session);
String path = hudiSplit.location();
long start = hudiSplit.start();
long length = hudiSplit.length();
HudiBaseFile baseFile = hudiSplit.baseFile().get();
String path = baseFile.getPath();
long start = baseFile.getStart();
long length = baseFile.getLength();
try {
AggregatedMemoryContext memoryContext = newSimpleAggregatedMemoryContext();
dataSource = createDataSource(inputFile, OptionalLong.of(hudiSplit.fileSize()), options, memoryContext, dataSourceStats);
dataSource = createDataSource(inputFile, OptionalLong.of(baseFile.getFileSize()), options, memoryContext, dataSourceStats);
ParquetMetadata parquetMetadata = MetadataReader.readFooter(dataSource, Optional.empty());
FileMetadata fileMetaData = parquetMetadata.getFileMetaData();
MessageType fileSchema = fileMetaData.getSchema();
Expand Down
Loading
Loading