Skip to content

[WIP] Support data skipping based on column stats in Hudi connector #25184

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 5 commits into
base: master
Choose a base branch
from
+2,635 −2,873
Draft

[WIP] Support data skipping based on column stats in Hudi connector #25184

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 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
d7031f3
[WIP] Support data skipping based on column stats in Hudi connector
yihua Feb 28, 2025
30b1e6d
Add data skipping based on col stats to HudiBackgroundSplitLoader
yihua Mar 12, 2025
94fbe24
Fix point lookups in log reading in the metadata table
yihua Mar 14, 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
194 changes: 178 additions & 16 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,24 @@
<artifactId>avro</artifactId>
</dependency>

<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-common</artifactId>
<version>${dep.hudi.version}</version>
<exclusions>
<exclusion>
<groupId>*</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 +228,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 @@ -297,9 +316,9 @@
</dependency>

<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-client-common</artifactId>
<version>${dep.hudi.version}</version>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>${dep.hbase.version}</version>
<scope>test</scope>
<exclusions>
<exclusion>
Expand All @@ -310,9 +329,152 @@
</dependency>

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

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

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

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

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

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

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

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

<dependency>
<groupId>org.apache.hbase.thirdparty</groupId>
<artifactId>hbase-shaded-miscellaneous</artifactId>
<version>4.1.1</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>*</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>

<dependency>
<groupId>org.apache.hbase.thirdparty</groupId>
<artifactId>hbase-shaded-netty</artifactId>
<version>4.1.1</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>*</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>

<dependency>
<groupId>org.apache.hbase.thirdparty</groupId>
<artifactId>hbase-shaded-protobuf</artifactId>
<version>4.1.1</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>*</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>

<dependency>
<groupId>org.apache.hbase.thirdparty</groupId>
<artifactId>hbase-unsafe</artifactId>
<version>4.1.1</version>
<scope>test</scope>
<exclusions>
<exclusion>
Expand All @@ -324,7 +486,7 @@

<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-hadoop-common</artifactId>
<artifactId>hudi-client-common</artifactId>
<version>${dep.hudi.version}</version>
<scope>test</scope>
<exclusions>
Expand All @@ -337,7 +499,7 @@

<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-io</artifactId>
<artifactId>hudi-hadoop-common</artifactId>
<version>${dep.hudi.version}</version>
<scope>test</scope>
<exclusions>
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
Loading