You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by vh...@apache.org on 2025/08/27 11:48:38 UTC
(hudi) branch HUDI-9767__upstream-trino-changes created (now 194e58526b69)
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a change to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
at 194e58526b69 [ENG-30507] Incorrect query results for Merge-On-Read (RT) tables when column stats are enabled (#58)
This branch includes the following new commits:
new ac351f1b72dd [ENG-28391] Increase Default MaxOutstandingSplits and SplitLoaderParallelism (#45)
new 3c22563f0545 [HUDI-9525] Extend file system cache support to Hudi connector (#36)
new 0c33fd3b3d7e [ENG-29373] Fix split generation parallelism for a non-partitioned table (#48)
new d36d8cbf215c [MINOR] Revert changes made to non-hudi modules (#46)
new d34aaa3a1b96 [MINOR] Added optimizations for HudiColumnStatsIndexSupport (#47)
new e1fec7514308 [MINOR] Cleanup HudiSplitFactory to extend cache support (#50)
new 4f4bfc0fd883 [ENG-30914] Fix flakiness when testing cache correctness (#60)
new d13b7ec17661 [ENG-30888][Trino] Enable Metadata Table by default (#59)
new 5ef46667a6d3 [ENG-31327] [Trino] Fix flaky tests due to table stats computation lagging behind query execution (#62)
new 846d02942f30 [ENG-29378] Implement Metadata table based Partition listing (#49)
new d1c6b02c0c6e [ENG-30371] Fix Case Sensitivity Issues Between Table and Catalog Schemas (#57)
new b1494e866489 [ENG-31282][Trino] Workers should use latest commit time from table handle (#61)
new 194e58526b69 [ENG-30507] Incorrect query results for Merge-On-Read (RT) tables when column stats are enabled (#58)
The 13 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(hudi) 06/13: [MINOR] Cleanup HudiSplitFactory to extend cache support (#50)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit e1fec7514308d2233e5861b84d46fc69428faabf
Author: voonhous <vo...@gmail.com>
AuthorDate: Tue Aug 5 08:43:09 2025 +0800
[MINOR] Cleanup HudiSplitFactory to extend cache support (#50)
Co-authored-by: Y Ethan Guo <et...@gmail.com>
(cherry picked from commit 78ef5876aa761eea5642c7fa417855ff18b74cbe)
---
.../trino/plugin/hudi/split/HudiSplitFactory.java | 159 ++++++++++++++-------
1 file changed, 105 insertions(+), 54 deletions(-)
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiSplitFactory.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiSplitFactory.java
index 89e287a9cfe0..09dbf5bd5020 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiSplitFactory.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiSplitFactory.java
@@ -62,6 +62,13 @@ public class HudiSplitFactory
return createHudiSplits(hudiTableHandle, partitionKeys, fileSlice, commitTime, hudiSplitWeightProvider, targetSplitSize, cachingHostAddressProvider);
}
+ /**
+ * Creates a list of Hudi splits from a file slice.
+ * <p>
+ * For Copy-On-Write (COW) tables or read-optimized Merge-On-Read (MOR) tables, the base file may be broken into multiple smaller splits.
+ * <p>
+ * For regular MOR tables, a single split is created for the combination of the base file and its log files.
+ */
public static List<HudiSplit> createHudiSplits(
HudiTableHandle hudiTableHandle,
List<HivePartitionKey> partitionKeys,
@@ -72,69 +79,113 @@ public class HudiSplitFactory
CachingHostAddressProvider cachingHostAddressProvider)
{
if (fileSlice.isEmpty()) {
- throw new TrinoException(HUDI_FILESYSTEM_ERROR, format("Not a valid file slice: %s", fileSlice.toString()));
+ throw new TrinoException(HUDI_FILESYSTEM_ERROR, format("Not a valid file slice: %s", fileSlice));
}
- // Handle MERGE_ON_READ tables to be read in read_optimized mode
- // IMPORTANT: These tables will have a COPY_ON_WRITE table type due to how `HudiTableTypeUtils#fromInputFormat`
- if (fileSlice.getLogFiles().findAny().isEmpty()
- || hudiTableHandle.getTableType().equals(HoodieTableType.COPY_ON_WRITE)) {
- // Base file only
- checkArgument(fileSlice.getBaseFile().isPresent(),
- "Hudi base file must exist if there is no log file in the file slice");
- HoodieBaseFile baseFile = fileSlice.getBaseFile().get();
- long fileSize = baseFile.getFileSize();
- List<HostAddress> addresses = cachingHostAddressProvider.getHosts(
- baseFile.getPath(), ImmutableList.of());
-
- if (fileSize == 0) {
- return ImmutableList.of(new HudiSplit(
- HudiBaseFile.of(baseFile),
- Collections.emptyList(),
- commitTime,
- hudiTableHandle.getRegularPredicates(),
- partitionKeys,
- hudiSplitWeightProvider.calculateSplitWeight(fileSize),
- addresses));
- }
-
- ImmutableList.Builder<HudiSplit> splits = ImmutableList.builder();
- long targetSplitSizeInBytes = Math.max(targetSplitSize.toBytes(), baseFile.getPathInfo().getBlockSize());
-
- long bytesRemaining = fileSize;
- while (((double) bytesRemaining) / targetSplitSizeInBytes > SPLIT_SLOP) {
- splits.add(new HudiSplit(
- HudiBaseFile.of(baseFile, fileSize - bytesRemaining, targetSplitSizeInBytes),
- Collections.emptyList(),
- commitTime,
- hudiTableHandle.getRegularPredicates(),
- partitionKeys,
- hudiSplitWeightProvider.calculateSplitWeight(targetSplitSizeInBytes),
- addresses));
- bytesRemaining -= targetSplitSizeInBytes;
- }
- if (bytesRemaining > 0) {
- splits.add(new HudiSplit(
- HudiBaseFile.of(baseFile, fileSize - bytesRemaining, bytesRemaining),
- Collections.emptyList(),
- commitTime,
- hudiTableHandle.getRegularPredicates(),
- partitionKeys,
- hudiSplitWeightProvider.calculateSplitWeight(bytesRemaining),
- addresses));
- }
- return splits.build();
+ if (isCopyOnWriteOrReadOptimized(hudiTableHandle, fileSlice)) {
+ // Handle MERGE_ON_READ tables to be read in read_optimized mode
+ // IMPORTANT: These tables will have a COPY_ON_WRITE table type due to how `HudiTableTypeUtils#fromInputFormat`
+ return createSplitsForBaseFile(hudiTableHandle, partitionKeys, fileSlice, commitTime, hudiSplitWeightProvider, targetSplitSize, cachingHostAddressProvider);
+ }
+ return createSplitForMergeOnRead(hudiTableHandle, partitionKeys, fileSlice, commitTime, hudiSplitWeightProvider, cachingHostAddressProvider);
+ }
+
+ /**
+ * Checks if the file slice should be treated as a base-file-only read, which applies to COW tables and MOR tables in read-optimized mode.
+ */
+ private static boolean isCopyOnWriteOrReadOptimized(HudiTableHandle hudiTableHandle, FileSlice fileSlice)
+ {
+ return fileSlice.getLogFiles().findAny().isEmpty()
+ || hudiTableHandle.getTableType().equals(HoodieTableType.COPY_ON_WRITE);
+ }
+
+ /**
+ * Creates splits for a base file, potentially breaking it into multiple smaller splits.
+ */
+ private static List<HudiSplit> createSplitsForBaseFile(
+ HudiTableHandle hudiTableHandle,
+ List<HivePartitionKey> partitionKeys,
+ FileSlice fileSlice,
+ String commitTime,
+ HudiSplitWeightProvider hudiSplitWeightProvider,
+ DataSize targetSplitSize,
+ CachingHostAddressProvider cachingHostAddressProvider)
+ {
+ checkArgument(fileSlice.getBaseFile().isPresent(),
+ "Hudi base file must exist if there are no log files in the file slice");
+
+ HoodieBaseFile baseFile = fileSlice.getBaseFile().get();
+ long fileSize = baseFile.getFileSize();
+ List<HostAddress> addresses = cachingHostAddressProvider.getHosts(baseFile.getPath(), ImmutableList.of());
+
+ // If the file is empty, create a single split to represent it
+ if (fileSize == 0) {
+ HudiSplit split = new HudiSplit(
+ HudiBaseFile.of(baseFile),
+ Collections.emptyList(),
+ commitTime,
+ hudiTableHandle.getRegularPredicates(),
+ partitionKeys,
+ hudiSplitWeightProvider.calculateSplitWeight(0),
+ addresses);
+ return ImmutableList.of(split);
}
- // Base and log files
+ ImmutableList.Builder<HudiSplit> splits = ImmutableList.builder();
+ long targetSplitSizeInBytes = Math.max(targetSplitSize.toBytes(), baseFile.getPathInfo().getBlockSize());
+
+ long bytesRemaining = fileSize;
+ while (((double) bytesRemaining) / targetSplitSizeInBytes > SPLIT_SLOP) {
+ splits.add(new HudiSplit(
+ HudiBaseFile.of(baseFile, fileSize - bytesRemaining, targetSplitSizeInBytes),
+ Collections.emptyList(),
+ commitTime,
+ hudiTableHandle.getRegularPredicates(),
+ partitionKeys,
+ hudiSplitWeightProvider.calculateSplitWeight(targetSplitSizeInBytes),
+ addresses));
+ bytesRemaining -= targetSplitSizeInBytes;
+ }
+ if (bytesRemaining > 0) {
+ splits.add(new HudiSplit(
+ HudiBaseFile.of(baseFile, fileSize - bytesRemaining, bytesRemaining),
+ Collections.emptyList(),
+ commitTime,
+ hudiTableHandle.getRegularPredicates(),
+ partitionKeys,
+ hudiSplitWeightProvider.calculateSplitWeight(bytesRemaining),
+ addresses));
+ }
+
+ return splits.build();
+ }
+
+ /**
+ * Creates a single split for a Merge-On-Read file slice, including the base file and log files.
+ */
+ private static List<HudiSplit> createSplitForMergeOnRead(
+ HudiTableHandle hudiTableHandle,
+ List<HivePartitionKey> partitionKeys,
+ FileSlice fileSlice,
+ String commitTime,
+ HudiSplitWeightProvider hudiSplitWeightProvider,
+ CachingHostAddressProvider cachingHostAddressProvider)
+ {
+ // NOTE: Some file slices may not have base files
Option<HoodieBaseFile> baseFileOption = fileSlice.getBaseFile();
- return ImmutableList.of(new HudiSplit(
- baseFileOption.isPresent() ? HudiBaseFile.of(baseFileOption.get()) : null,
+ List<HostAddress> addresses = baseFileOption
+ .map(baseFile -> cachingHostAddressProvider.getHosts(baseFile.getPath(), ImmutableList.of()))
+ .orElse(ImmutableList.of());
+
+ HudiSplit split = new HudiSplit(
+ baseFileOption.map(HudiBaseFile::of).orElse(null),
fileSlice.getLogFiles().map(HudiLogFile::of).toList(),
commitTime,
hudiTableHandle.getRegularPredicates(),
partitionKeys,
hudiSplitWeightProvider.calculateSplitWeight(fileSlice.getTotalFileSize()),
- ImmutableList.of()));
+ addresses);
+
+ return ImmutableList.of(split);
}
}
(hudi) 04/13: [MINOR] Revert changes made to non-hudi modules (#46)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d36d8cbf215c01de7e29716e9d230a083f036424
Author: voonhous <vo...@gmail.com>
AuthorDate: Tue Jul 29 14:35:28 2025 +0800
[MINOR] Revert changes made to non-hudi modules (#46)
(cherry picked from commit 9ae51ccc9cb4db30f56e736356c2c688bc3558f4)
---
.../java/io/trino/plugin/hudi/HudiUtilTest.java | 151 +++++++++++
.../trino/plugin/hive/avro/AvroHiveFileUtils.java | 297 +++++++++++++++++++++
2 files changed, 448 insertions(+)
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/HudiUtilTest.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/HudiUtilTest.java
new file mode 100644
index 000000000000..523065afd852
--- /dev/null
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/HudiUtilTest.java
@@ -0,0 +1,151 @@
+/*
+ * 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 com.google.common.collect.ImmutableList;
+import io.trino.metastore.HiveType;
+import org.apache.avro.Schema;
+import org.junit.jupiter.api.Test;
+
+import java.io.UncheckedIOException;
+import java.util.List;
+
+import static io.trino.plugin.hudi.HudiUtil.constructSchema;
+import static io.trino.plugin.hudi.testing.TypeInfoHelper.INT_TYPE_INFO;
+import static io.trino.plugin.hudi.testing.TypeInfoHelper.LONG_TYPE_INFO;
+import static io.trino.plugin.hudi.testing.TypeInfoHelper.STRING_TYPE_INFO;
+import static io.trino.plugin.hudi.testing.TypeInfoHelper.listHiveType;
+import static io.trino.plugin.hudi.testing.TypeInfoHelper.mapHiveType;
+import static io.trino.plugin.hudi.testing.TypeInfoHelper.structHiveType;
+import static java.util.Collections.emptyList;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class HudiUtilTest
+{
+ @Test
+ void testConstructSchemaWithSimpleTypes()
+ {
+ List<String> columnNames = List.of("id", "price", "product_name", "is_available");
+ List<HiveType> columnTypes = List.of(
+ HiveType.HIVE_LONG,
+ HiveType.HIVE_DOUBLE,
+ HiveType.HIVE_STRING,
+ HiveType.HIVE_BOOLEAN);
+
+ Schema schema = constructSchema(columnNames, columnTypes);
+
+ assertThat(schema.getName()).isEqualTo("baseRecord");
+ assertThat(schema.getFields()).hasSize(4);
+
+ // Verify that each field is correctly typed and is a UNION with NULL
+ assertField(schema, "id", Schema.Type.LONG);
+ assertField(schema, "price", Schema.Type.DOUBLE);
+ assertField(schema, "product_name", Schema.Type.STRING);
+ assertField(schema, "is_available", Schema.Type.BOOLEAN);
+ }
+
+ @Test
+ void testConstructSchemaWithComplexTypes()
+ {
+ List<String> columnNames = List.of("tags", "ratings", "metadata");
+
+ // Create complex Hive types
+ HiveType tagsType = listHiveType(STRING_TYPE_INFO);
+ HiveType ratingsType = mapHiveType(STRING_TYPE_INFO, INT_TYPE_INFO);
+ HiveType metadataType = structHiveType(
+ ImmutableList.of("author", "version"),
+ ImmutableList.of(STRING_TYPE_INFO, LONG_TYPE_INFO));
+
+ List<HiveType> columnTypes = List.of(tagsType, ratingsType, metadataType);
+
+ Schema schema = constructSchema(columnNames, columnTypes);
+
+ assertThat(schema.getName()).isEqualTo("baseRecord");
+ assertThat(schema.getFields()).hasSize(3);
+
+ // Verify array field
+ Schema tagsFieldSchema = assertAndGetNonNullSchema(schema.getField("tags"));
+ assertThat(tagsFieldSchema.getType()).isEqualTo(Schema.Type.ARRAY);
+ assertNonNullUnion(tagsFieldSchema.getElementType(), Schema.Type.STRING);
+
+ // Verify map field
+ Schema ratingsFieldSchema = assertAndGetNonNullSchema(schema.getField("ratings"));
+ assertThat(ratingsFieldSchema.getType()).isEqualTo(Schema.Type.MAP);
+ assertNonNullUnion(ratingsFieldSchema.getValueType(), Schema.Type.INT);
+
+ // Verify struct (record) field
+ Schema metadataFieldSchema = assertAndGetNonNullSchema(schema.getField("metadata"));
+ assertThat(metadataFieldSchema.getType()).isEqualTo(Schema.Type.RECORD);
+ assertThat(metadataFieldSchema.getFields()).hasSize(2);
+ assertField(metadataFieldSchema, "author", Schema.Type.STRING);
+ assertField(metadataFieldSchema, "version", Schema.Type.LONG);
+ }
+
+ @Test
+ void testConstructSchemaWithEmptyColumns()
+ {
+ assertThatThrownBy(() -> constructSchema(emptyList(), emptyList()))
+ .isInstanceOf(UncheckedIOException.class)
+ .hasMessageContaining("Failed to construct Avro schema");
+ }
+
+ @Test
+ void testConstructSchemaWithMismatchedColumns()
+ {
+ List<String> columnNames = List.of("col1", "col2");
+ List<HiveType> columnTypes = List.of(HiveType.HIVE_INT); // Only one type
+
+ assertThatThrownBy(() -> constructSchema(columnNames, columnTypes))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Number of column name and column type differs");
+ }
+
+ /**
+ * Helper method to assert a field exists and is a UNION of [NULL, expectedType].
+ */
+ private void assertField(Schema recordSchema, String fieldName, Schema.Type expectedType)
+ {
+ Schema fieldSchema = assertAndGetNonNullSchema(recordSchema.getField(fieldName));
+ assertThat(fieldSchema.getType()).isEqualTo(expectedType);
+ }
+
+ /**
+ * Helper method to assert that a field's schema is a UNION of [NULL, other_type]
+ * and returns the non-null schema part for further assertions.
+ */
+ private Schema assertAndGetNonNullSchema(Schema.Field field)
+ {
+ assertThat(field).isNotNull();
+ return assertNonNullUnion(field.schema(), null);
+ }
+
+ /**
+ * Asserts a schema is a UNION of [NULL, other_type] and returns the non-null part.
+ * Optionally checks the type of the non-null part.
+ */
+ private Schema assertNonNullUnion(Schema unionSchema, Schema.Type expectedType)
+ {
+ assertThat(unionSchema.isUnion()).isTrue();
+ assertThat(unionSchema.getTypes()).hasSize(2);
+ // The Trino utility always places NULL first in the union
+ assertThat(unionSchema.getTypes().get(0).getType()).isEqualTo(Schema.Type.NULL);
+
+ Schema nonNullSchema = unionSchema.getTypes().get(1);
+ if (expectedType != null) {
+ assertThat(nonNullSchema.getType()).isEqualTo(expectedType);
+ }
+ return nonNullSchema;
+ }
+}
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/avro/AvroHiveFileUtils.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/avro/AvroHiveFileUtils.java
new file mode 100644
index 000000000000..37d897590576
--- /dev/null
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/avro/AvroHiveFileUtils.java
@@ -0,0 +1,297 @@
+/*
+ * 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.hive.avro;
+
+import com.google.common.base.Splitter;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoInputFile;
+import io.trino.filesystem.TrinoInputStream;
+import io.trino.hive.formats.avro.NativeLogicalTypesAvroTypeManager;
+import io.trino.metastore.HiveType;
+import io.trino.metastore.type.CharTypeInfo;
+import io.trino.metastore.type.DecimalTypeInfo;
+import io.trino.metastore.type.ListTypeInfo;
+import io.trino.metastore.type.MapTypeInfo;
+import io.trino.metastore.type.PrimitiveCategory;
+import io.trino.metastore.type.PrimitiveTypeInfo;
+import io.trino.metastore.type.StructTypeInfo;
+import io.trino.metastore.type.TypeInfo;
+import io.trino.metastore.type.UnionTypeInfo;
+import io.trino.metastore.type.VarcharTypeInfo;
+import org.apache.avro.LogicalType;
+import org.apache.avro.LogicalTypes;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
+import static io.trino.hive.formats.avro.AvroHiveConstants.CHAR_TYPE_LOGICAL_NAME;
+import static io.trino.hive.formats.avro.AvroHiveConstants.SCHEMA_DOC;
+import static io.trino.hive.formats.avro.AvroHiveConstants.SCHEMA_LITERAL;
+import static io.trino.hive.formats.avro.AvroHiveConstants.SCHEMA_NAME;
+import static io.trino.hive.formats.avro.AvroHiveConstants.SCHEMA_NAMESPACE;
+import static io.trino.hive.formats.avro.AvroHiveConstants.SCHEMA_NONE;
+import static io.trino.hive.formats.avro.AvroHiveConstants.SCHEMA_URL;
+import static io.trino.hive.formats.avro.AvroHiveConstants.TABLE_NAME;
+import static io.trino.hive.formats.avro.AvroHiveConstants.VARCHAR_AND_CHAR_LOGICAL_TYPE_LENGTH_PROP;
+import static io.trino.hive.formats.avro.AvroHiveConstants.VARCHAR_TYPE_LOGICAL_NAME;
+import static io.trino.metastore.Table.TABLE_COMMENT;
+import static io.trino.plugin.hive.util.HiveUtil.getColumnNames;
+import static io.trino.plugin.hive.util.HiveUtil.getColumnTypes;
+import static io.trino.plugin.hive.util.SerdeConstants.LIST_COLUMN_COMMENTS;
+import static java.util.Collections.emptyList;
+import static java.util.function.Predicate.not;
+import static java.util.function.UnaryOperator.identity;
+
+public final class AvroHiveFileUtils
+{
+ private final AtomicInteger recordNameSuffix = new AtomicInteger(0);
+
+ private AvroHiveFileUtils() {}
+
+ // Lifted and shifted from org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils.determineSchemaOrThrowException
+ public static Schema determineSchemaOrThrowException(TrinoFileSystem fileSystem, Map<String, String> properties)
+ throws IOException
+ {
+ // Try pull schema from literal table property
+ String schemaString = properties.getOrDefault(SCHEMA_LITERAL, "");
+ if (!schemaString.isBlank() && !schemaString.equals(SCHEMA_NONE)) {
+ return getSchemaParser().parse(schemaString);
+ }
+
+ // Try pull schema directly from URL
+ String schemaURL = properties.getOrDefault(SCHEMA_URL, "");
+ if (!schemaURL.isBlank()) {
+ TrinoInputFile schemaFile = fileSystem.newInputFile(Location.of(schemaURL));
+ if (!schemaFile.exists()) {
+ throw new IOException("No avro schema file not found at " + schemaURL);
+ }
+ try (TrinoInputStream inputStream = schemaFile.newStream()) {
+ return getSchemaParser().parse(inputStream);
+ }
+ catch (IOException e) {
+ throw new IOException("Unable to read avro schema file from given path: " + schemaURL, e);
+ }
+ }
+ return getSchemaFromProperties(properties);
+ }
+
+ private static Schema getSchemaFromProperties(Map<String, String> schema)
+ throws IOException
+ {
+ List<String> columnNames = getColumnNames(schema);
+ List<HiveType> columnTypes = getColumnTypes(schema);
+ if (columnNames.isEmpty() || columnTypes.isEmpty()) {
+ throw new IOException("Unable to parse column names or column types from schema to create Avro Schema");
+ }
+ if (columnNames.size() != columnTypes.size()) {
+ throw new IllegalArgumentException("Avro Schema initialization failed. Number of column name and column type differs. columnNames = %s, columnTypes = %s".formatted(columnNames, columnTypes));
+ }
+ List<String> columnComments = Optional.ofNullable(schema.get(LIST_COLUMN_COMMENTS))
+ .filter(not(String::isBlank))
+ .map(Splitter.on('\0')::splitToList)
+ .orElse(emptyList());
+
+ String tableName = schema.get(TABLE_NAME);
+ String tableComment = schema.get(TABLE_COMMENT);
+
+ return constructSchemaFromParts(
+ columnNames,
+ columnTypes,
+ columnComments,
+ Optional.ofNullable(schema.get(SCHEMA_NAMESPACE)),
+ Optional.ofNullable(schema.getOrDefault(SCHEMA_NAME, tableName)),
+ Optional.ofNullable(schema.getOrDefault(SCHEMA_DOC, tableComment)));
+ }
+
+ private static Schema constructSchemaFromParts(List<String> columnNames, List<HiveType> columnTypes,
+ List<String> columnComments, Optional<String> namespace, Optional<String> name, Optional<String> doc)
+ {
+ // create instance of this class to keep nested record naming consistent for any given inputs
+ AvroHiveFileUtils recordIncrementingUtil = new AvroHiveFileUtils();
+ SchemaBuilder.RecordBuilder<Schema> schemaBuilder = SchemaBuilder.record(name.orElse("baseRecord"));
+ namespace.ifPresent(schemaBuilder::namespace);
+ doc.ifPresent(schemaBuilder::doc);
+ SchemaBuilder.FieldAssembler<Schema> fieldBuilder = schemaBuilder.fields();
+
+ for (int i = 0; i < columnNames.size(); ++i) {
+ String comment = columnComments.size() > i ? columnComments.get(i) : null;
+ Schema fieldSchema = recordIncrementingUtil.avroSchemaForHiveType(columnTypes.get(i));
+ fieldBuilder = fieldBuilder
+ .name(columnNames.get(i))
+ .doc(comment)
+ .type(fieldSchema)
+ .withDefault(null);
+ }
+ return fieldBuilder.endRecord();
+ }
+
+ private Schema avroSchemaForHiveType(HiveType hiveType)
+ {
+ Schema schema = switch (hiveType.getCategory()) {
+ case PRIMITIVE -> createAvroPrimitive(hiveType);
+ case LIST -> {
+ ListTypeInfo listTypeInfo = (ListTypeInfo) hiveType.getTypeInfo();
+ yield Schema.createArray(avroSchemaForHiveType(HiveType.fromTypeInfo(listTypeInfo.getListElementTypeInfo())));
+ }
+ case MAP -> {
+ MapTypeInfo mapTypeInfo = ((MapTypeInfo) hiveType.getTypeInfo());
+ TypeInfo keyTypeInfo = mapTypeInfo.getMapKeyTypeInfo();
+ if (!(keyTypeInfo instanceof PrimitiveTypeInfo primitiveKeyTypeInfo) ||
+ primitiveKeyTypeInfo.getPrimitiveCategory() != PrimitiveCategory.STRING) {
+ throw new UnsupportedOperationException("Key of Map must be a String");
+ }
+ TypeInfo valueTypeInfo = mapTypeInfo.getMapValueTypeInfo();
+ yield Schema.createMap(avroSchemaForHiveType(HiveType.fromTypeInfo(valueTypeInfo)));
+ }
+ case STRUCT -> createAvroRecord(hiveType);
+ case UNION -> {
+ List<Schema> childSchemas = new ArrayList<>();
+ for (TypeInfo childTypeInfo : ((UnionTypeInfo) hiveType.getTypeInfo()).getAllUnionObjectTypeInfos()) {
+ final Schema childSchema = avroSchemaForHiveType(HiveType.fromTypeInfo(childTypeInfo));
+ if (childSchema.getType() == Schema.Type.UNION) {
+ childSchemas.addAll(childSchema.getTypes());
+ }
+ else {
+ childSchemas.add(childSchema);
+ }
+ }
+ yield Schema.createUnion(removeDuplicateNullSchemas(childSchemas));
+ }
+ };
+
+ return wrapInUnionWithNull(schema);
+ }
+
+ private static Schema createAvroPrimitive(HiveType hiveType)
+ {
+ if (!(hiveType.getTypeInfo() instanceof PrimitiveTypeInfo primitiveTypeInfo)) {
+ throw new IllegalStateException("HiveType in primitive category must have PrimitiveTypeInfo");
+ }
+ return switch (primitiveTypeInfo.getPrimitiveCategory()) {
+ case STRING -> Schema.create(Schema.Type.STRING);
+ case CHAR -> {
+ Schema charSchema = SchemaBuilder.builder().type(Schema.create(Schema.Type.STRING));
+ charSchema.addProp(LogicalType.LOGICAL_TYPE_PROP, CHAR_TYPE_LOGICAL_NAME);
+ charSchema.addProp(VARCHAR_AND_CHAR_LOGICAL_TYPE_LENGTH_PROP, ((CharTypeInfo) hiveType.getTypeInfo()).getLength());
+ yield charSchema;
+ }
+ case VARCHAR -> {
+ Schema varcharSchema = SchemaBuilder.builder().type(Schema.create(Schema.Type.STRING));
+ varcharSchema.addProp(LogicalType.LOGICAL_TYPE_PROP, VARCHAR_TYPE_LOGICAL_NAME);
+ varcharSchema.addProp(VARCHAR_AND_CHAR_LOGICAL_TYPE_LENGTH_PROP, ((VarcharTypeInfo) hiveType.getTypeInfo()).getLength());
+ yield varcharSchema;
+ }
+ case BINARY -> Schema.create(Schema.Type.BYTES);
+ case BYTE, SHORT, INT -> Schema.create(Schema.Type.INT);
+ case LONG -> Schema.create(Schema.Type.LONG);
+ case FLOAT -> Schema.create(Schema.Type.FLOAT);
+ case DOUBLE -> Schema.create(Schema.Type.DOUBLE);
+ case BOOLEAN -> Schema.create(Schema.Type.BOOLEAN);
+ case DECIMAL -> {
+ DecimalTypeInfo decimalTypeInfo = (DecimalTypeInfo) hiveType.getTypeInfo();
+ LogicalTypes.Decimal decimalLogicalType = LogicalTypes.decimal(decimalTypeInfo.precision(), decimalTypeInfo.scale());
+ yield decimalLogicalType.addToSchema(Schema.create(Schema.Type.BYTES));
+ }
+ case DATE -> NativeLogicalTypesAvroTypeManager.DATE_SCHEMA;
+ case TIMESTAMP -> NativeLogicalTypesAvroTypeManager.TIMESTAMP_MILLIS_SCHEMA;
+ case VOID -> Schema.create(Schema.Type.NULL);
+ default -> throw new UnsupportedOperationException(hiveType + " is not supported.");
+ };
+ }
+
+ private Schema createAvroRecord(HiveType hiveType)
+ {
+ if (!(hiveType.getTypeInfo() instanceof StructTypeInfo structTypeInfo)) {
+ throw new IllegalStateException("HiveType type info must be Struct Type info to make Avro Record");
+ }
+
+ final List<String> allStructFieldNames =
+ structTypeInfo.getAllStructFieldNames();
+ final List<TypeInfo> allStructFieldTypeInfo =
+ structTypeInfo.getAllStructFieldTypeInfos();
+ if (allStructFieldNames.size() != allStructFieldTypeInfo.size()) {
+ throw new IllegalArgumentException("Failed to generate avro schema from hive schema. " +
+ "name and column type differs. names = " + allStructFieldNames + ", types = " +
+ allStructFieldTypeInfo);
+ }
+
+ SchemaBuilder.FieldAssembler<Schema> fieldAssembler = SchemaBuilder
+ .record("record_" + recordNameSuffix.getAndIncrement())
+ .doc(structTypeInfo.toString())
+ .fields();
+
+ for (int i = 0; i < allStructFieldNames.size(); ++i) {
+ final TypeInfo childTypeInfo = allStructFieldTypeInfo.get(i);
+ final Schema fieldSchema = avroSchemaForHiveType(HiveType.fromTypeInfo(childTypeInfo));
+ fieldAssembler = fieldAssembler
+ .name(allStructFieldNames.get(i))
+ .doc(childTypeInfo.toString())
+ .type(fieldSchema)
+ .withDefault(null);
+ }
+ return fieldAssembler.endRecord();
+ }
+
+ public static Schema wrapInUnionWithNull(Schema schema)
+ {
+ return switch (schema.getType()) {
+ case NULL -> schema;
+ case UNION -> Schema.createUnion(removeDuplicateNullSchemas(schema.getTypes()));
+ default -> Schema.createUnion(Arrays.asList(Schema.create(Schema.Type.NULL), schema));
+ };
+ }
+
+ private static List<Schema> removeDuplicateNullSchemas(List<Schema> childSchemas)
+ {
+ List<Schema> prunedSchemas = new ArrayList<>();
+ boolean isNullPresent = false;
+ for (Schema schema : childSchemas) {
+ if (schema.getType() == Schema.Type.NULL) {
+ isNullPresent = true;
+ }
+ else {
+ prunedSchemas.add(schema);
+ }
+ }
+ if (isNullPresent) {
+ prunedSchemas.add(0, Schema.create(Schema.Type.NULL));
+ }
+
+ return prunedSchemas;
+ }
+
+ static Map<String, String> getCanonicalToGivenFieldName(Schema schema)
+ {
+ // Lower case top level fields to allow for manually set avro schema (passed in via avro_schema_literal or avro_schema_url) to have uppercase field names
+ return schema.getFields().stream()
+ .map(Schema.Field::name)
+ .collect(toImmutableMap(fieldName -> fieldName.toLowerCase(Locale.ENGLISH), identity()));
+ }
+
+ private static Schema.Parser getSchemaParser()
+ {
+ // HIVE-24797: Disable validate default values when parsing Avro schemas.
+ return new Schema.Parser().setValidateDefaults(false);
+ }
+}
(hudi) 02/13: [HUDI-9525] Extend file system cache support to Hudi connector (#36)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 3c22563f05451b17328cbd2d721906a3a409ebb9
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Wed Jul 2 09:38:29 2025 +0800
[HUDI-9525] Extend file system cache support to Hudi connector (#36)
(cherry picked from commit cc01d6c9be7f79dc8aecf5429bc35b8b689e0b86)
---
hudi-trino-plugin/pom.xml | 13 ++
.../main/java/io/trino/plugin/hudi/HudiConfig.java | 14 ++
.../io/trino/plugin/hudi/HudiConnectorFactory.java | 27 +++-
.../main/java/io/trino/plugin/hudi/HudiModule.java | 5 +
.../main/java/io/trino/plugin/hudi/HudiSplit.java | 23 ++++
.../io/trino/plugin/hudi/HudiSplitManager.java | 9 +-
.../java/io/trino/plugin/hudi/HudiSplitSource.java | 5 +-
.../hudi/split/HudiBackgroundSplitLoader.java | 4 +-
.../trino/plugin/hudi/split/HudiSplitFactory.java | 26 +++-
.../java/io/trino/plugin/hudi/HudiQueryRunner.java | 7 +
.../hudi/TestHudiAlluxioCacheFileOperations.java | 149 +++++++++++++++++++++
.../hudi/TestHudiAlluxioCachingSmokeTest.java | 57 ++++++++
.../java/io/trino/plugin/hudi/TestHudiConfig.java | 7 +-
.../hudi/TestHudiMemoryCacheFileOperations.java | 141 +++++++++++++++++++
.../plugin/hudi/TestHudiNoCacheFileOperations.java | 141 +++++++++++++++++++
.../io/trino/plugin/hudi/TestHudiSmokeTest.java | 7 +
.../plugin/hudi/split/TestHudiSplitFactory.java | 4 +-
.../trino/plugin/hudi/util/FileOperationUtils.java | 85 ++++++++++++
.../plugin/hudi/cache/HudiCacheKeyProvider.java | 38 ++++++
19 files changed, 747 insertions(+), 15 deletions(-)
diff --git a/hudi-trino-plugin/pom.xml b/hudi-trino-plugin/pom.xml
index 7ce3a658af8c..08a8ebfdee32 100644
--- a/hudi-trino-plugin/pom.xml
+++ b/hudi-trino-plugin/pom.xml
@@ -229,6 +229,12 @@
<scope>runtime</scope>
</dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-sdk-trace</artifactId>
+ <scope>runtime</scope>
+ </dependency>
+
<dependency>
<groupId>io.trino</groupId>
<artifactId>trino-hive-formats</artifactId>
@@ -259,6 +265,13 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>io.trino</groupId>
+ <artifactId>trino-filesystem</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>io.trino</groupId>
<artifactId>trino-hdfs</artifactId>
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
index 4a595f5044fe..3177f7c4cd23 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
@@ -64,6 +64,7 @@ public class HudiConfig
private Duration columnStatsWaitTimeout = new Duration(1, SECONDS);
private Duration recordIndexWaitTimeout = new Duration(2, SECONDS);
private Duration secondaryIndexWaitTimeout = new Duration(2, SECONDS);
+ private boolean metadataCacheEnabled = true;
public List<String> getColumnsToHide()
{
@@ -407,4 +408,17 @@ public class HudiConfig
{
return secondaryIndexWaitTimeout;
}
+
+ public boolean isMetadataCacheEnabled()
+ {
+ return metadataCacheEnabled;
+ }
+
+ @Config("hudi.metadata.cache.enabled")
+ @ConfigDescription("Enables in-memory caching of Hudi metadata files on coordinator if fs.cache.enabled is set to false")
+ public HudiConfig setMetadataCacheEnabled(boolean metadataCacheEnabled)
+ {
+ this.metadataCacheEnabled = metadataCacheEnabled;
+ return this;
+ }
}
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConnectorFactory.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConnectorFactory.java
index f7dd96d76799..0db65192baf8 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConnectorFactory.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConnectorFactory.java
@@ -14,11 +14,13 @@
package io.trino.plugin.hudi;
import com.google.common.collect.ImmutableSet;
+import com.google.inject.Binder;
import com.google.inject.Injector;
import com.google.inject.Key;
import com.google.inject.Module;
import io.airlift.bootstrap.Bootstrap;
import io.airlift.bootstrap.LifeCycleManager;
+import io.airlift.configuration.AbstractConfigurationAwareModule;
import io.airlift.json.JsonModule;
import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.api.trace.Tracer;
@@ -48,6 +50,7 @@ import java.util.Set;
import static com.google.inject.util.Modules.EMPTY_MODULE;
import static io.trino.plugin.base.Versions.checkStrictSpiVersionMatch;
+import static java.util.Objects.requireNonNull;
public class HudiConnectorFactory
implements ConnectorFactory
@@ -78,7 +81,7 @@ public class HudiConnectorFactory
new JsonModule(),
new HudiModule(),
new HiveMetastoreModule(Optional.empty()),
- new FileSystemModule(catalogName, context.getNodeManager(), context.getOpenTelemetry(), false),
+ new HudiFileSystemModule(catalogName, context),
new MBeanServerModule(),
module.orElse(EMPTY_MODULE),
binder -> {
@@ -115,4 +118,26 @@ public class HudiConnectorFactory
hudiTableProperties.getTableProperties());
}
}
+
+ private static class HudiFileSystemModule
+ extends AbstractConfigurationAwareModule
+ {
+ private final String catalogName;
+ private final NodeManager nodeManager;
+ private final OpenTelemetry openTelemetry;
+
+ public HudiFileSystemModule(String catalogName, ConnectorContext context)
+ {
+ this.catalogName = requireNonNull(catalogName, "catalogName is null");
+ this.nodeManager = context.getNodeManager();
+ this.openTelemetry = context.getOpenTelemetry();
+ }
+
+ @Override
+ protected void setup(Binder binder)
+ {
+ boolean metadataCacheEnabled = buildConfigObject(HudiConfig.class).isMetadataCacheEnabled();
+ install(new FileSystemModule(catalogName, nodeManager, openTelemetry, metadataCacheEnabled));
+ }
+ }
}
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiModule.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiModule.java
index 929cd834c844..bd3c1923ebad 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiModule.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiModule.java
@@ -19,6 +19,7 @@ import com.google.inject.Module;
import com.google.inject.Provides;
import com.google.inject.Scopes;
import com.google.inject.Singleton;
+import io.trino.filesystem.cache.CacheKeyProvider;
import io.trino.metastore.HiveMetastore;
import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
import io.trino.plugin.base.session.SessionPropertiesProvider;
@@ -28,6 +29,7 @@ import io.trino.plugin.hive.HiveTransactionHandle;
import io.trino.plugin.hive.metastore.thrift.TranslateHiveViews;
import io.trino.plugin.hive.parquet.ParquetReaderConfig;
import io.trino.plugin.hive.parquet.ParquetWriterConfig;
+import io.trino.plugin.hudi.cache.HudiCacheKeyProvider;
import io.trino.plugin.hudi.stats.ForHudiTableStatistics;
import io.trino.spi.connector.ConnectorNodePartitioningProvider;
import io.trino.spi.connector.ConnectorPageSourceProvider;
@@ -39,6 +41,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.function.BiFunction;
import static com.google.inject.multibindings.Multibinder.newSetBinder;
+import static com.google.inject.multibindings.OptionalBinder.newOptionalBinder;
import static io.airlift.concurrent.Threads.daemonThreadsNamed;
import static io.airlift.configuration.ConfigBinder.configBinder;
import static io.trino.plugin.base.ClosingBinder.closingBinder;
@@ -74,6 +77,8 @@ public class HudiModule
binder.bind(FileFormatDataSourceStats.class).in(Scopes.SINGLETON);
newExporter(binder).export(FileFormatDataSourceStats.class).withGeneratedName();
+ newOptionalBinder(binder, CacheKeyProvider.class).setBinding().to(HudiCacheKeyProvider.class).in(Scopes.SINGLETON);
+
closingBinder(binder).registerExecutor(Key.get(ExecutorService.class, ForHudiTableStatistics.class));
closingBinder(binder).registerExecutor(Key.get(ExecutorService.class, ForHudiSplitManager.class));
closingBinder(binder).registerExecutor(Key.get(ScheduledExecutorService.class, ForHudiSplitSource.class));
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplit.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplit.java
index c0c15eada837..e2de10ca98a4 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplit.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplit.java
@@ -14,6 +14,7 @@
package io.trino.plugin.hudi;
import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
@@ -21,6 +22,7 @@ import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hive.HivePartitionKey;
import io.trino.plugin.hudi.file.HudiBaseFile;
import io.trino.plugin.hudi.file.HudiLogFile;
+import io.trino.spi.HostAddress;
import io.trino.spi.SplitWeight;
import io.trino.spi.connector.ConnectorSplit;
import io.trino.spi.predicate.TupleDomain;
@@ -46,6 +48,7 @@ public class HudiSplit
private final TupleDomain<HiveColumnHandle> predicate;
private final List<HivePartitionKey> partitionKeys;
private final SplitWeight splitWeight;
+ private final List<HostAddress> cachingHostAddresses;
@JsonCreator
public HudiSplit(
@@ -55,6 +58,18 @@ public class HudiSplit
@JsonProperty("predicate") TupleDomain<HiveColumnHandle> predicate,
@JsonProperty("partitionKeys") List<HivePartitionKey> partitionKeys,
@JsonProperty("splitWeight") SplitWeight splitWeight)
+ {
+ this(baseFile, logFiles, commitTime, predicate, partitionKeys, splitWeight, ImmutableList.of());
+ }
+
+ public HudiSplit(
+ HudiBaseFile baseFile,
+ List<HudiLogFile> logFiles,
+ String commitTime,
+ TupleDomain<HiveColumnHandle> predicate,
+ List<HivePartitionKey> partitionKeys,
+ SplitWeight splitWeight,
+ List<HostAddress> cachingHostAddresses)
{
this.baseFile = Optional.ofNullable(baseFile);
this.logFiles = requireNonNull(logFiles, "logFiles is null");
@@ -62,6 +77,7 @@ public class HudiSplit
this.predicate = requireNonNull(predicate, "predicate is null");
this.partitionKeys = ImmutableList.copyOf(requireNonNull(partitionKeys, "partitionKeys is null"));
this.splitWeight = requireNonNull(splitWeight, "splitWeight is null");
+ this.cachingHostAddresses = requireNonNull(cachingHostAddresses, "cachingHostAddresses is null");
}
@Override
@@ -111,6 +127,13 @@ public class HudiSplit
return partitionKeys;
}
+ @JsonIgnore
+ @Override
+ public List<HostAddress> getAddresses()
+ {
+ return cachingHostAddresses;
+ }
+
@Override
public long getRetainedSizeInBytes()
{
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitManager.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitManager.java
index 8f774e3f5f15..79f03ccd07b2 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitManager.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitManager.java
@@ -17,6 +17,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.inject.Inject;
import io.airlift.log.Logger;
+import io.trino.filesystem.cache.CachingHostAddressProvider;
import io.trino.metastore.HiveMetastore;
import io.trino.metastore.Partition;
import io.trino.metastore.StorageFormat;
@@ -60,16 +61,19 @@ public class HudiSplitManager
private final BiFunction<ConnectorIdentity, HiveTransactionHandle, HiveMetastore> metastoreProvider;
private final ExecutorService executor;
private final ScheduledExecutorService splitLoaderExecutorService;
+ private final CachingHostAddressProvider cachingHostAddressProvider;
@Inject
public HudiSplitManager(
BiFunction<ConnectorIdentity, HiveTransactionHandle, HiveMetastore> metastoreProvider,
@ForHudiSplitManager ExecutorService executor,
- @ForHudiSplitSource ScheduledExecutorService splitLoaderExecutorService)
+ @ForHudiSplitSource ScheduledExecutorService splitLoaderExecutorService,
+ CachingHostAddressProvider cachingHostAddressProvider)
{
this.metastoreProvider = requireNonNull(metastoreProvider, "metastoreProvider is null");
this.executor = requireNonNull(executor, "executor is null");
this.splitLoaderExecutorService = requireNonNull(splitLoaderExecutorService, "splitLoaderExecutorService is null");
+ this.cachingHostAddressProvider = requireNonNull(cachingHostAddressProvider, "cachingHostAddressProvider is null");
}
@Override
@@ -99,7 +103,8 @@ public class HudiSplitManager
getMaxOutstandingSplits(session),
lazyAllPartitions,
dynamicFilter,
- getDynamicFilteringWaitTimeout(session));
+ getDynamicFilteringWaitTimeout(session),
+ cachingHostAddressProvider);
return new ClassLoaderSafeConnectorSplitSource(splitSource, HudiSplitManager.class.getClassLoader());
}
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
index f0286aea7e95..bfa356fc5c27 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
@@ -19,6 +19,7 @@ import com.google.common.util.concurrent.Futures;
import io.airlift.concurrent.BoundedExecutor;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
+import io.trino.filesystem.cache.CachingHostAddressProvider;
import io.trino.metastore.Partition;
import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hive.HivePartitionKey;
@@ -89,7 +90,8 @@ public class HudiSplitSource
int maxOutstandingSplits,
Lazy<Map<String, Partition>> lazyPartitions,
DynamicFilter dynamicFilter,
- Duration dynamicFilteringWaitTimeoutMillis)
+ Duration dynamicFilteringWaitTimeoutMillis,
+ CachingHostAddressProvider cachingHostAddressProvider)
{
boolean enableMetadataTable = isHudiMetadataTableEnabled(session);
Lazy<HoodieTableMetadata> lazyTableMetadata = Lazy.lazily(() -> {
@@ -121,6 +123,7 @@ public class HudiSplitSource
lazyPartitions,
enableMetadataTable,
lazyTableMetadata,
+ cachingHostAddressProvider,
throwable -> {
trinoException.compareAndSet(null, new TrinoException(HUDI_CANNOT_OPEN_SPLIT,
"Failed to generate splits for " + tableHandle.getSchemaTableName(), throwable));
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
index 56c89ea92634..df5436d1579b 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
@@ -16,6 +16,7 @@ package io.trino.plugin.hudi.split;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.log.Logger;
+import io.trino.filesystem.cache.CachingHostAddressProvider;
import io.trino.metastore.Partition;
import io.trino.plugin.hive.util.AsyncQueue;
import io.trino.plugin.hudi.HudiTableHandle;
@@ -72,6 +73,7 @@ public class HudiBackgroundSplitLoader
Lazy<Map<String, Partition>> lazyPartitionMap,
boolean enableMetadataTable,
Lazy<HoodieTableMetadata> lazyTableMetadata,
+ CachingHostAddressProvider cachingHostAddressProvider,
Consumer<Throwable> errorListener)
{
this.tableHandle = requireNonNull(tableHandle, "tableHandle is null");
@@ -79,7 +81,7 @@ public class HudiBackgroundSplitLoader
this.asyncQueue = requireNonNull(asyncQueue, "asyncQueue is null");
this.splitGeneratorExecutor = requireNonNull(splitGeneratorExecutor, "splitGeneratorExecutorService is null");
this.splitGeneratorNumThreads = getSplitGeneratorParallelism(session);
- this.hudiSplitFactory = new HudiSplitFactory(tableHandle, hudiSplitWeightProvider, getTargetSplitSize(session));
+ this.hudiSplitFactory = new HudiSplitFactory(tableHandle, hudiSplitWeightProvider, getTargetSplitSize(session), cachingHostAddressProvider);
this.lazyPartitions = Lazy.lazily(() -> requireNonNull(lazyPartitionMap, "partitions is null").get().keySet().stream().toList());
this.enableMetadataTable = enableMetadataTable;
this.errorListener = requireNonNull(errorListener, "errorListener is null");
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiSplitFactory.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiSplitFactory.java
index 872f56cb352d..89e287a9cfe0 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiSplitFactory.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiSplitFactory.java
@@ -15,11 +15,13 @@ package io.trino.plugin.hudi.split;
import com.google.common.collect.ImmutableList;
import io.airlift.units.DataSize;
+import io.trino.filesystem.cache.CachingHostAddressProvider;
import io.trino.plugin.hive.HivePartitionKey;
import io.trino.plugin.hudi.HudiSplit;
import io.trino.plugin.hudi.HudiTableHandle;
import io.trino.plugin.hudi.file.HudiBaseFile;
import io.trino.plugin.hudi.file.HudiLogFile;
+import io.trino.spi.HostAddress;
import io.trino.spi.TrinoException;
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieBaseFile;
@@ -41,20 +43,23 @@ public class HudiSplitFactory
private final HudiTableHandle hudiTableHandle;
private final HudiSplitWeightProvider hudiSplitWeightProvider;
private final DataSize targetSplitSize;
+ private final CachingHostAddressProvider cachingHostAddressProvider;
public HudiSplitFactory(
HudiTableHandle hudiTableHandle,
HudiSplitWeightProvider hudiSplitWeightProvider,
- DataSize targetSplitSize)
+ DataSize targetSplitSize,
+ CachingHostAddressProvider cachingHostAddressProvider)
{
this.hudiTableHandle = requireNonNull(hudiTableHandle, "hudiTableHandle is null");
this.hudiSplitWeightProvider = requireNonNull(hudiSplitWeightProvider, "hudiSplitWeightProvider is null");
this.targetSplitSize = requireNonNull(targetSplitSize, "targetSplitSize is null");
+ this.cachingHostAddressProvider = requireNonNull(cachingHostAddressProvider, "cachingHostAddressProvider is null");
}
public List<HudiSplit> createSplits(List<HivePartitionKey> partitionKeys, FileSlice fileSlice, String commitTime)
{
- return createHudiSplits(hudiTableHandle, partitionKeys, fileSlice, commitTime, hudiSplitWeightProvider, targetSplitSize);
+ return createHudiSplits(hudiTableHandle, partitionKeys, fileSlice, commitTime, hudiSplitWeightProvider, targetSplitSize, cachingHostAddressProvider);
}
public static List<HudiSplit> createHudiSplits(
@@ -63,7 +68,8 @@ public class HudiSplitFactory
FileSlice fileSlice,
String commitTime,
HudiSplitWeightProvider hudiSplitWeightProvider,
- DataSize targetSplitSize)
+ DataSize targetSplitSize,
+ CachingHostAddressProvider cachingHostAddressProvider)
{
if (fileSlice.isEmpty()) {
throw new TrinoException(HUDI_FILESYSTEM_ERROR, format("Not a valid file slice: %s", fileSlice.toString()));
@@ -78,6 +84,8 @@ public class HudiSplitFactory
"Hudi base file must exist if there is no log file in the file slice");
HoodieBaseFile baseFile = fileSlice.getBaseFile().get();
long fileSize = baseFile.getFileSize();
+ List<HostAddress> addresses = cachingHostAddressProvider.getHosts(
+ baseFile.getPath(), ImmutableList.of());
if (fileSize == 0) {
return ImmutableList.of(new HudiSplit(
@@ -86,7 +94,8 @@ public class HudiSplitFactory
commitTime,
hudiTableHandle.getRegularPredicates(),
partitionKeys,
- hudiSplitWeightProvider.calculateSplitWeight(fileSize)));
+ hudiSplitWeightProvider.calculateSplitWeight(fileSize),
+ addresses));
}
ImmutableList.Builder<HudiSplit> splits = ImmutableList.builder();
@@ -100,7 +109,8 @@ public class HudiSplitFactory
commitTime,
hudiTableHandle.getRegularPredicates(),
partitionKeys,
- hudiSplitWeightProvider.calculateSplitWeight(targetSplitSizeInBytes)));
+ hudiSplitWeightProvider.calculateSplitWeight(targetSplitSizeInBytes),
+ addresses));
bytesRemaining -= targetSplitSizeInBytes;
}
if (bytesRemaining > 0) {
@@ -110,7 +120,8 @@ public class HudiSplitFactory
commitTime,
hudiTableHandle.getRegularPredicates(),
partitionKeys,
- hudiSplitWeightProvider.calculateSplitWeight(bytesRemaining)));
+ hudiSplitWeightProvider.calculateSplitWeight(bytesRemaining),
+ addresses));
}
return splits.build();
}
@@ -123,6 +134,7 @@ public class HudiSplitFactory
commitTime,
hudiTableHandle.getRegularPredicates(),
partitionKeys,
- hudiSplitWeightProvider.calculateSplitWeight(fileSlice.getTotalFileSize())));
+ hudiSplitWeightProvider.calculateSplitWeight(fileSlice.getTotalFileSize()),
+ ImmutableList.of()));
}
}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/HudiQueryRunner.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/HudiQueryRunner.java
index c13e2b282727..0114eb8400a2 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/HudiQueryRunner.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/HudiQueryRunner.java
@@ -97,6 +97,13 @@ public final class HudiQueryRunner
return this;
}
+ @CanIgnoreReturnValue
+ public Builder addConnectorProperties(Map<String, String> properties)
+ {
+ this.connectorProperties.putAll(properties);
+ return this;
+ }
+
@Override
public DistributedQueryRunner build()
throws Exception
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
new file mode 100644
index 000000000000..7101cb897d77
--- /dev/null
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
@@ -0,0 +1,149 @@
+/*
+ * 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 com.google.common.collect.HashMultiset;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMultiset;
+import com.google.common.collect.Multiset;
+import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer;
+import io.trino.plugin.hudi.util.FileOperationUtils.FileOperation;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.DistributedQueryRunner;
+import io.trino.testing.QueryRunner;
+import org.intellij.lang.annotations.Language;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+
+import static com.google.common.io.MoreFiles.deleteRecursively;
+import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
+import static io.trino.filesystem.tracing.CacheFileSystemTraceUtils.getCacheOperationSpans;
+import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_FG_PT_V8_MOR;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.DATA;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.INDEX_DEFINITION;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE_PROPERTIES;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TABLE_PROPERTIES;
+import static io.trino.testing.MultisetAssertions.assertMultisetsEqual;
+import static java.util.stream.Collectors.toCollection;
+
+@Execution(ExecutionMode.SAME_THREAD)
+public class TestHudiAlluxioCacheFileOperations
+ extends AbstractTestQueryFramework
+{
+ private Path cacheDirectory;
+
+ @Override
+ protected DistributedQueryRunner createQueryRunner()
+ throws Exception
+ {
+ cacheDirectory = Files.createTempDirectory("cache");
+ closeAfterClass(() -> deleteRecursively(cacheDirectory, ALLOW_INSECURE));
+
+ Map<String, String> hudiProperties = ImmutableMap.<String, String>builder()
+ .put("hudi.metadata-enabled", "true")
+ .put("fs.cache.enabled", "true")
+ .put("fs.cache.directories", cacheDirectory.toAbsolutePath().toString())
+ .put("fs.cache.max-sizes", "100MB")
+ .put("hudi.metadata.cache.enabled", "false")
+ .buildOrThrow();
+
+ return HudiQueryRunner.builder()
+ .addConnectorProperties(hudiProperties)
+ .setDataLoader(new ResourceHudiTablesInitializer())
+ .setWorkerCount(0)
+ .build();
+ }
+
+ @Test
+ public void testSelectWithFilter()
+ {
+ @Language("SQL") String query = "SELECT * FROM " + HUDI_MULTI_FG_PT_V8_MOR + " WHERE country='SG'";
+ assertFileSystemAccesses(
+ query,
+ ImmutableMultiset.<FileOperation>builder()
+ .addCopies(new FileOperation("Alluxio.readCached", DATA), 2)
+ .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 24)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
+ .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 8)
+ .addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
+ .add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
+ .addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
+ .build());
+
+ assertFileSystemAccesses(
+ query,
+ ImmutableMultiset.<FileOperation>builder()
+ .addCopies(new FileOperation("Alluxio.readCached", DATA), 2)
+ .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 24)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
+ .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 8)
+ .addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
+ .add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
+ .addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
+ .build());
+ }
+
+ @Test
+ public void testJoin()
+ {
+ @Language("SQL") String query = "SELECT t1.id, t1.name, t1.price, t1.ts FROM " +
+ HUDI_MULTI_FG_PT_V8_MOR + " t1 " +
+ "INNER JOIN " + HUDI_MULTI_FG_PT_V8_MOR + " t2 ON t1.id = t2.id " +
+ "WHERE t2.price <= 102";
+
+ assertFileSystemAccesses(query,
+ ImmutableMultiset.<FileOperation>builder()
+ .addCopies(new FileOperation("Alluxio.readCached", DATA), 6)
+ .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 282)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 37)
+ .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 89)
+ .addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 5)
+ .addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 3)
+ .addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 5)
+ .build());
+
+ assertFileSystemAccesses(query,
+ ImmutableMultiset.<FileOperation>builder()
+ .addCopies(new FileOperation("Alluxio.readCached", DATA), 6)
+ .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 209)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 27)
+ .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 65)
+ .addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
+ .addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
+ .addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 4)
+ .build());
+ }
+
+ private void assertFileSystemAccesses(@Language("SQL") String query, Multiset<FileOperation> expectedCacheAccesses)
+ {
+ DistributedQueryRunner queryRunner = getDistributedQueryRunner();
+ queryRunner.executeWithPlan(queryRunner.getDefaultSession(), query);
+ assertMultisetsEqual(getFileOperations(queryRunner), expectedCacheAccesses);
+ }
+
+ public static Multiset<FileOperation> getFileOperations(QueryRunner queryRunner)
+ {
+ return getCacheOperationSpans(queryRunner)
+ .stream()
+ .filter(span -> !span.getName().startsWith("InputFile.exists"))
+ .map(FileOperation::create)
+ .collect(toCollection(HashMultiset::create));
+ }
+}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java
new file mode 100644
index 000000000000..0a082e1cb159
--- /dev/null
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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 com.google.common.collect.ImmutableMap;
+import com.google.common.io.Closer;
+import org.junit.jupiter.api.AfterAll;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import static com.google.common.io.MoreFiles.deleteRecursively;
+import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
+
+public class TestHudiAlluxioCachingSmokeTest
+ extends TestHudiSmokeTest
+{
+ private final Path cacheDirectory;
+ private final Closer closer = Closer.create();
+
+ TestHudiAlluxioCachingSmokeTest()
+ throws IOException
+ {
+ super();
+ cacheDirectory = Files.createTempDirectory("cache");
+ closer.register(() -> deleteRecursively(cacheDirectory, ALLOW_INSECURE));
+ }
+
+ @AfterAll
+ public void tearDown()
+ throws Exception
+ {
+ closer.close();
+ }
+
+ @Override
+ public ImmutableMap<String, String> getAdditionalHudiProperties()
+ {
+ return ImmutableMap.<String, String>builder()
+ .put("fs.cache.enabled", "true")
+ .put("fs.cache.directories", cacheDirectory.toAbsolutePath().toString())
+ .put("fs.cache.max-sizes", "1GB")
+ .buildOrThrow();
+ }
+}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
index c618559cdf67..a860b94ac241 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
@@ -56,7 +56,8 @@ public class TestHudiConfig
.setDynamicFilteringWaitTimeout(Duration.valueOf("1s"))
.setColumnStatsWaitTimeout(Duration.valueOf("1s"))
.setRecordIndexWaitTimeout(Duration.valueOf("2s"))
- .setSecondaryIndexWaitTimeout(Duration.valueOf("2s")));
+ .setSecondaryIndexWaitTimeout(Duration.valueOf("2s"))
+ .setMetadataCacheEnabled(true));
}
@Test
@@ -88,6 +89,7 @@ public class TestHudiConfig
.put("hudi.index.column-stats.wait-timeout", "2s")
.put("hudi.index.record-index.wait-timeout", "1s")
.put("hudi.index.secondary-index.wait-timeout", "1s")
+ .put("hudi.metadata.cache.enabled", "false")
.buildOrThrow();
HudiConfig expected = new HudiConfig()
@@ -115,7 +117,8 @@ public class TestHudiConfig
.setDynamicFilteringWaitTimeout(Duration.valueOf("2s"))
.setColumnStatsWaitTimeout(Duration.valueOf("2s"))
.setRecordIndexWaitTimeout(Duration.valueOf("1s"))
- .setSecondaryIndexWaitTimeout(Duration.valueOf("1s"));
+ .setSecondaryIndexWaitTimeout(Duration.valueOf("1s"))
+ .setMetadataCacheEnabled(false);
assertFullMapping(properties, expected);
}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
new file mode 100644
index 000000000000..707a53b22ed4
--- /dev/null
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
@@ -0,0 +1,141 @@
+/*
+ * 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 com.google.common.collect.HashMultiset;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMultiset;
+import com.google.common.collect.Multiset;
+import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer;
+import io.trino.plugin.hudi.util.FileOperationUtils.FileOperation;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.DistributedQueryRunner;
+import io.trino.testing.QueryRunner;
+import org.intellij.lang.annotations.Language;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+
+import java.util.Map;
+
+import static io.trino.filesystem.tracing.CacheFileSystemTraceUtils.getFileLocation;
+import static io.trino.filesystem.tracing.CacheFileSystemTraceUtils.isTrinoSchemaOrPermissions;
+import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_FG_PT_V8_MOR;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.DATA;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.INDEX_DEFINITION;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE_PROPERTIES;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TABLE_PROPERTIES;
+import static io.trino.testing.MultisetAssertions.assertMultisetsEqual;
+import static java.util.stream.Collectors.toCollection;
+
+@Execution(ExecutionMode.SAME_THREAD)
+public class TestHudiMemoryCacheFileOperations
+ extends AbstractTestQueryFramework
+{
+ @Override
+ protected DistributedQueryRunner createQueryRunner()
+ throws Exception
+ {
+ Map<String, String> hudiProperties = ImmutableMap.<String, String>builder()
+ .put("hudi.metadata-enabled", "true")
+ .put("hudi.metadata.cache.enabled", "true")
+ .put("fs.cache.enabled", "false")
+ .buildOrThrow();
+
+ return HudiQueryRunner.builder()
+ .addConnectorProperties(hudiProperties)
+ .setDataLoader(new ResourceHudiTablesInitializer())
+ .setWorkerCount(0)
+ .build();
+ }
+
+ @Test
+ public void testSelectWithFilter()
+ {
+ @Language("SQL") String query = "SELECT * FROM " + HUDI_MULTI_FG_PT_V8_MOR + " WHERE country='SG'";
+ assertFileSystemAccesses(
+ query,
+ ImmutableMultiset.<FileOperation>builder()
+ .addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 2)
+ .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 3)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 5)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
+ .addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
+ .add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
+ .addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
+ .build());
+
+ assertFileSystemAccesses(
+ query,
+ ImmutableMultiset.<FileOperation>builder()
+ .addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 2)
+ .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 3)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 5)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
+ .addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
+ .add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
+ .addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
+ .build());
+ }
+
+ @Test
+ public void testJoin()
+ {
+ @Language("SQL") String query = "SELECT t1.id, t1.name, t1.price, t1.ts FROM " +
+ HUDI_MULTI_FG_PT_V8_MOR + " t1 " +
+ "INNER JOIN " + HUDI_MULTI_FG_PT_V8_MOR + " t2 ON t1.id = t2.id " +
+ "WHERE t2.price <= 102";
+
+ assertFileSystemAccesses(query,
+ ImmutableMultiset.<FileOperation>builder()
+ .addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 6)
+ .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 37)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 52)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 37)
+ .addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 5)
+ .addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 3)
+ .addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 5)
+ .build());
+
+ assertFileSystemAccesses(query,
+ ImmutableMultiset.<FileOperation>builder()
+ .addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 6)
+ .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 27)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 38)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 27)
+ .addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
+ .addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
+ .addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 4)
+ .build());
+ }
+
+ private void assertFileSystemAccesses(@Language("SQL") String query, Multiset<FileOperation> expectedCacheAccesses)
+ {
+ DistributedQueryRunner queryRunner = getDistributedQueryRunner();
+ queryRunner.executeWithPlan(queryRunner.getDefaultSession(), query);
+ assertMultisetsEqual(getFileOperations(queryRunner), expectedCacheAccesses);
+ }
+
+ private static Multiset<FileOperation> getFileOperations(QueryRunner queryRunner)
+ {
+ return queryRunner.getSpans().stream()
+ .filter(span -> span.getName().startsWith("Input.") || span.getName().startsWith("InputFile.") || span.getName().startsWith("FileSystemCache."))
+ .filter(span -> !span.getName().startsWith("InputFile.newInput"))
+ .filter(span -> !span.getName().startsWith("InputFile.exists"))
+ .filter(span -> !isTrinoSchemaOrPermissions(getFileLocation(span)))
+ .map(FileOperation::create)
+ .collect(toCollection(HashMultiset::create));
+ }
+}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
new file mode 100644
index 000000000000..0786e7497455
--- /dev/null
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
@@ -0,0 +1,141 @@
+/*
+ * 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 com.google.common.collect.HashMultiset;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMultiset;
+import com.google.common.collect.Multiset;
+import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer;
+import io.trino.plugin.hudi.util.FileOperationUtils;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.DistributedQueryRunner;
+import io.trino.testing.QueryRunner;
+import org.intellij.lang.annotations.Language;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+
+import java.util.Map;
+
+import static io.trino.filesystem.tracing.CacheFileSystemTraceUtils.getFileLocation;
+import static io.trino.filesystem.tracing.CacheFileSystemTraceUtils.isTrinoSchemaOrPermissions;
+import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_FG_PT_V8_MOR;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.DATA;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.INDEX_DEFINITION;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE_PROPERTIES;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TABLE_PROPERTIES;
+import static io.trino.testing.MultisetAssertions.assertMultisetsEqual;
+import static java.util.stream.Collectors.toCollection;
+
+@Execution(ExecutionMode.SAME_THREAD)
+public class TestHudiNoCacheFileOperations
+ extends AbstractTestQueryFramework
+{
+ @Override
+ protected DistributedQueryRunner createQueryRunner()
+ throws Exception
+ {
+ Map<String, String> hudiProperties = ImmutableMap.<String, String>builder()
+ .put("hudi.metadata-enabled", "true")
+ .put("hudi.metadata.cache.enabled", "false")
+ .put("fs.cache.enabled", "false")
+ .buildOrThrow();
+
+ return HudiQueryRunner.builder()
+ .addConnectorProperties(hudiProperties)
+ .setDataLoader(new ResourceHudiTablesInitializer())
+ .setWorkerCount(0)
+ .build();
+ }
+
+ @Test
+ public void testSelectWithFilter()
+ {
+ @Language("SQL") String query = "SELECT * FROM " + HUDI_MULTI_FG_PT_V8_MOR + " WHERE country='SG'";
+ assertFileSystemAccesses(
+ query,
+ ImmutableMultiset.<FileOperationUtils.FileOperation>builder()
+ .addCopies(new FileOperationUtils.FileOperation("Input.readTail", DATA), 2)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 3)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 5)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
+ .add(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
+ .build());
+
+ assertFileSystemAccesses(
+ query,
+ ImmutableMultiset.<FileOperationUtils.FileOperation>builder()
+ .addCopies(new FileOperationUtils.FileOperation("Input.readTail", DATA), 2)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 3)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 5)
+ .add(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
+ .build());
+ }
+
+ @Test
+ public void testJoin()
+ {
+ @Language("SQL") String query = "SELECT t1.id, t1.name, t1.price, t1.ts FROM " +
+ HUDI_MULTI_FG_PT_V8_MOR + " t1 " +
+ "INNER JOIN " + HUDI_MULTI_FG_PT_V8_MOR + " t2 ON t1.id = t2.id " +
+ "WHERE t2.price <= 102";
+
+ assertFileSystemAccesses(query,
+ ImmutableMultiset.<FileOperationUtils.FileOperation>builder()
+ .addCopies(new FileOperationUtils.FileOperation("Input.readTail", DATA), 6)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 37)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 37)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", INDEX_DEFINITION), 5)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 52)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 3)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 5)
+ .build());
+
+ assertFileSystemAccesses(query,
+ ImmutableMultiset.<FileOperationUtils.FileOperation>builder()
+ .addCopies(new FileOperationUtils.FileOperation("Input.readTail", DATA), 6)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 27)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 27)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 38)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 4)
+ .build());
+ }
+
+ private void assertFileSystemAccesses(@Language("SQL") String query, Multiset<FileOperationUtils.FileOperation> expectedCacheAccesses)
+ {
+ DistributedQueryRunner queryRunner = getDistributedQueryRunner();
+ queryRunner.executeWithPlan(queryRunner.getDefaultSession(), query);
+ assertMultisetsEqual(getFileOperations(queryRunner), expectedCacheAccesses);
+ }
+
+ private static Multiset<FileOperationUtils.FileOperation> getFileOperations(QueryRunner queryRunner)
+ {
+ return queryRunner.getSpans().stream()
+ .filter(span -> span.getName().startsWith("Input.") || span.getName().startsWith("InputFile.") || span.getName().startsWith("FileSystemCache."))
+ .filter(span -> !span.getName().startsWith("InputFile.newInput"))
+ .filter(span -> !span.getName().startsWith("InputFile.exists"))
+ .filter(span -> !isTrinoSchemaOrPermissions(getFileLocation(span)))
+ .map(FileOperationUtils.FileOperation::create)
+ .collect(toCollection(HashMultiset::create));
+ }
+}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
index 3f7300576789..fddf996c644f 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
@@ -14,6 +14,7 @@
package io.trino.plugin.hudi;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import com.google.common.io.Resources;
import io.trino.Session;
import io.trino.filesystem.Location;
@@ -95,9 +96,15 @@ public class TestHudiSmokeTest
{
return HudiQueryRunner.builder()
.setDataLoader(new ResourceHudiTablesInitializer())
+ .addConnectorProperties(getAdditionalHudiProperties())
.build();
}
+ protected ImmutableMap<String, String> getAdditionalHudiProperties()
+ {
+ return ImmutableMap.of();
+ }
+
@Test
public void testReadNonPartitionedTable()
{
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java
index 37d4eac6e88b..19e7f7fdf1ec 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java
@@ -15,6 +15,7 @@ package io.trino.plugin.hudi.split;
import com.google.common.collect.ImmutableList;
import io.airlift.units.DataSize;
+import io.trino.filesystem.cache.DefaultCachingHostAddressProvider;
import io.trino.plugin.hive.HivePartitionKey;
import io.trino.plugin.hudi.HudiSplit;
import io.trino.plugin.hudi.HudiTableHandle;
@@ -151,7 +152,8 @@ public class TestHudiSplitFactory
FileSlice fileSlice = createFileSlice(baseFileSize, logFileSize);
List<HudiSplit> splits = HudiSplitFactory.createHudiSplits(
- tableHandle, PARTITION_KEYS, fileSlice, COMMIT_TIME, weightProvider, targetSplitSize);
+ tableHandle, PARTITION_KEYS, fileSlice, COMMIT_TIME, weightProvider, targetSplitSize,
+ new DefaultCachingHostAddressProvider());
assertThat(splits).hasSize(expectedSplitInfo.size());
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/util/FileOperationUtils.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/util/FileOperationUtils.java
new file mode 100644
index 000000000000..a53fdeecc393
--- /dev/null
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/util/FileOperationUtils.java
@@ -0,0 +1,85 @@
+/*
+ * 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 io.opentelemetry.sdk.trace.data.SpanData;
+
+import static io.trino.filesystem.tracing.CacheFileSystemTraceUtils.getFileLocation;
+import static java.util.Objects.requireNonNull;
+
+public final class FileOperationUtils
+{
+ private FileOperationUtils() {}
+
+ public record FileOperation(String operationType, FileType fileType)
+ {
+ public FileOperation
+ {
+ requireNonNull(operationType, "operationType is null");
+ requireNonNull(fileType, "fileType is null");
+ }
+
+ public static FileOperation create(SpanData span)
+ {
+ String path = getFileLocation(span);
+ return new FileOperation(span.getName(), FileOperationUtils.FileType.fromFilePath(path));
+ }
+ }
+
+ public enum FileType
+ {
+ METADATA_TABLE_PROPERTIES,
+ METADATA_TABLE,
+ TIMELINE,
+ TABLE_PROPERTIES,
+ INDEX_DEFINITION,
+ OTHER_TABLE_METADATA,
+ COMMIT_INFLIGHT,
+ DATA,
+ LOG,
+ METASTORE;
+
+ public static FileType fromFilePath(String path)
+ {
+ if (path.contains("/.hoodie")) {
+ if (path.endsWith("/.hoodie/metadata/.hoodie/hoodie.properties")) {
+ return METADATA_TABLE_PROPERTIES;
+ }
+ if (path.contains("/.hoodie/metadata")) {
+ return METADATA_TABLE;
+ }
+ if (path.contains("/.hoodie/timeline")) {
+ return TIMELINE;
+ }
+ if (path.endsWith("/.hoodie/hoodie.properties")) {
+ return TABLE_PROPERTIES;
+ }
+ if (path.endsWith("/.hoodie/.index_defs/index.json")) {
+ return INDEX_DEFINITION;
+ }
+ return OTHER_TABLE_METADATA;
+ }
+ if (path.endsWith(".parquet")) {
+ return DATA;
+ }
+ if (path.contains(".log")) {
+ return LOG;
+ }
+ if (path.endsWith(".trinoSchema") || path.contains("/.trinoPermissions/")) {
+ return METASTORE;
+ }
+ throw new IllegalArgumentException("File not recognized: " + path);
+ }
+ }
+}
diff --git a/plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/cache/HudiCacheKeyProvider.java b/plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/cache/HudiCacheKeyProvider.java
new file mode 100644
index 000000000000..cd07e1ab3f13
--- /dev/null
+++ b/plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/cache/HudiCacheKeyProvider.java
@@ -0,0 +1,38 @@
+/*
+ * 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.cache;
+
+import io.trino.filesystem.TrinoInputFile;
+import io.trino.filesystem.cache.CacheKeyProvider;
+
+import java.util.Optional;
+
+public class HudiCacheKeyProvider
+ implements CacheKeyProvider
+{
+ @Override
+ public Optional<String> getCacheKey(TrinoInputFile inputFile)
+ {
+ String path = inputFile.location().path();
+ if (path.endsWith(".trinoSchema") || path.contains("/.trinoPermissions/")
+ || path.endsWith("/.hoodie/hoodie.properties")
+ || path.endsWith("/.hoodie/.index_defs/index.json")) {
+ // Needed to avoid caching files from FileHiveMetastore on coordinator during tests
+ // and Hudi table properties file that can be mutable
+ return Optional.empty();
+ }
+ // Other Hudi data and metadata files are immutable
+ return Optional.of(path);
+ }
+}
(hudi) 08/13: [ENG-30888][Trino] Enable Metadata Table by default (#59)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d13b7ec1766123eb72048cf404f7ae25be54bd38
Author: vamsikarnika <va...@gmail.com>
AuthorDate: Wed Aug 6 03:45:56 2025 +0530
[ENG-30888][Trino] Enable Metadata Table by default (#59)
Co-authored-by: Vamsi <va...@onehouse.ai>
(cherry picked from commit a868d5c5dcb2165859dfbecda7dddb2c4b6b4caa)
---
.../src/main/java/io/trino/plugin/hudi/HudiConfig.java | 2 +-
.../src/test/java/io/trino/plugin/hudi/TestHudiConfig.java | 6 +++---
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
index 3177f7c4cd23..743eb65938d6 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
@@ -40,7 +40,7 @@ public class HudiConfig
private List<String> columnsToHide = ImmutableList.of();
private boolean tableStatisticsEnabled = true;
private int tableStatisticsExecutorParallelism = 4;
- private boolean metadataEnabled;
+ private boolean metadataEnabled = true;
private boolean shouldUseParquetColumnNames = true;
private boolean shouldUseParquetColumnIndex;
private boolean sizeBasedSplitWeightsEnabled = true;
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
index a860b94ac241..22f6e6609106 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
@@ -34,7 +34,7 @@ public class TestHudiConfig
assertRecordedDefaults(recordDefaults(HudiConfig.class)
.setColumnsToHide(ImmutableList.of())
.setTableStatisticsEnabled(true)
- .setMetadataEnabled(false)
+ .setMetadataEnabled(true)
.setUseParquetColumnNames(true)
.setUseParquetColumnIndex(false)
.setTableStatisticsExecutorParallelism(4)
@@ -66,7 +66,7 @@ public class TestHudiConfig
Map<String, String> properties = ImmutableMap.<String, String>builder()
.put("hudi.columns-to-hide", "_hoodie_record_key")
.put("hudi.table-statistics-enabled", "false")
- .put("hudi.metadata-enabled", "true")
+ .put("hudi.metadata-enabled", "false")
.put("hudi.parquet.use-column-names", "false")
.put("hudi.parquet.use-column-index", "true")
.put("hudi.table-statistics-executor-parallelism", "16")
@@ -95,7 +95,7 @@ public class TestHudiConfig
HudiConfig expected = new HudiConfig()
.setColumnsToHide(ImmutableList.of("_hoodie_record_key"))
.setTableStatisticsEnabled(false)
- .setMetadataEnabled(true)
+ .setMetadataEnabled(false)
.setUseParquetColumnNames(false)
.setUseParquetColumnIndex(true)
.setTableStatisticsExecutorParallelism(16)
(hudi) 10/13: [ENG-29378] Implement Metadata table based Partition listing (#49)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 846d02942f3008ca952ba27a87879e917d93e016
Author: vamsikarnika <va...@gmail.com>
AuthorDate: Wed Aug 20 15:28:01 2025 +0800
[ENG-29378] Implement Metadata table based Partition listing (#49)
Co-authored-by: Vamsi <va...@onehouse.ai>
(cherry picked from commit 2725aade174454921e49082c59ea3250b36db96f)
---
hudi-trino-plugin/pom.xml | 24 +++
.../main/java/io/trino/plugin/hudi/HudiConfig.java | 14 ++
.../trino/plugin/hudi/HudiSessionProperties.java | 11 ++
.../java/io/trino/plugin/hudi/HudiSplitSource.java | 13 +-
.../main/java/io/trino/plugin/hudi/HudiUtil.java | 3 +-
.../hudi/partition/HiveHudiPartitionInfo.java | 9 +-
.../hudi/partition/HudiPartitionInfoLoader.java | 32 ++-
.../plugin/hudi/query/HudiDirectoryLister.java | 3 -
.../hudi/query/HudiSnapshotDirectoryLister.java | 25 +--
.../hudi/split/HudiBackgroundSplitLoader.java | 216 ++++++++++++++++++++-
.../hudi/TestHudiAlluxioCacheFileOperations.java | 24 +--
.../java/io/trino/plugin/hudi/TestHudiConfig.java | 3 +
.../hudi/TestHudiMemoryCacheFileOperations.java | 24 +--
.../plugin/hudi/TestHudiNoCacheFileOperations.java | 24 +--
.../io/trino/plugin/hudi/TestHudiSmokeTest.java | 20 ++
.../testing/ResourceHudiTablesInitializer.java | 22 +++
.../hudi_non_extractable_partition_path.md | 48 +++++
.../hudi_non_extractable_partition_path.zip | Bin 0 -> 75758 bytes
18 files changed, 417 insertions(+), 98 deletions(-)
diff --git a/hudi-trino-plugin/pom.xml b/hudi-trino-plugin/pom.xml
index 08a8ebfdee32..4ff439d1e35c 100644
--- a/hudi-trino-plugin/pom.xml
+++ b/hudi-trino-plugin/pom.xml
@@ -147,6 +147,18 @@
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hive-sync</artifactId>
+ <version>${dep.hudi.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-io</artifactId>
@@ -159,6 +171,18 @@
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-sync-common</artifactId>
+ <version>${dep.hudi.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-column</artifactId>
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
index 743eb65938d6..c0df68088c93 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
@@ -65,6 +65,7 @@ public class HudiConfig
private Duration recordIndexWaitTimeout = new Duration(2, SECONDS);
private Duration secondaryIndexWaitTimeout = new Duration(2, SECONDS);
private boolean metadataCacheEnabled = true;
+ private boolean metadataPartitionListingEnabled = true;
public List<String> getColumnsToHide()
{
@@ -421,4 +422,17 @@ public class HudiConfig
this.metadataCacheEnabled = metadataCacheEnabled;
return this;
}
+
+ public boolean isMetadataPartitionListingEnabled()
+ {
+ return metadataPartitionListingEnabled;
+ }
+
+ @Config("hudi.metadata.partition-listing.enabled")
+ @ConfigDescription("Enables listing table partitions through the metadata table.")
+ public HudiConfig setMetadataPartitionListingEnabled(boolean metadataPartitionListingEnabled)
+ {
+ this.metadataPartitionListingEnabled = metadataPartitionListingEnabled;
+ return this;
+ }
}
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSessionProperties.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSessionProperties.java
index c1414f13ee89..2ad9da482806 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSessionProperties.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSessionProperties.java
@@ -73,6 +73,7 @@ public class HudiSessionProperties
static final String COLUMN_STATS_WAIT_TIMEOUT = "column_stats_wait_timeout";
static final String RECORD_INDEX_WAIT_TIMEOUT = "record_index_wait_timeout";
static final String SECONDARY_INDEX_WAIT_TIMEOUT = "secondary_index_wait_timeout";
+ static final String METADATA_PARTITION_LISTING_ENABLED = "metadata_partition_listing_enabled";
private final List<PropertyMetadata<?>> sessionProperties;
@@ -238,6 +239,11 @@ public class HudiSessionProperties
DYNAMIC_FILTERING_WAIT_TIMEOUT,
"Duration to wait for completion of dynamic filters during split generation",
hudiConfig.getDynamicFilteringWaitTimeout(),
+ false),
+ booleanProperty(
+ METADATA_PARTITION_LISTING_ENABLED,
+ "Enable metadata table based partition listing",
+ hudiConfig.isMetadataPartitionListingEnabled(),
false));
}
@@ -392,4 +398,9 @@ public class HudiSessionProperties
{
return session.getProperty(SECONDARY_INDEX_WAIT_TIMEOUT, Duration.class);
}
+
+ public static boolean isMetadataPartitionListingEnabled(ConnectorSession session)
+ {
+ return session.getProperty(METADATA_PARTITION_LISTING_ENABLED, Boolean.class);
+ }
}
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
index f7e773d983f2..278ca2e463c7 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
@@ -16,6 +16,7 @@ package io.trino.plugin.hudi;
import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.Futures;
+import io.airlift.log.Logger;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
import io.trino.filesystem.cache.CachingHostAddressProvider;
@@ -42,6 +43,7 @@ import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.engine.HoodieLocalEngineContext;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.util.Lazy;
@@ -71,6 +73,8 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
public class HudiSplitSource
implements ConnectorSplitSource
{
+ private static final Logger log = Logger.get(HudiSplitSource.class);
+
private static final ConnectorSplitBatch EMPTY_BATCH = new ConnectorSplitBatch(ImmutableList.of(), false);
private final AsyncQueue<ConnectorSplit> queue;
private final ScheduledFuture splitLoaderFuture;
@@ -93,22 +97,25 @@ public class HudiSplitSource
{
boolean enableMetadataTable = isHudiMetadataTableEnabled(session);
Lazy<HoodieTableMetadata> lazyTableMetadata = Lazy.lazily(() -> {
+ HoodieTimer timer = HoodieTimer.start();
HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder()
.enable(enableMetadataTable)
.build();
HoodieTableMetaClient metaClient = tableHandle.getMetaClient();
HoodieEngineContext engineContext = new HoodieLocalEngineContext(metaClient.getStorage().getConf());
- return HoodieTableMetadata.create(
+
+ HoodieTableMetadata tableMetadata = HoodieTableMetadata.create(
engineContext,
tableHandle.getMetaClient().getStorage(), metadataConfig, metaClient.getBasePath().toString(), true);
+ log.info("Loaded table metadata for table: %s in %s ms", tableHandle.getSchemaTableName(), timer.endTimer());
+ return tableMetadata;
});
HudiDirectoryLister hudiDirectoryLister = new HudiSnapshotDirectoryLister(
session,
tableHandle,
enableMetadataTable,
- lazyTableMetadata,
- lazyPartitions);
+ lazyTableMetadata);
this.queue = new ThrottledAsyncQueue<>(maxSplitsPerSecond, maxOutstandingSplits, executor);
HudiBackgroundSplitLoader splitLoader = new HudiBackgroundSplitLoader(
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiUtil.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiUtil.java
index 05c85efdbe24..0f61ceb1beb4 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiUtil.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiUtil.java
@@ -113,10 +113,11 @@ public final class HudiUtil
SchemaTableName tableName,
String hivePartitionName,
List<HiveColumnHandle> partitionColumnHandles,
+ List<String> partitionValues,
TupleDomain<HiveColumnHandle> constraintSummary)
{
HivePartition partition = HivePartitionManager.parsePartition(
- tableName, hivePartitionName, partitionColumnHandles);
+ tableName, hivePartitionName, partitionColumnHandles, partitionValues);
return partitionMatches(partitionColumnHandles, constraintSummary, partition);
}
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/partition/HiveHudiPartitionInfo.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/partition/HiveHudiPartitionInfo.java
index f7e6fb1fa81b..c660fdb10b25 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/partition/HiveHudiPartitionInfo.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/partition/HiveHudiPartitionInfo.java
@@ -30,13 +30,13 @@ public class HiveHudiPartitionInfo
implements HudiPartitionInfo
{
public static final String NON_PARTITION = "";
-
- private final SchemaTableName schemaTableName;
private final List<HiveColumnHandle> partitionColumnHandles;
private final TupleDomain<HiveColumnHandle> constraintSummary;
private final String hivePartitionName;
private final String relativePartitionPath;
private final List<HivePartitionKey> hivePartitionKeys;
+ private final SchemaTableName schemaTableName;
+ private final List<String> partitionValues;
public HiveHudiPartitionInfo(
SchemaTableName schemaTableName,
@@ -55,6 +55,7 @@ public class HiveHudiPartitionInfo
Location.of(partition.getStorage().getLocation()));
this.hivePartitionKeys = buildPartitionKeys(
partitionColumnHandles, partition.getValues());
+ this.partitionValues = partition.getValues();
}
@Override
@@ -75,7 +76,7 @@ public class HiveHudiPartitionInfo
if (hivePartitionName.equals(NON_PARTITION)) {
return true;
}
- return partitionMatchesPredicates(schemaTableName, hivePartitionName, partitionColumnHandles, constraintSummary);
+ return partitionMatchesPredicates(schemaTableName, hivePartitionName, partitionColumnHandles, partitionValues, constraintSummary);
}
public String getHivePartitionName()
@@ -89,7 +90,7 @@ public class HiveHudiPartitionInfo
String fullPartitionPath = fullPartitionLocation.path();
if (!fullPartitionPath.startsWith(basePath)) {
- throw new IllegalArgumentException("Partition location does not belong to base-location");
+ throw new IllegalArgumentException("Partition [" + fullPartitionPath + "] does not belong to base-location " + basePath);
}
String baseLocationParent = baseLocation.parentDirectory().path();
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/partition/HudiPartitionInfoLoader.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/partition/HudiPartitionInfoLoader.java
index 874297b9b692..0146cd4baea3 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/partition/HudiPartitionInfoLoader.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/partition/HudiPartitionInfoLoader.java
@@ -23,9 +23,6 @@ import org.apache.hudi.common.model.FileSlice;
import java.util.Deque;
import java.util.List;
-import java.util.Optional;
-
-import static io.trino.plugin.hudi.partition.HiveHudiPartitionInfo.NON_PARTITION;
public class HudiPartitionInfoLoader
implements Runnable
@@ -33,7 +30,7 @@ public class HudiPartitionInfoLoader
private final HudiDirectoryLister hudiDirectoryLister;
private final HudiSplitFactory hudiSplitFactory;
private final AsyncQueue<ConnectorSplit> asyncQueue;
- private final Deque<String> partitionQueue;
+ private final Deque<HiveHudiPartitionInfo> partitionQueue;
private final String commitTime;
private final boolean useIndex;
@@ -44,7 +41,7 @@ public class HudiPartitionInfoLoader
String commitTime,
HudiSplitFactory hudiSplitFactory,
AsyncQueue<ConnectorSplit> asyncQueue,
- Deque<String> partitionQueue,
+ Deque<HiveHudiPartitionInfo> partitionQueue,
boolean useIndex)
{
this.hudiDirectoryLister = hudiDirectoryLister;
@@ -60,27 +57,22 @@ public class HudiPartitionInfoLoader
public void run()
{
while (isRunning || !partitionQueue.isEmpty()) {
- String partitionName = partitionQueue.poll();
+ HiveHudiPartitionInfo hudiPartitionInfo = partitionQueue.poll();
- if (partitionName != null) {
- generateSplitsFromPartition(partitionName);
+ if (hudiPartitionInfo != null && hudiPartitionInfo.getHivePartitionName() != null) {
+ generateSplitsFromPartition(hudiPartitionInfo);
}
}
}
- private void generateSplitsFromPartition(String partitionName)
+ private void generateSplitsFromPartition(HiveHudiPartitionInfo hudiPartitionInfo)
{
- Optional<HudiPartitionInfo> partitionInfo = hudiDirectoryLister.getPartitionInfo(partitionName);
- partitionInfo.ifPresent(hudiPartitionInfo -> {
- if (hudiPartitionInfo.doesMatchPredicates() || partitionName.equals(NON_PARTITION)) {
- List<HivePartitionKey> partitionKeys = hudiPartitionInfo.getHivePartitionKeys();
- List<FileSlice> partitionFileSlices = hudiDirectoryLister.listStatus(hudiPartitionInfo, useIndex);
- partitionFileSlices.stream()
- .flatMap(slice -> hudiSplitFactory.createSplits(partitionKeys, slice, commitTime).stream())
- .map(asyncQueue::offer)
- .forEachOrdered(MoreFutures::getFutureValue);
- }
- });
+ List<HivePartitionKey> partitionKeys = hudiPartitionInfo.getHivePartitionKeys();
+ List<FileSlice> partitionFileSlices = hudiDirectoryLister.listStatus(hudiPartitionInfo, useIndex);
+ partitionFileSlices.stream()
+ .flatMap(slice -> hudiSplitFactory.createSplits(partitionKeys, slice, this.commitTime).stream())
+ .map(asyncQueue::offer)
+ .forEachOrdered(MoreFutures::getFutureValue);
}
public void stopRunning()
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiDirectoryLister.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiDirectoryLister.java
index fb8f1903358e..38c4952676e5 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiDirectoryLister.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiDirectoryLister.java
@@ -18,12 +18,9 @@ import org.apache.hudi.common.model.FileSlice;
import java.io.Closeable;
import java.util.List;
-import java.util.Optional;
public interface HudiDirectoryLister
extends Closeable
{
List<FileSlice> listStatus(HudiPartitionInfo partitionInfo, boolean useIndex);
-
- Optional<HudiPartitionInfo> getPartitionInfo(String partition);
}
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiSnapshotDirectoryLister.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiSnapshotDirectoryLister.java
index dfd7ef7d3c39..7e235578620b 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiSnapshotDirectoryLister.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiSnapshotDirectoryLister.java
@@ -15,10 +15,7 @@ package io.trino.plugin.hudi.query;
import com.google.common.collect.ImmutableList;
import io.airlift.log.Logger;
-import io.trino.filesystem.Location;
-import io.trino.metastore.Partition;
import io.trino.plugin.hudi.HudiTableHandle;
-import io.trino.plugin.hudi.partition.HiveHudiPartitionInfo;
import io.trino.plugin.hudi.partition.HudiPartitionInfo;
import io.trino.plugin.hudi.query.index.HudiIndexSupport;
import io.trino.plugin.hudi.query.index.IndexSupportFactory;
@@ -32,9 +29,7 @@ import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.util.Lazy;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -46,15 +41,13 @@ public class HudiSnapshotDirectoryLister
private static final Logger log = Logger.get(HudiSnapshotDirectoryLister.class);
private final HudiTableHandle tableHandle;
private final Lazy<HoodieTableFileSystemView> lazyFileSystemView;
- private final Lazy<Map<String, HudiPartitionInfo>> lazyAllPartitionInfoMap;
private final Optional<HudiIndexSupport> indexSupportOpt;
public HudiSnapshotDirectoryLister(
ConnectorSession session,
HudiTableHandle tableHandle,
boolean enableMetadataTable,
- Lazy<HoodieTableMetadata> lazyTableMetadata,
- Lazy<Map<String, Partition>> lazyAllPartitions)
+ Lazy<HoodieTableMetadata> lazyTableMetadata)
{
this.tableHandle = tableHandle;
SchemaTableName schemaTableName = tableHandle.getSchemaTableName();
@@ -69,16 +62,6 @@ public class HudiSnapshotDirectoryLister
return fileSystemView;
});
- this.lazyAllPartitionInfoMap = Lazy.lazily(() -> lazyAllPartitions.get().entrySet().stream()
- .collect(Collectors.toMap(
- Map.Entry::getKey,
- e -> new HiveHudiPartitionInfo(
- schemaTableName,
- Location.of(tableHandle.getBasePath()),
- e.getKey(),
- e.getValue(),
- tableHandle.getPartitionColumns(),
- tableHandle.getPartitionPredicates()))));
Lazy<HoodieTableMetaClient> lazyMetaClient = Lazy.lazily(tableHandle::getMetaClient);
this.indexSupportOpt = enableMetadataTable ?
IndexSupportFactory.createIndexSupport(schemaTableName, lazyMetaClient, lazyTableMetadata, tableHandle.getRegularPredicates(), session) : Optional.empty();
@@ -107,12 +90,6 @@ public class HudiSnapshotDirectoryLister
return collect;
}
- @Override
- public Optional<HudiPartitionInfo> getPartitionInfo(String partition)
- {
- return Optional.ofNullable(lazyAllPartitionInfoMap.get().get(partition));
- }
-
@Override
public void close()
{
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
index c4c91ccadd71..7c1ae6606edb 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
@@ -13,14 +13,20 @@
*/
package io.trino.plugin.hudi.split;
+import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.concurrent.BoundedExecutor;
import io.airlift.log.Logger;
+import io.trino.filesystem.Location;
import io.trino.filesystem.cache.CachingHostAddressProvider;
+import io.trino.metastore.Column;
import io.trino.metastore.Partition;
+import io.trino.metastore.StorageFormat;
+import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hive.util.AsyncQueue;
import io.trino.plugin.hudi.HudiTableHandle;
+import io.trino.plugin.hudi.partition.HiveHudiPartitionInfo;
import io.trino.plugin.hudi.partition.HudiPartitionInfoLoader;
import io.trino.plugin.hudi.query.HudiDirectoryLister;
import io.trino.plugin.hudi.query.index.HudiPartitionStatsIndexSupport;
@@ -29,11 +35,22 @@ import io.trino.spi.TrinoException;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.ConnectorSplit;
import io.trino.spi.connector.SchemaTableName;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hive.HiveStylePartitionValueExtractor;
+import org.apache.hudi.hive.MultiPartKeysValueExtractor;
+import org.apache.hudi.hive.NonPartitionedExtractor;
+import org.apache.hudi.hive.SinglePartPartitionValueExtractor;
import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.sync.common.model.PartitionValueExtractor;
import org.apache.hudi.util.Lazy;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
+import java.util.HexFormat;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -42,28 +59,39 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static io.airlift.concurrent.MoreFutures.addExceptionCallback;
import static io.trino.plugin.hudi.HudiErrorCode.HUDI_CANNOT_OPEN_SPLIT;
import static io.trino.plugin.hudi.HudiSessionProperties.getSplitGeneratorParallelism;
import static io.trino.plugin.hudi.HudiSessionProperties.getTargetSplitSize;
+import static io.trino.plugin.hudi.HudiSessionProperties.isMetadataPartitionListingEnabled;
+import static io.trino.plugin.hudi.partition.HiveHudiPartitionInfo.NON_PARTITION;
import static java.util.Objects.requireNonNull;
public class HudiBackgroundSplitLoader
implements Runnable
{
private static final Logger log = Logger.get(HudiBackgroundSplitLoader.class);
+ public static final String DELIMITER_STR = "/";
+ public static final String EQUALS_STR = "=";
+
private final HudiTableHandle tableHandle;
private final HudiDirectoryLister hudiDirectoryLister;
private final AsyncQueue<ConnectorSplit> asyncQueue;
private final Executor executor;
private final int splitGeneratorNumThreads;
private final HudiSplitFactory hudiSplitFactory;
- private final Lazy<List<String>> lazyPartitions;
+ private final Lazy<Map<String, Partition>> lazyPartitionMap;
private final Consumer<Throwable> errorListener;
private final boolean enableMetadataTable;
+ private final Lazy<HoodieTableMetadata> lazyTableMetadata;
private final Optional<HudiPartitionStatsIndexSupport> partitionIndexSupportOpt;
+ private final boolean isMetadataPartitionListingEnabled;
+ private final Lazy<HoodieTableMetaClient> lazyMetaClient;
public HudiBackgroundSplitLoader(
ConnectorSession session,
@@ -83,13 +111,16 @@ public class HudiBackgroundSplitLoader
this.asyncQueue = requireNonNull(asyncQueue, "asyncQueue is null");
this.splitGeneratorNumThreads = getSplitGeneratorParallelism(session);
this.hudiSplitFactory = new HudiSplitFactory(tableHandle, hudiSplitWeightProvider, getTargetSplitSize(session), cachingHostAddressProvider);
- this.lazyPartitions = Lazy.lazily(() -> requireNonNull(lazyPartitionMap, "partitions is null").get().keySet().stream().toList());
+ this.lazyPartitionMap = requireNonNull(lazyPartitionMap, "partitions is null");
this.enableMetadataTable = enableMetadataTable;
this.executor = requireNonNull(executor, "executor is null");
this.errorListener = requireNonNull(errorListener, "errorListener is null");
SchemaTableName schemaTableName = tableHandle.getSchemaTableName();
+ this.lazyTableMetadata = lazyTableMetadata;
+ this.lazyMetaClient = Lazy.lazily(tableHandle::getMetaClient);
this.partitionIndexSupportOpt = enableMetadataTable ?
- IndexSupportFactory.createPartitionStatsIndexSupport(schemaTableName, Lazy.lazily(tableHandle::getMetaClient), lazyTableMetadata, tableHandle.getRegularPredicates(), session) : Optional.empty();
+ IndexSupportFactory.createPartitionStatsIndexSupport(schemaTableName, lazyMetaClient, lazyTableMetadata, tableHandle.getRegularPredicates(), session) : Optional.empty();
+ this.isMetadataPartitionListingEnabled = isMetadataPartitionListingEnabled(session);
}
@Override
@@ -113,14 +144,16 @@ public class HudiBackgroundSplitLoader
private void generateSplits(boolean useIndex)
{
// Attempt to apply partition pruning using partition stats index
- Optional<List<String>> effectivePartitionsOpt = useIndex && partitionIndexSupportOpt.isPresent() ?
- partitionIndexSupportOpt.get().prunePartitions(lazyPartitions.get()) : Optional.empty();
+ Deque<HiveHudiPartitionInfo> partitionQueue = getPartitionInfos(useIndex);
+ if (partitionQueue.isEmpty()) {
+ asyncQueue.finish();
+ return;
+ }
- Deque<String> partitionQueue = new ConcurrentLinkedDeque<>(effectivePartitionsOpt.orElse(lazyPartitions.get()));
List<HudiPartitionInfoLoader> splitGenerators = new ArrayList<>();
List<ListenableFuture<Void>> futures = new ArrayList<>();
- int splitGeneratorParallelism = Math.min(splitGeneratorNumThreads, lazyPartitions.get().size());
+ int splitGeneratorParallelism = Math.clamp(splitGeneratorNumThreads, 1, partitionQueue.size());
Executor splitGeneratorExecutor = new BoundedExecutor(executor, splitGeneratorParallelism);
for (int i = 0; i < splitGeneratorParallelism; i++) {
@@ -149,4 +182,173 @@ public class HudiBackgroundSplitLoader
throw new TrinoException(HUDI_CANNOT_OPEN_SPLIT, "Error generating Hudi split", e);
}
}
+
+ private Deque<HiveHudiPartitionInfo> getPartitionInfos(boolean useIndex)
+ {
+ Map<String, Partition> metadataPartitions;
+ if (enableMetadataTable && isMetadataPartitionListingEnabled) {
+ try {
+ PartitionValueExtractor partitionValueExtractor = getPartitionValueExtractor(lazyMetaClient.get().getTableConfig());
+ List<HiveColumnHandle> hivePartitionColumns = tableHandle.getPartitionColumns();
+
+ List<Column> partitionColumns = hivePartitionColumns.stream()
+ .map(column -> new Column(
+ column.getName(),
+ column.getHiveType(),
+ column.getComment(),
+ Collections.emptyMap()))
+ .toList();
+ log.info("Listing partitions for %s.%s via metadata table using partition value extractor %s",
+ tableHandle.getSchemaName(), tableHandle.getTableName(), partitionValueExtractor.getClass().getSimpleName());
+ metadataPartitions = lazyTableMetadata.get()
+ .getAllPartitionPaths().stream()
+ .map(partitionPath -> buildPartition(partitionPath, partitionColumns, tableHandle, partitionValueExtractor))
+ .collect(Collectors.toMap(
+ this::getHivePartitionName,
+ Function.identity()));
+
+ if (metadataPartitions.isEmpty()) {
+ log.warn("No partitions found via metadata table for %s.%s, switching to metastore-based listing.",
+ tableHandle.getSchemaName(), tableHandle.getTableName());
+ metadataPartitions = lazyPartitionMap.get();
+ }
+ }
+ catch (Exception e) {
+ log.error(e, "Failed to get partitions from metadata table %s.%s, falling back to metastore based partition listing",
+ tableHandle.getSchemaName(), tableHandle.getTableName());
+ metadataPartitions = lazyPartitionMap.get();
+ }
+ }
+ else {
+ metadataPartitions = lazyPartitionMap.get();
+ }
+
+ List<String> allPartitions = new ArrayList<>(metadataPartitions.keySet());
+
+ List<String> effectivePartitions = Optional.ofNullable(useIndex && partitionIndexSupportOpt.isPresent()
+ ? partitionIndexSupportOpt.get().prunePartitions(allPartitions).orElse(null)
+ : null).orElse(allPartitions);
+
+ Map<String, Partition> finalMetadataPartitions = metadataPartitions;
+ List<HiveHudiPartitionInfo> hiveHudiPartitionInfos = effectivePartitions.stream()
+ .map(partitionName -> buildHiveHudiPartitionInfo(tableHandle, partitionName, finalMetadataPartitions.get(partitionName)))
+ .filter(hudiPartitionInfo -> hudiPartitionInfo.doesMatchPredicates() || hudiPartitionInfo.getHivePartitionName().equals(NON_PARTITION))
+ .toList();
+
+ return new ConcurrentLinkedDeque<>(hiveHudiPartitionInfos);
+ }
+
+ private HiveHudiPartitionInfo buildHiveHudiPartitionInfo(HudiTableHandle tableHandle, String partitionName, Partition partition)
+ {
+ return new HiveHudiPartitionInfo(
+ tableHandle.getSchemaTableName(),
+ Location.of(tableHandle.getBasePath()),
+ partitionName,
+ partition,
+ tableHandle.getPartitionColumns(),
+ tableHandle.getPartitionPredicates());
+ }
+
+ private Partition buildPartition(String partitionPath, List<Column> partitionColumns, HudiTableHandle tableHandle, PartitionValueExtractor partitionValueExtractor)
+ {
+ if (partitionPath == null || partitionPath.isEmpty()) {
+ return Partition.builder()
+ .setDatabaseName(tableHandle.getSchemaName())
+ .setTableName(tableHandle.getTableName())
+ .withStorage(storageBuilder ->
+ storageBuilder.setLocation(tableHandle.getBasePath())
+ .setStorageFormat(StorageFormat.NULL_STORAGE_FORMAT))
+ .setColumns(ImmutableList.of())
+ .setValues(ImmutableList.of())
+ .build();
+ }
+
+ List<String> values = partitionValueExtractor.extractPartitionValuesInPath(partitionPath).stream().map(this::unescapePathName).toList();
+ if (partitionColumns.size() != values.size()) {
+ throw new HoodieException("Cannot extract partition values from partition path: " + partitionPath);
+ }
+
+ return Partition.builder()
+ .setDatabaseName(tableHandle.getSchemaName())
+ .setTableName(tableHandle.getTableName())
+ .withStorage(storageBuilder ->
+ storageBuilder.setLocation(getFullPath(tableHandle.getBasePath(), partitionPath))
+ // Storage format is unused by the Hudi connector
+ .setStorageFormat(StorageFormat.NULL_STORAGE_FORMAT))
+ .setColumns(partitionColumns)
+ .setValues(values)
+ .build();
+ }
+
+ private String getFullPath(String basePath, String relativePartitionPath)
+ {
+ return basePath.endsWith(DELIMITER_STR)
+ ? basePath + relativePartitionPath
+ : basePath + DELIMITER_STR + relativePartitionPath;
+ }
+
+ String getHivePartitionName(Partition partition)
+ {
+ List<Column> columns = partition.getColumns();
+ List<String> values = partition.getValues();
+
+ return IntStream.range(0, columns.size())
+ .mapToObj(i -> columns.get(i).getName() + EQUALS_STR + values.get(i))
+ .collect(Collectors.joining(DELIMITER_STR));
+ }
+
+ private PartitionValueExtractor getPartitionValueExtractor(HoodieTableConfig tableConfig)
+ {
+ Option<String[]> partitionFieldsOpt = tableConfig.getPartitionFields();
+
+ if (partitionFieldsOpt.isEmpty()) {
+ return new NonPartitionedExtractor();
+ }
+
+ String[] partitionFields = partitionFieldsOpt.get();
+ if (partitionFields.length == 1) {
+ return Boolean.parseBoolean(tableConfig.getHiveStylePartitioningEnable()) ?
+ new HiveStylePartitionValueExtractor() : new SinglePartPartitionValueExtractor();
+ }
+
+ return new MultiPartKeysValueExtractor();
+ }
+
+ private String unescapePathName(String path)
+ {
+ // fast path, no escaped characters and therefore no copying necessary
+ int escapedAtIndex = path.indexOf('%');
+ if (escapedAtIndex < 0 || escapedAtIndex + 2 >= path.length()) {
+ return path;
+ }
+
+ // slow path, unescape into a new string copy
+ StringBuilder sb = new StringBuilder();
+ int fromIndex = 0;
+ while (escapedAtIndex >= 0 && escapedAtIndex + 2 < path.length()) {
+ // preceding sequence without escaped characters
+ if (escapedAtIndex > fromIndex) {
+ sb.append(path, fromIndex, escapedAtIndex);
+ }
+ // try to parse the to digits after the percent sign as hex
+ try {
+ int code = HexFormat.fromHexDigits(path, escapedAtIndex + 1, escapedAtIndex + 3);
+ sb.append((char) code);
+ // advance past the percent sign and both hex digits
+ fromIndex = escapedAtIndex + 3;
+ }
+ catch (NumberFormatException e) {
+ // invalid escape sequence, only advance past the percent sign
+ sb.append('%');
+ fromIndex = escapedAtIndex + 1;
+ }
+ // find next escaped character
+ escapedAtIndex = path.indexOf('%', fromIndex);
+ }
+ // trailing sequence without escaped characters
+ if (fromIndex < path.length()) {
+ sb.append(path, fromIndex, path.length());
+ }
+ return sb.toString();
+ }
}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
index 81254ca23f72..1308cff269a9 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
@@ -82,9 +82,9 @@ public class TestHudiAlluxioCacheFileOperations
query,
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("Alluxio.readCached", DATA), 2)
- .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 24)
- .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
- .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 8)
+ .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 27)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
+ .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 10)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
.add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
@@ -94,9 +94,9 @@ public class TestHudiAlluxioCacheFileOperations
query,
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("Alluxio.readCached", DATA), 2)
- .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 24)
- .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
- .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 8)
+ .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 27)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
+ .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 10)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
.add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
@@ -115,9 +115,9 @@ public class TestHudiAlluxioCacheFileOperations
assertFileSystemAccesses(query,
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("Alluxio.readCached", DATA), 6)
- .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 282)
- .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 37)
- .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 89)
+ .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 288)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 39)
+ .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 93)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 5)
.addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 3)
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 5)
@@ -126,9 +126,9 @@ public class TestHudiAlluxioCacheFileOperations
assertFileSystemAccesses(query,
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("Alluxio.readCached", DATA), 6)
- .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 209)
- .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 27)
- .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 65)
+ .addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 215)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 29)
+ .addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 69)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
.addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 4)
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
index 22f6e6609106..7e605a310a98 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
@@ -57,6 +57,7 @@ public class TestHudiConfig
.setColumnStatsWaitTimeout(Duration.valueOf("1s"))
.setRecordIndexWaitTimeout(Duration.valueOf("2s"))
.setSecondaryIndexWaitTimeout(Duration.valueOf("2s"))
+ .setMetadataPartitionListingEnabled(true)
.setMetadataCacheEnabled(true));
}
@@ -90,6 +91,7 @@ public class TestHudiConfig
.put("hudi.index.record-index.wait-timeout", "1s")
.put("hudi.index.secondary-index.wait-timeout", "1s")
.put("hudi.metadata.cache.enabled", "false")
+ .put("hudi.metadata.partition-listing.enabled", "false")
.buildOrThrow();
HudiConfig expected = new HudiConfig()
@@ -118,6 +120,7 @@ public class TestHudiConfig
.setColumnStatsWaitTimeout(Duration.valueOf("2s"))
.setRecordIndexWaitTimeout(Duration.valueOf("1s"))
.setSecondaryIndexWaitTimeout(Duration.valueOf("1s"))
+ .setMetadataPartitionListingEnabled(false)
.setMetadataCacheEnabled(false);
assertFullMapping(properties, expected);
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
index 2df228e7bb9e..3f6d97f16919 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
@@ -72,9 +72,9 @@ public class TestHudiMemoryCacheFileOperations
query,
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 2)
- .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 3)
- .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 5)
- .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
+ .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 4)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 6)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
.add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
@@ -84,9 +84,9 @@ public class TestHudiMemoryCacheFileOperations
query,
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 2)
- .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 3)
- .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 5)
- .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
+ .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 4)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 6)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
.add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
@@ -105,9 +105,9 @@ public class TestHudiMemoryCacheFileOperations
assertFileSystemAccesses(query,
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 6)
- .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 37)
- .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 52)
- .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 37)
+ .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 39)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 54)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 39)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 5)
.addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 3)
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 5)
@@ -116,9 +116,9 @@ public class TestHudiMemoryCacheFileOperations
assertFileSystemAccesses(query,
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 6)
- .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 27)
- .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 38)
- .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 27)
+ .addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 29)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 40)
+ .addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 29)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
.addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 4)
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
index b91a4b91c892..416914d3f7e2 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
@@ -72,9 +72,9 @@ public class TestHudiNoCacheFileOperations
query,
ImmutableMultiset.<FileOperationUtils.FileOperation>builder()
.addCopies(new FileOperationUtils.FileOperation("Input.readTail", DATA), 2)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 3)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 5)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 4)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 6)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
.add(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
@@ -84,10 +84,10 @@ public class TestHudiNoCacheFileOperations
query,
ImmutableMultiset.<FileOperationUtils.FileOperation>builder()
.addCopies(new FileOperationUtils.FileOperation("Input.readTail", DATA), 2)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 3)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 3)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 4)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 5)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 6)
.add(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
.build());
@@ -105,10 +105,10 @@ public class TestHudiNoCacheFileOperations
assertFileSystemAccesses(query,
ImmutableMultiset.<FileOperationUtils.FileOperation>builder()
.addCopies(new FileOperationUtils.FileOperation("Input.readTail", DATA), 6)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 37)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 37)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 39)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 39)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", INDEX_DEFINITION), 5)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 52)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 54)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 3)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 5)
.build());
@@ -116,10 +116,10 @@ public class TestHudiNoCacheFileOperations
assertFileSystemAccesses(query,
ImmutableMultiset.<FileOperationUtils.FileOperation>builder()
.addCopies(new FileOperationUtils.FileOperation("Input.readTail", DATA), 6)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 27)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 27)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.lastModified", METADATA_TABLE), 29)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 29)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
- .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 38)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 40)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 4)
.build());
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
index fddf996c644f..a8f90f826d05 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
@@ -75,6 +75,7 @@ import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.Testing
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COW_PT_TBL;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_CUSTOM_KEYGEN_PT_V8_MOR;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_PT_V8_MOR;
+import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_NON_EXTRACTABLE_PARTITION_PATH;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_NON_PART_COW;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_STOCK_TICKS_COW;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_STOCK_TICKS_MOR;
@@ -131,6 +132,20 @@ public class TestHudiSmokeTest
"SELECT * FROM VALUES ('2018-08-31', '99')");
}
+ @Test
+ public void testReadNonExtractablePartitionPathTable()
+ {
+ Session session = SessionBuilder.from(getSession())
+ .withMdtEnabled(true)
+ .build();
+ String res = getQueryRunner().execute(session, "SELECT * FROM " + HUDI_NON_EXTRACTABLE_PARTITION_PATH).toString();
+ System.out.println(res);
+ assertQuery(session, "SELECT name FROM " + HUDI_NON_EXTRACTABLE_PARTITION_PATH + " where dt='2018-10-05'",
+ "SELECT * FROM VALUES ('Alice'), ('Bob')");
+ assertQuery(session, "SELECT name FROM " + HUDI_NON_EXTRACTABLE_PARTITION_PATH + " where dt='2018-10-05' and hh='10'",
+ "SELECT * FROM VALUES ('Alice'), ('Bob')");
+ }
+
@Test
public void testReadPartitionedCOWTableVer8()
{
@@ -142,6 +157,11 @@ public class TestHudiSmokeTest
"SELECT * FROM VALUES ('2018-08-31', '99')");
}
+ // Verifies query results on a partitioned COW table (Ver8).
+ // Since the table defines only a single partition column, its partition
+ // values can be correctly parsed by Hudi's PartitionValueExtractor.
+ // This test asserts that grouping and aggregations on the partition column
+ // return the expected results.
@Test
public void testReadPartitionedMORTableVer8()
{
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
index 5783da409e56..da8e7e19c15c 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
@@ -347,6 +347,7 @@ public class ResourceHudiTablesInitializer
HUDI_TIMESTAMP_KEYGEN_PT_EPOCH_TO_YYYY_MM_DD_HH_V8_MOR(hudiTimestampKeygenColumns(), hudiTimestampKeygenPartitionColumns(), hudiTimestampKeygenPartitions("EPOCHMILLISECONDS"), true),
HUDI_TIMESTAMP_KEYGEN_PT_SCALAR_TO_YYYY_MM_DD_HH_V8_MOR(hudiTimestampKeygenColumns(), hudiTimestampKeygenPartitionColumns(), hudiTimestampKeygenPartitions("SCALAR"), true),
HUDI_CUSTOM_KEYGEN_PT_V8_MOR(hudiCustomKeyGenColumns(), hudiCustomKeyGenPartitionColumns(), hudiCustomKeyGenPartitions(), false),
+ HUDI_NON_EXTRACTABLE_PARTITION_PATH(multiPartitionRegularColumns(), multiPartitionColumns(), multiPartitionsWithNonExtractablePartitionPaths(), false),
/**/;
private static final List<Column> HUDI_META_COLUMNS = ImmutableList.of(
@@ -527,6 +528,27 @@ public class ResourceHudiTablesInitializer
"dt=2021-12-09/hh=11", "dt=2021-12-09/hh=11");
}
+ /**
+ * Returns a sample map of partition specs containing multiple partition keys separated by slashes.
+ *
+ * Example:
+ * "dt=2018-10-05/hh=10" -> "2018/10/05/10"
+ *
+ * Note:
+ * - The partition spec has 2 partition keys (dt, hh).
+ * - However, the corresponding value string has 4 segments when split by slashes
+ * (year, month, day, hour).
+ * - Standard Hudi partition extractors will not be able to correctly parse this mapping,
+ * since they expect the number of slash-separated values to match the number of partition keys
+ * if there's more than one partition field.
+ */
+ private static Map<String, String> multiPartitionsWithNonExtractablePartitionPaths()
+ {
+ return ImmutableMap.of(
+ "dt=2018-10-05/hh=10", "2018/10/05/10",
+ "dt=2018-10-06/hh=5", "2018/10/06/5");
+ }
+
private static List<Column> hudiMultiFgRegularColumns()
{
return ImmutableList.of(
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_extractable_partition_path.md b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_extractable_partition_path.md
new file mode 100644
index 000000000000..db288b2dfc23
--- /dev/null
+++ b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_extractable_partition_path.md
@@ -0,0 +1,48 @@
+## Create script
+
+Structure of table:
+- COW table with partition columns that Hudi’s partition extractor is unable to parse.
+- No log files
+
+```scala
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.SaveMode
+import org.apache.spark.sql.Row
+
+val schema = StructType(Seq(
+ StructField("id", LongType, nullable = false),
+ StructField("name", StringType, nullable = true),
+ StructField("ts", LongType, nullable = true),
+ StructField("dt", StringType, nullable = false),
+ StructField("hh", StringType, nullable = false)
+))
+
+
+val data = Seq(
+ Row(1L, "Alice", 1723272000000L, "2018-10-05", "10"),
+ Row(2L, "Bob", 1723358400000L, "2018-10-05", "10"),
+ Row(3L, "Charlie", 1723444800000L, "2018-10-06", "5"),
+ Row(4L, "David", 1723531200000L, "2018-10-07", "5")
+)
+
+val dfRaw = spark.createDataFrame(
+ spark.sparkContext.parallelize(data),
+ schema
+)
+
+
+val df = dfRaw.withColumn("dt", date_format(to_date($"dt", "yyyy-MM-dd"), "yyyy/MM/dd"))
+
+var basePath = "file:///tmp/hudi_non_extractable_partition_path"
+
+df.write
+ .format("hudi")
+ .option("hoodie.table.name", "hudi_non_extractable_partition_path")
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.partitionpath.field", "dt,hh")
+ .option("hoodie.datasource.write.hive_style_partitioning", "false")
+ .option("hoodie.datasource.write.operation", "insert")
+ .mode(SaveMode.Overwrite)
+ .save(basePath)
+```
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_extractable_partition_path.zip b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_extractable_partition_path.zip
new file mode 100644
index 000000000000..c7e6b6f0127b
Binary files /dev/null and b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_extractable_partition_path.zip differ
(hudi) 13/13: [ENG-30507] Incorrect query results for Merge-On-Read (RT) tables when column stats are enabled (#58)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 194e58526b6992b878c13e72fbcf2fcbd0bcc9da
Author: vamsikarnika <va...@gmail.com>
AuthorDate: Mon Aug 25 22:25:43 2025 +0800
[ENG-30507] Incorrect query results for Merge-On-Read (RT) tables when column stats are enabled (#58)
* Co-authored-by: Vamsi <va...@onehouse.ai>
(cherry picked from commit 5e3ebd5b4edd0623e041109cc769f0123bcbd4a7)
---
.../query/index/HudiColumnStatsIndexSupport.java | 23 ++++--
.../io/trino/plugin/hudi/TestHudiSmokeTest.java | 30 ++++++++
.../testing/ResourceHudiTablesInitializer.java | 9 +++
.../hudi-testing-data/hudi_non_part_mor.md | 77 +++++++++++++++++++++
.../hudi-testing-data/hudi_non_part_mor.zip | Bin 0 -> 82188 bytes
5 files changed, 133 insertions(+), 6 deletions(-)
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java
index 9b3291b6fc4b..0d1c9eaa6cd0 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java
@@ -203,13 +203,24 @@ public class HudiColumnStatsIndexSupport
TupleDomain<String> regularColumnPredicates,
List<String> regularColumns)
{
- String fileSliceName = fileSlice.getBaseFile().map(BaseFile::getFileName).orElse("");
- // If no stats exist for this specific file, we cannot prune it.
- if (!domainsWithStats.containsKey(fileSliceName)) {
- return false;
+ List<String> filesToLookUp = new ArrayList<>();
+ fileSlice.getBaseFile()
+ .map(BaseFile::getFileName)
+ .ifPresent(filesToLookUp::add);
+
+ if (fileSlice.hasLogFiles()) {
+ fileSlice.getLogFiles().forEach(logFile -> filesToLookUp.add(logFile.getFileName()));
}
- Map<String, Domain> fileDomainsWithStats = domainsWithStats.get(fileSliceName);
- return !evaluateStatisticPredicate(regularColumnPredicates, fileDomainsWithStats, regularColumns);
+
+ // if any log or base file in the file slice matches the predicate, all files in the file slice needs to be read
+ return filesToLookUp.stream().allMatch(fileName -> {
+ // If no stats exist for this specific file, we cannot prune it.
+ if (!domainsWithStats.containsKey(fileName)) {
+ return false;
+ }
+ Map<String, Domain> fileDomainsWithStats = domainsWithStats.get(fileName);
+ return !evaluateStatisticPredicate(regularColumnPredicates, fileDomainsWithStats, regularColumns);
+ });
}
protected static boolean evaluateStatisticPredicate(
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
index 3c98a85f5b2a..1080d5e89b7b 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
@@ -80,6 +80,7 @@ import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.Testing
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_PT_V8_MOR;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_NON_EXTRACTABLE_PARTITION_PATH;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_NON_PART_COW;
+import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_NON_PART_MOR;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_STOCK_TICKS_COW;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_STOCK_TICKS_MOR;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_TIMESTAMP_KEYGEN_PT_EPOCH_TO_YYYY_MM_DD_HH_V8_MOR;
@@ -684,6 +685,35 @@ public class TestHudiSmokeTest
assertThat(prunedSplits).isEqualTo(1);
}
+ @Test
+ public void testColStatsFileSkippingMORTable()
+ {
+ Session session = SessionBuilder.from(getSession())
+ .withMdtEnabled(true)
+ .withColStatsIndexEnabled(true)
+ .withColumnStatsTimeout("1s")
+ .withRecordLevelIndexEnabled(false)
+ .withSecondaryIndexEnabled(false)
+ .withPartitionStatsIndexEnabled(false)
+ .build();
+ MaterializedResult roTableRes = getQueryRunner().execute(session, "SELECT id, name FROM " + HUDI_NON_PART_MOR + " where name = 'Alice'");
+ MaterializedResult rtTableRes = getQueryRunner().execute(session, "SELECT id, name FROM " + HUDI_NON_PART_MOR + "_rt where name = 'Cathy'");
+
+ // verify ro table returns results from base file
+ int roTableSplits = roTableRes.getStatementStats().get().getTotalSplits();
+ int roTableRows = roTableRes.getRowCount();
+ assertThat(roTableSplits).isEqualTo(1);
+ assertThat(roTableRows).isEqualTo(1);
+ assertThat(roTableRes.getMaterializedRows().getFirst().getField(1)).isEqualTo("Alice");
+
+ // verify rt table returns results from log file
+ int rtTableSplits = rtTableRes.getStatementStats().get().getTotalSplits();
+ int rtTableRows = rtTableRes.getRowCount();
+ assertThat(rtTableRows).isEqualTo(1);
+ assertThat(rtTableSplits).isEqualTo(1);
+ assertThat(rtTableRes.getMaterializedRows().getFirst().getField(1)).isEqualTo("Cathy");
+ }
+
@ParameterizedTest
@EnumSource(
value = ResourceHudiTablesInitializer.TestingTable.class,
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
index 420e10e1ada6..cebbd7a51d2a 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
@@ -333,6 +333,7 @@ public class ResourceHudiTablesInitializer
public enum TestingTable
{
HUDI_NON_PART_COW(nonPartitionRegularColumns()),
+ HUDI_NON_PART_MOR(simpleRegularColumns(), ImmutableList.of(), ImmutableMap.of(), true),
HUDI_TRIPS_COW_V8(tripsRegularColumns()),
HUDI_COW_TABLE_WITH_FIELD_NAMES_IN_CAPS(hudiTableWithFieldNamesInCapsRegularColumns()),
HUDI_COW_PT_TABLE_WITH_FIELD_NAMES_IN_CAPS(hudiTableWithFieldNamesInCapsRegularColumns(), hudiTableWithFieldNamesInCapsPartitionColumns(), hudiTableWithFieldNamesInCapsPartitions(), false), // delete
@@ -468,6 +469,14 @@ public class ResourceHudiTablesInitializer
return ImmutableList.of(column("country", HIVE_STRING));
}
+ private static List<Column> simpleRegularColumns()
+ {
+ return ImmutableList.of(
+ column("id", HIVE_STRING),
+ column("name", HIVE_STRING),
+ column("age", HIVE_INT));
+ }
+
private static List<Column> tripsRegularColumns()
{
return ImmutableList.of(
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_part_mor.md b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_part_mor.md
new file mode 100644
index 000000000000..90b168a74156
--- /dev/null
+++ b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_part_mor.md
@@ -0,0 +1,77 @@
+# Hudi Test Resources
+
+## Generating Hudi Resources
+
+Follow these steps to create the `hudi_non_part_mor` test table and utilize it for testing.
+
+### Download and install spark version 3.3 - 3.5
+### Open spark scala shell
+
+Execute the following command in the terminal to create the table locally:
+
+### Generate Resources
+
+* Open the `spark-shell` terminal using below command
+```
+./spark-shell \
+ --packages org.apache.hudi:hudi-spark3.5-bundle_2.12:1.0.2,org.apache.hadoop:hadoop-aws:3.3.1,com.amazonaws:aws-java-sdk-bundle:1.12.538 \
+ --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
+ --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \
+ --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
+ --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'
+```
+* Execute the following Spark scala code to create the `hudi_non_part_mor` table:
+
+```
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.Row
+
+val schema = StructType(Seq(
+ StructField("id", StringType, nullable = false),
+ StructField("name", StringType, nullable = true),
+ StructField("age", IntegerType, nullable = true)
+))
+
+
+val data = Seq(
+ Row("1", "Alice", 30),
+ Row("2", "Bob", 25)
+)
+
+val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(data),
+ schema
+)
+
+var basePath = "file:///tmp/hudi_non_part_mor"
+
+df.write()
+ .format("hudi")
+ .mode("Append")
+ .option("hoodie.table.name", "hudi_non_part_mor")
+ .option("hoodie.datasource.write.table.type", "MERGE_ON_READ")
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.operation", "bulk_insert")
+ .option("hoodie.metadata.index.column.stats.enable", "true")
+ .save(basePath);
+
+val data = Seq(
+ Row("1", "Cathy", 30),
+ Row("2", "David", 25)
+)
+
+val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(data),
+ schema
+)
+
+df.write
+ .format("hudi")
+ .mode("Append")
+ .option("hoodie.table.name", "hudi_non_part_mor")
+ .option("hoodie.datasource.write.table.type", "MERGE_ON_READ")
+ .option("hoodie.datasource.write.recordkey.field","id")
+ .option("hoodie.datasource.write.operation","upsert")
+ .option("hoodie.metadata.index.column.stats.enable", "true")
+ .save(basePath)
+```
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_part_mor.zip b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_part_mor.zip
new file mode 100644
index 000000000000..521f1583beb4
Binary files /dev/null and b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_non_part_mor.zip differ
(hudi) 11/13: [ENG-30371] Fix Case Sensitivity Issues Between Table and Catalog Schemas (#57)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d1c6b02c0c6eefeb1c482e1694e77965b387d9a3
Author: vamsikarnika <va...@gmail.com>
AuthorDate: Sun Aug 24 23:24:50 2025 +0800
[ENG-30371] Fix Case Sensitivity Issues Between Table and Catalog Schemas (#57)
Co-authored-by: Vamsi <va...@onehouse.ai>
(cherry picked from commit b921dcb45920f4b65cc06ba6e007499355110d3a)
---
hudi-trino-plugin/pom.xml | 5 ++
.../main/java/io/trino/plugin/hudi/HudiConfig.java | 14 +++
.../java/io/trino/plugin/hudi/HudiErrorCode.java | 1 +
.../java/io/trino/plugin/hudi/HudiMetadata.java | 11 ++-
.../trino/plugin/hudi/HudiPageSourceProvider.java | 14 +--
.../trino/plugin/hudi/HudiSessionProperties.java | 13 ++-
.../java/io/trino/plugin/hudi/HudiTableHandle.java | 55 +++++++++++-
.../main/java/io/trino/plugin/hudi/HudiUtil.java | 96 ++++++++++++++++++++-
.../hudi/query/HudiSnapshotDirectoryLister.java | 2 +-
.../query/index/HudiColumnStatsIndexSupport.java | 11 ++-
.../index/HudiPartitionStatsIndexSupport.java | 6 +-
.../query/index/HudiRecordLevelIndexSupport.java | 6 +-
.../query/index/HudiSecondaryIndexSupport.java | 8 +-
.../hudi/query/index/IndexSupportFactory.java | 33 +++++--
.../hudi/split/HudiBackgroundSplitLoader.java | 4 +-
.../trino/plugin/hudi/util/HudiAvroSerializer.java | 3 +-
.../java/io/trino/plugin/hudi/SessionBuilder.java | 6 ++
.../hudi/TestHudiAlluxioCacheFileOperations.java | 18 ++++
.../java/io/trino/plugin/hudi/TestHudiConfig.java | 7 +-
.../hudi/TestHudiMemoryCacheFileOperations.java | 10 +++
.../plugin/hudi/TestHudiNoCacheFileOperations.java | 10 +++
.../io/trino/plugin/hudi/TestHudiSmokeTest.java | 95 ++++++++++++++++++++
.../plugin/hudi/split/TestHudiSplitFactory.java | 15 ++--
.../testing/ResourceHudiTablesInitializer.java | 30 +++++++
.../hudi_cow_pt_table_with_field_names_in_caps.md | 57 ++++++++++++
.../hudi_cow_pt_table_with_field_names_in_caps.zip | Bin 0 -> 139357 bytes
.../hudi_cow_table_with_field_names_in_caps.md | 41 +++++++++
.../hudi_cow_table_with_field_names_in_caps.zip | Bin 0 -> 78817 bytes
...able_with_multi_keys_and_field_names_in_caps.md | 41 +++++++++
...ble_with_multi_keys_and_field_names_in_caps.zip | Bin 0 -> 82439 bytes
.../hudi_mor_table_with_field_names_in_caps.md | 41 +++++++++
.../hudi_mor_table_with_field_names_in_caps.zip | Bin 0 -> 79103 bytes
32 files changed, 592 insertions(+), 61 deletions(-)
diff --git a/hudi-trino-plugin/pom.xml b/hudi-trino-plugin/pom.xml
index 4ff439d1e35c..126bfb48e001 100644
--- a/hudi-trino-plugin/pom.xml
+++ b/hudi-trino-plugin/pom.xml
@@ -73,6 +73,11 @@
<artifactId>units</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.trino</groupId>
+ <artifactId>trino-cache</artifactId>
+ </dependency>
+
<dependency>
<groupId>io.trino</groupId>
<artifactId>trino-filesystem</artifactId>
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
index c0df68088c93..2355744b6976 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
@@ -55,6 +55,7 @@ public class HudiConfig
private boolean queryPartitionFilterRequired;
private boolean ignoreAbsentPartitions;
private Duration dynamicFilteringWaitTimeout = new Duration(1, SECONDS);
+ private boolean resolveColumnNameCasingEnabled = true;
// Internal configuration for debugging and testing
private boolean isRecordLevelIndexEnabled = true;
@@ -435,4 +436,17 @@ public class HudiConfig
this.metadataPartitionListingEnabled = metadataPartitionListingEnabled;
return this;
}
+
+ public boolean isResolveColumnNameCasingEnabled()
+ {
+ return resolveColumnNameCasingEnabled;
+ }
+
+ @Config("hudi.table.resolve-column-name-casing.enabled")
+ @ConfigDescription("Reconcile column names between the catalog schema and the Hudi table to handle case differences")
+ public HudiConfig setResolveColumnNameCasingEnabled(boolean resolveColumnNameCasingEnabled)
+ {
+ this.resolveColumnNameCasingEnabled = resolveColumnNameCasingEnabled;
+ return this;
+ }
}
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiErrorCode.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiErrorCode.java
index 6b34fedc8853..f9dca132d17c 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiErrorCode.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiErrorCode.java
@@ -34,6 +34,7 @@ public enum HudiErrorCode
HUDI_UNSUPPORTED_TABLE_TYPE(9, EXTERNAL),
HUDI_NO_VALID_COMMIT(10, EXTERNAL),
HUDI_META_CLIENT_ERROR(11, EXTERNAL),
+ HUDI_SCHEMA_ERROR(12, EXTERNAL)
/**/;
private final ErrorCode errorCode;
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiMetadata.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiMetadata.java
index bbe5564b1d15..02df12bab2dd 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiMetadata.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiMetadata.java
@@ -47,6 +47,7 @@ import io.trino.spi.predicate.TupleDomain;
import io.trino.spi.statistics.Estimate;
import io.trino.spi.statistics.TableStatistics;
import io.trino.spi.type.TypeManager;
+import org.apache.avro.Schema;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
@@ -84,10 +85,12 @@ import static io.trino.plugin.hive.util.HiveUtil.isHudiTable;
import static io.trino.plugin.hudi.HudiSessionProperties.getColumnsToHide;
import static io.trino.plugin.hudi.HudiSessionProperties.isHudiMetadataTableEnabled;
import static io.trino.plugin.hudi.HudiSessionProperties.isQueryPartitionFilterRequired;
+import static io.trino.plugin.hudi.HudiSessionProperties.isResolveColumnNameCasingEnabled;
import static io.trino.plugin.hudi.HudiSessionProperties.isTableStatisticsEnabled;
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.buildTableMetaClient;
+import static io.trino.plugin.hudi.HudiUtil.getLatestTableSchema;
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;
@@ -158,10 +161,13 @@ public class HudiMetadata
TrinoFileSystem fileSystem = fileSystemFactory.create(session);
String inputFormat = table.getStorage().getStorageFormat().getInputFormat();
HoodieTableType hoodieTableType = HudiTableTypeUtils.fromInputFormat(inputFormat);
+ Lazy<HoodieTableMetaClient> lazyMetaClient = Lazy.lazily(() -> buildTableMetaClient(fileSystem, tableName.toString(), basePath));
+ Optional<Lazy<Schema>> hudiTableSchema = isResolveColumnNameCasingEnabled(session) ?
+ Optional.of(Lazy.lazily(() -> getLatestTableSchema(lazyMetaClient.get(), tableName.getTableName()))) : Optional.empty();
return new HudiTableHandle(
Optional.of(table),
- Optional.of(Lazy.lazily(() -> buildTableMetaClient(fileSystem, tableName.toString(), basePath))),
+ Optional.of(lazyMetaClient),
tableName.getSchemaName(),
tableName.getTableName(),
table.getStorage().getLocation(),
@@ -169,7 +175,8 @@ public class HudiMetadata
getPartitionKeyColumnHandles(table, typeManager),
ImmutableSet.of(),
TupleDomain.all(),
- TupleDomain.all());
+ TupleDomain.all(),
+ hudiTableSchema);
}
@Override
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
index 500934044b37..7d3812e2d71b 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
@@ -56,7 +56,6 @@ import org.apache.avro.Schema;
import org.apache.avro.generic.IndexedRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.common.table.TableSchemaResolver;
import org.apache.hudi.common.table.read.HoodieFileGroupReader;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
@@ -93,7 +92,6 @@ import static io.trino.plugin.hive.parquet.ParquetPageSourceFactory.getParquetTu
import static io.trino.plugin.hudi.HudiErrorCode.HUDI_BAD_DATA;
import static io.trino.plugin.hudi.HudiErrorCode.HUDI_CANNOT_OPEN_SPLIT;
import static io.trino.plugin.hudi.HudiErrorCode.HUDI_CURSOR_ERROR;
-import static io.trino.plugin.hudi.HudiErrorCode.HUDI_FILESYSTEM_ERROR;
import static io.trino.plugin.hudi.HudiSessionProperties.getParquetMaxReadBlockRowCount;
import static io.trino.plugin.hudi.HudiSessionProperties.getParquetMaxReadBlockSize;
import static io.trino.plugin.hudi.HudiSessionProperties.getParquetSmallFileThreshold;
@@ -105,6 +103,7 @@ import static io.trino.plugin.hudi.HudiSessionProperties.useParquetBloomFilter;
import static io.trino.plugin.hudi.HudiUtil.buildTableMetaClient;
import static io.trino.plugin.hudi.HudiUtil.constructSchema;
import static io.trino.plugin.hudi.HudiUtil.convertToFileSlice;
+import static io.trino.plugin.hudi.HudiUtil.getLatestTableSchema;
import static io.trino.plugin.hudi.HudiUtil.prependHudiMetaColumns;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
@@ -221,20 +220,15 @@ public class HudiPageSourceProvider
HoodieTableMetaClient metaClient = buildTableMetaClient(
fileSystemFactory.create(session), hudiTableHandle.getSchemaTableName().toString(), hudiTableHandle.getBasePath());
String latestCommitTime = metaClient.getCommitsTimeline().lastInstant().get().requestedTime();
- Schema dataSchema;
- try {
- dataSchema = new TableSchemaResolver(metaClient).getTableAvroSchema(latestCommitTime);
- }
- catch (Throwable e) {
- // Unable to find table schema
- throw new TrinoException(HUDI_FILESYSTEM_ERROR, e);
- }
HudiTrinoReaderContext readerContext = new HudiTrinoReaderContext(
dataPageSource,
dataColumnHandles,
hudiMetaAndDataColumnHandles,
synthesizedColumnHandler);
+ Schema dataSchema =
+ Optional.ofNullable(hudiTableHandle.getTableSchema())
+ .orElseGet(() -> getLatestTableSchema(metaClient, hudiTableHandle.getTableName()));
// Construct an Avro schema for log file reader
Schema requestedSchema = constructSchema(dataSchema, hudiMetaAndDataColumnHandles.stream().map(HiveColumnHandle::getName).toList());
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSessionProperties.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSessionProperties.java
index 2ad9da482806..e3645e0582c2 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSessionProperties.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSessionProperties.java
@@ -64,6 +64,7 @@ public class HudiSessionProperties
static final String QUERY_PARTITION_FILTER_REQUIRED = "query_partition_filter_required";
private static final String IGNORE_ABSENT_PARTITIONS = "ignore_absent_partitions";
static final String DYNAMIC_FILTERING_WAIT_TIMEOUT = "dynamic_filtering_wait_timeout";
+ static final String RESOLVE_COLUMN_NAME_CASING_ENABLED = "resolve_column_name_casing_enabled";
// Internal configuration for debugging and testing
static final String RECORD_LEVEL_INDEX_ENABLED = "record_level_index_enabled";
@@ -244,7 +245,12 @@ public class HudiSessionProperties
METADATA_PARTITION_LISTING_ENABLED,
"Enable metadata table based partition listing",
hudiConfig.isMetadataPartitionListingEnabled(),
- false));
+ false),
+ booleanProperty(
+ RESOLVE_COLUMN_NAME_CASING_ENABLED,
+ "Enable resolve column name casing",
+ hudiConfig.isResolveColumnNameCasingEnabled(),
+ true));
}
@Override
@@ -379,6 +385,11 @@ public class HudiSessionProperties
return !isRecordLevelIndexEnabled(session) && !isSecondaryIndexEnabled(session) && !isColumnStatsIndexEnabled(session);
}
+ public static boolean isResolveColumnNameCasingEnabled(ConnectorSession session)
+ {
+ return session.getProperty(RESOLVE_COLUMN_NAME_CASING_ENABLED, Boolean.class);
+ }
+
public static Duration getDynamicFilteringWaitTimeout(ConnectorSession session)
{
return session.getProperty(DYNAMIC_FILTERING_WAIT_TIMEOUT, Duration.class);
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiTableHandle.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiTableHandle.java
index d6a152446ef0..6ff4198c5646 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiTableHandle.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiTableHandle.java
@@ -17,15 +17,18 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableSet;
+import io.airlift.log.Logger;
import io.trino.metastore.Table;
import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.spi.TrinoException;
import io.trino.spi.connector.ConnectorTableHandle;
import io.trino.spi.connector.SchemaTableName;
import io.trino.spi.predicate.TupleDomain;
+import org.apache.avro.Schema;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.util.Lazy;
import java.util.List;
@@ -39,6 +42,7 @@ import static java.util.Objects.requireNonNull;
public class HudiTableHandle
implements ConnectorTableHandle
{
+ private static final Logger log = Logger.get(HudiTableHandle.class);
private final String schemaName;
private final String tableName;
private final String basePath;
@@ -48,6 +52,7 @@ public class HudiTableHandle
private final Set<HiveColumnHandle> constraintColumns;
private final TupleDomain<HiveColumnHandle> partitionPredicates;
private final TupleDomain<HiveColumnHandle> regularPredicates;
+ private final Optional<Lazy<Schema>> hudiTableSchema;
// Coordinator-only
private final transient Optional<Table> table;
private final transient Optional<Lazy<HoodieTableMetaClient>> lazyMetaClient;
@@ -61,9 +66,11 @@ public class HudiTableHandle
@JsonProperty("tableType") HoodieTableType tableType,
@JsonProperty("partitionColumns") List<HiveColumnHandle> partitionColumns,
@JsonProperty("partitionPredicates") TupleDomain<HiveColumnHandle> partitionPredicates,
- @JsonProperty("regularPredicates") TupleDomain<HiveColumnHandle> regularPredicates)
+ @JsonProperty("regularPredicates") TupleDomain<HiveColumnHandle> regularPredicates,
+ @JsonProperty("tableSchemaStr") String tableSchemaStr)
{
- this(Optional.empty(), Optional.empty(), schemaName, tableName, basePath, tableType, partitionColumns, ImmutableSet.of(), partitionPredicates, regularPredicates);
+ this(Optional.empty(), Optional.empty(), schemaName, tableName, basePath, tableType, partitionColumns, ImmutableSet.of(),
+ partitionPredicates, regularPredicates, buildTableSchema(tableSchemaStr));
}
public HudiTableHandle(
@@ -76,7 +83,8 @@ public class HudiTableHandle
List<HiveColumnHandle> partitionColumns,
Set<HiveColumnHandle> constraintColumns,
TupleDomain<HiveColumnHandle> partitionPredicates,
- TupleDomain<HiveColumnHandle> regularPredicates)
+ TupleDomain<HiveColumnHandle> regularPredicates,
+ Optional<Lazy<Schema>> hudiTableSchema)
{
this.table = requireNonNull(table, "table is null");
this.lazyMetaClient = requireNonNull(lazyMetaClient, "lazyMetaClient is null");
@@ -95,6 +103,29 @@ public class HudiTableHandle
this.constraintColumns = requireNonNull(constraintColumns, "constraintColumns is null");
this.partitionPredicates = requireNonNull(partitionPredicates, "partitionPredicates is null");
this.regularPredicates = requireNonNull(regularPredicates, "regularPredicates is null");
+ this.hudiTableSchema = requireNonNull(hudiTableSchema, "hudiTableSchema is null");
+ }
+
+ /**
+ * Builds a lazily-parsed Avro schema from the given schema string.
+ * <p>
+ * Returns {@code Optional.empty()} if the input string is null/empty
+ * or if parsing the schema fails.
+ */
+ private static Optional<Lazy<Schema>> buildTableSchema(String tableSchemaStr)
+ {
+ if (StringUtils.isNullOrEmpty(tableSchemaStr)) {
+ return Optional.empty();
+ }
+
+ try {
+ Lazy<Schema> lazySchema = Lazy.lazily(() -> new Schema.Parser().parse(tableSchemaStr));
+ return Optional.of(lazySchema);
+ }
+ catch (Exception e) {
+ log.warn(e, "Failed to parse table schema: %s", tableSchemaStr);
+ return Optional.empty();
+ }
}
public Table getTable()
@@ -157,6 +188,21 @@ public class HudiTableHandle
return partitionColumns;
}
+ @JsonProperty
+ public String getTableSchemaStr()
+ {
+ return hudiTableSchema
+ .map(Lazy::get)
+ .map(Schema::toString)
+ .orElse("");
+ }
+
+ @JsonIgnore
+ public Schema getTableSchema()
+ {
+ return hudiTableSchema.map(Lazy::get).orElse(null);
+ }
+
// do not serialize constraint columns as they are not needed on workers
@JsonIgnore
public Set<HiveColumnHandle> getConstraintColumns()
@@ -190,7 +236,8 @@ public class HudiTableHandle
partitionColumns,
constraintColumns,
partitionPredicates.intersect(partitionTupleDomain),
- regularPredicates.intersect(regularTupleDomain));
+ regularPredicates.intersect(regularTupleDomain),
+ hudiTableSchema);
}
@Override
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiUtil.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiUtil.java
index 0f61ceb1beb4..a0be987f4753 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiUtil.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiUtil.java
@@ -13,8 +13,12 @@
*/
package io.trino.plugin.hudi;
+import com.google.common.cache.Cache;
+import com.google.common.cache.Weigher;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import io.airlift.log.Logger;
+import io.trino.cache.EvictableCacheBuilder;
import io.trino.filesystem.FileIterator;
import io.trino.filesystem.Location;
import io.trino.filesystem.TrinoFileSystem;
@@ -41,7 +45,9 @@ import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.exception.TableNotFoundException;
import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.storage.StoragePath;
@@ -51,12 +57,16 @@ import java.io.UncheckedIOException;
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 java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import static io.airlift.slice.SizeOf.estimatedSizeOf;
import static io.trino.plugin.hive.HiveErrorCode.HIVE_INVALID_METADATA;
import static io.trino.plugin.hive.util.HiveUtil.checkCondition;
import static io.trino.plugin.hive.util.SerdeConstants.LIST_COLUMNS;
@@ -64,11 +74,31 @@ import static io.trino.plugin.hive.util.SerdeConstants.LIST_COLUMN_TYPES;
import static io.trino.plugin.hudi.HudiErrorCode.HUDI_BAD_DATA;
import static io.trino.plugin.hudi.HudiErrorCode.HUDI_FILESYSTEM_ERROR;
import static io.trino.plugin.hudi.HudiErrorCode.HUDI_META_CLIENT_ERROR;
+import static io.trino.plugin.hudi.HudiErrorCode.HUDI_SCHEMA_ERROR;
import static io.trino.plugin.hudi.HudiErrorCode.HUDI_UNSUPPORTED_FILE_FORMAT;
+import static java.lang.Math.toIntExact;
import static org.apache.hudi.common.model.HoodieRecord.HOODIE_META_COLUMNS;
public final class HudiUtil
{
+ private static final Logger log = Logger.get(HudiUtil.class);
+ private static final Cache<Schema, Map<String, Schema.Field>> SCHEMA_FIELD_CACHE =
+ EvictableCacheBuilder.newBuilder()
+ .maximumWeight(10L * 1000L * 1024L) // 10MB
+ .weigher((Weigher<Schema, Map<String, Schema.Field>>) (schema, fieldMap) -> {
+ // approximate size estimation of schema size
+ long schemaSize = estimatedSizeOf(schema.toString());
+
+ long fieldsSize = fieldMap.entrySet().stream()
+ .mapToLong(e -> estimatedSizeOf(e.getKey()) + estimatedSizeOf(e.getValue().name()))
+ .sum();
+
+ return toIntExact(schemaSize + fieldsSize);
+ })
+ .expireAfterWrite(5, TimeUnit.MINUTES)
+ .shareNothingWhenDisabled()
+ .build();
+
private HudiUtil() {}
public static HoodieFileFormat getHudiFileFormat(String path)
@@ -203,7 +233,9 @@ public final class HudiUtil
SchemaBuilder.RecordBuilder<Schema> schemaBuilder = SchemaBuilder.record("baseRecord");
SchemaBuilder.FieldAssembler<Schema> fieldBuilder = schemaBuilder.fields();
for (String columnName : columnNames) {
- Schema originalFieldSchema = dataSchema.getField(columnName).schema();
+ Schema.Field field = getFieldFromSchema(columnName, dataSchema);
+ Schema originalFieldSchema = field.schema();
+
Schema typeForNewField;
// Check if the original field schema is already nullable (i.e., a UNION containing NULL)
@@ -215,13 +247,59 @@ public final class HudiUtil
}
fieldBuilder = fieldBuilder
- .name(columnName)
+ .name(field.name())
.type(typeForNewField)
.withDefault(null);
}
return fieldBuilder.endRecord();
}
+ private static Map<String, Schema.Field> buildFieldLookup(Schema schema)
+ {
+ return schema.getFields().stream()
+ .collect(Collectors.toMap(
+ f -> f.name().toLowerCase(Locale.ROOT),
+ f -> f));
+ }
+
+ /**
+ * Retrieves a field from the given Avro schema by column name.
+ * <p>
+ * The lookup proceeds in two steps:
+ * <ul>
+ * <li>First, attempts an exact match on the column name.</li>
+ * <li>If not found, falls back to a case-insensitive match using a cached lookup table</li>
+ * </ul>
+ * <p>
+ *
+ * @param columnName Column name to search for.
+ * @param schema Avro {@link Schema} in which to search.
+ * @return The matching {@link Schema.Field}, if found.
+ * @throws TrinoException if no field matches the given column name.
+ */
+ public static Schema.Field getFieldFromSchema(String columnName, Schema schema)
+ {
+ Schema.Field field = schema.getField(columnName);
+ if (field != null) {
+ return field;
+ }
+
+ try {
+ field = SCHEMA_FIELD_CACHE
+ .get(schema, () -> buildFieldLookup(schema)).get(columnName.toLowerCase(Locale.ROOT));
+ if (field != null) {
+ return field;
+ }
+ }
+ catch (ExecutionException e) {
+ throw new TrinoException(HUDI_SCHEMA_ERROR,
+ "Failed to build field lookup for schema", e);
+ }
+
+ throw new TrinoException(HUDI_SCHEMA_ERROR,
+ "Failed to get column " + columnName + " from table schema");
+ }
+
public static List<HiveColumnHandle> prependHudiMetaColumns(List<HiveColumnHandle> dataColumns)
{
//For efficient lookup
@@ -284,4 +362,18 @@ public final class HudiUtil
return new HoodieTableFileSystemView(
tableMetadata, metaClient, metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants());
}
+
+ public static Schema getLatestTableSchema(HoodieTableMetaClient metaClient, String tableName)
+ {
+ try {
+ HoodieTimer timer = HoodieTimer.start();
+ Schema schema = new TableSchemaResolver(metaClient).getTableAvroSchema();
+ log.info("Fetched table schema for table %s in %s ms", tableName, timer.endTimer());
+ return schema;
+ }
+ catch (Exception e) {
+ // failed to read schema
+ throw new TrinoException(HUDI_FILESYSTEM_ERROR, e);
+ }
+ }
}
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiSnapshotDirectoryLister.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiSnapshotDirectoryLister.java
index 7e235578620b..f177282bf439 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiSnapshotDirectoryLister.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/HudiSnapshotDirectoryLister.java
@@ -64,7 +64,7 @@ public class HudiSnapshotDirectoryLister
Lazy<HoodieTableMetaClient> lazyMetaClient = Lazy.lazily(tableHandle::getMetaClient);
this.indexSupportOpt = enableMetadataTable ?
- IndexSupportFactory.createIndexSupport(schemaTableName, lazyMetaClient, lazyTableMetadata, tableHandle.getRegularPredicates(), session) : Optional.empty();
+ IndexSupportFactory.createIndexSupport(tableHandle, lazyMetaClient, lazyTableMetadata, tableHandle.getRegularPredicates(), session) : Optional.empty();
}
@Override
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java
index 1f30f3d21743..9b3291b6fc4b 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java
@@ -19,7 +19,6 @@ import io.airlift.slice.Slice;
import io.airlift.slice.Slices;
import io.airlift.units.Duration;
import io.trino.parquet.predicate.TupleDomainParquetPredicate;
-import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hudi.util.TupleDomainUtils;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.SchemaTableName;
@@ -75,17 +74,17 @@ public class HudiColumnStatsIndexSupport
private final Duration columnStatsWaitTimeout;
private final long futureStartTimeMs;
- public HudiColumnStatsIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy<HoodieTableMetaClient> lazyMetaClient, Lazy<HoodieTableMetadata> lazyTableMetadata, TupleDomain<HiveColumnHandle> regularColumnPredicates)
+ public HudiColumnStatsIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy<HoodieTableMetaClient> lazyMetaClient, Lazy<HoodieTableMetadata> lazyTableMetadata, TupleDomain<String> regularColumnPredicates)
{
this(log, session, schemaTableName, lazyMetaClient, lazyTableMetadata, regularColumnPredicates);
}
- public HudiColumnStatsIndexSupport(Logger log, ConnectorSession session, SchemaTableName schemaTableName, Lazy<HoodieTableMetaClient> lazyMetaClient, Lazy<HoodieTableMetadata> lazyTableMetadata, TupleDomain<HiveColumnHandle> regularColumnPredicates)
+ public HudiColumnStatsIndexSupport(Logger log, ConnectorSession session, SchemaTableName schemaTableName, Lazy<HoodieTableMetaClient> lazyMetaClient, Lazy<HoodieTableMetadata> lazyTableMetadata, TupleDomain<String> regularColumnPredicates)
{
super(log, schemaTableName, lazyMetaClient);
this.columnStatsWaitTimeout = getColumnStatsWaitTimeout(session);
- this.regularColumnPredicates = regularColumnPredicates.transformKeys(HiveColumnHandle::getName);
- this.regularColumns = this.regularColumnPredicates.getDomains()
+ this.regularColumnPredicates = regularColumnPredicates;
+ this.regularColumns = regularColumnPredicates.getDomains()
.map(domains -> new ArrayList<>(domains.keySet()))
.orElse(new ArrayList<>());
if (regularColumnPredicates.isAll() || regularColumnPredicates.getDomains().isEmpty()) {
@@ -98,7 +97,7 @@ public class HudiColumnStatsIndexSupport
.map(col -> new ColumnIndexID(col).asBase64EncodedString()).collect(Collectors.toList());
Map<String, Type> columnTypes = regularColumnPredicates.getDomains().get().entrySet().stream()
- .collect(Collectors.toMap(entry -> entry.getKey().getName(), entry -> entry.getValue().getType()));
+ .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getType()));
domainsWithStatsFuture = CompletableFuture.supplyAsync(() -> {
HoodieTimer timer = HoodieTimer.start();
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiPartitionStatsIndexSupport.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiPartitionStatsIndexSupport.java
index 21dbdf57f262..3491d10cccfe 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiPartitionStatsIndexSupport.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiPartitionStatsIndexSupport.java
@@ -14,7 +14,6 @@
package io.trino.plugin.hudi.query.index;
import io.airlift.log.Logger;
-import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hudi.util.TupleDomainUtils;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.SchemaTableName;
@@ -44,7 +43,7 @@ public class HudiPartitionStatsIndexSupport
private static final Logger log = Logger.get(HudiColumnStatsIndexSupport.class);
private final Lazy<HoodieTableMetadata> lazyMetadataTable;
- public HudiPartitionStatsIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy<HoodieTableMetaClient> lazyMetaClient, Lazy<HoodieTableMetadata> lazyTableMetadata, TupleDomain<HiveColumnHandle> regularColumnPredicates)
+ public HudiPartitionStatsIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy<HoodieTableMetaClient> lazyMetaClient, Lazy<HoodieTableMetadata> lazyTableMetadata, TupleDomain<String> regularColumnPredicates)
{
super(log, session, schemaTableName, lazyMetaClient, lazyTableMetadata, regularColumnPredicates);
this.lazyMetadataTable = lazyTableMetadata;
@@ -67,8 +66,7 @@ public class HudiPartitionStatsIndexSupport
List<String> regularColumns = new ArrayList<>(filteredRegularPredicates.getDomains().get().keySet());
// Get columns to filter on
- List<String> encodedTargetColumnNames = regularColumns
- .stream()
+ List<String> encodedTargetColumnNames = regularColumns.stream()
.map(col -> new ColumnIndexID(col).asBase64EncodedString()).toList();
Map<String, Type> columnTypes = regularColumnPredicates.getDomains().get().entrySet().stream()
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiRecordLevelIndexSupport.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiRecordLevelIndexSupport.java
index 3c5e33f4e6a0..78f503f4b21e 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiRecordLevelIndexSupport.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiRecordLevelIndexSupport.java
@@ -16,7 +16,6 @@ package io.trino.plugin.hudi.query.index;
import io.airlift.log.Logger;
import io.airlift.slice.Slice;
import io.airlift.units.Duration;
-import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hudi.util.TupleDomainUtils;
import io.trino.spi.TrinoException;
import io.trino.spi.connector.ConnectorSession;
@@ -61,7 +60,7 @@ public class HudiRecordLevelIndexSupport
private final Duration recordIndexWaitTimeout;
private final long futureStartTimeMs;
- public HudiRecordLevelIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy<HoodieTableMetaClient> lazyMetaClient, Lazy<HoodieTableMetadata> lazyTableMetadata, TupleDomain<HiveColumnHandle> regularColumnPredicates)
+ public HudiRecordLevelIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy<HoodieTableMetaClient> lazyMetaClient, Lazy<HoodieTableMetadata> lazyTableMetadata, TupleDomain<String> regularColumnPredicates)
{
super(log, schemaTableName, lazyMetaClient);
this.recordIndexWaitTimeout = getRecordIndexWaitTimeout(session);
@@ -79,9 +78,8 @@ public class HudiRecordLevelIndexSupport
}
List<String> recordKeyFields = Arrays.asList(recordKeyFieldsOpt.get());
- TupleDomain<String> regularPredicatesTransformed = regularColumnPredicates.transformKeys(HiveColumnHandle::getName);
// Only extract the predicates relevant to the record key fields
- TupleDomain<String> filteredDomains = extractPredicatesForColumns(regularPredicatesTransformed, recordKeyFields);
+ TupleDomain<String> filteredDomains = extractPredicatesForColumns(regularColumnPredicates, recordKeyFields);
// Construct the actual record keys based on the filtered predicates using Hudi's encoding scheme
List<String> recordKeys = constructRecordKeys(filteredDomains, recordKeyFields);
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiSecondaryIndexSupport.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiSecondaryIndexSupport.java
index d078c9e6e9f6..23121902f72c 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiSecondaryIndexSupport.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiSecondaryIndexSupport.java
@@ -15,7 +15,6 @@ package io.trino.plugin.hudi.query.index;
import io.airlift.log.Logger;
import io.airlift.units.Duration;
-import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hudi.util.TupleDomainUtils;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.SchemaTableName;
@@ -51,11 +50,10 @@ public class HudiSecondaryIndexSupport
private final Duration secondaryIndexWaitTimeout;
private final long futureStartTimeMs;
- public HudiSecondaryIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy<HoodieTableMetaClient> lazyMetaClient, Lazy<HoodieTableMetadata> lazyTableMetadata, TupleDomain<HiveColumnHandle> regularColumnPredicates)
+ public HudiSecondaryIndexSupport(ConnectorSession session, SchemaTableName schemaTableName, Lazy<HoodieTableMetaClient> lazyMetaClient, Lazy<HoodieTableMetadata> lazyTableMetadata, TupleDomain<String> regularColumnPredicates)
{
super(log, schemaTableName, lazyMetaClient);
this.secondaryIndexWaitTimeout = getSecondaryIndexWaitTimeout(session);
- TupleDomain<String> regularPredicatesTransformed = regularColumnPredicates.transformKeys(HiveColumnHandle::getName);
this.relevantFileIdsFuture = CompletableFuture.supplyAsync(() -> {
HoodieTimer timer = HoodieTimer.start();
if (regularColumnPredicates.isAll() || lazyMetaClient.get().getIndexMetadata().isEmpty()) {
@@ -63,7 +61,7 @@ public class HudiSecondaryIndexSupport
return Optional.empty();
}
- Optional<Map.Entry<String, HoodieIndexDefinition>> firstApplicableIndex = findFirstApplicableSecondaryIndex(regularPredicatesTransformed);
+ Optional<Map.Entry<String, HoodieIndexDefinition>> firstApplicableIndex = findFirstApplicableSecondaryIndex(regularColumnPredicates);
if (firstApplicableIndex.isEmpty()) {
log.debug("Took %s ms but no secondary index definition found matching the query's referenced columns for table %s",
timer.endTimer(), schemaTableName);
@@ -75,7 +73,7 @@ public class HudiSecondaryIndexSupport
// `indexedColumns` should only contain one element as secondary indices only support one column
List<String> indexedColumns = applicableIndexEntry.getValue().getSourceFields();
log.debug(String.format("Using secondary index '%s' on columns %s for pruning.", indexName, indexedColumns));
- TupleDomain<String> indexPredicates = extractPredicatesForColumns(regularPredicatesTransformed, indexedColumns);
+ TupleDomain<String> indexPredicates = extractPredicatesForColumns(regularColumnPredicates, indexedColumns);
List<String> secondaryKeys = constructRecordKeys(indexPredicates, indexedColumns);
if (secondaryKeys.isEmpty()) {
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/IndexSupportFactory.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/IndexSupportFactory.java
index ab78e7662eb2..5d98177896c4 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/IndexSupportFactory.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/IndexSupportFactory.java
@@ -15,6 +15,7 @@ package io.trino.plugin.hudi.query.index;
import io.airlift.log.Logger;
import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hudi.HudiTableHandle;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.SchemaTableName;
import io.trino.spi.predicate.TupleDomain;
@@ -30,7 +31,9 @@ import static io.trino.plugin.hudi.HudiSessionProperties.isColumnStatsIndexEnabl
import static io.trino.plugin.hudi.HudiSessionProperties.isNoOpIndexEnabled;
import static io.trino.plugin.hudi.HudiSessionProperties.isPartitionStatsIndexEnabled;
import static io.trino.plugin.hudi.HudiSessionProperties.isRecordLevelIndexEnabled;
+import static io.trino.plugin.hudi.HudiSessionProperties.isResolveColumnNameCasingEnabled;
import static io.trino.plugin.hudi.HudiSessionProperties.isSecondaryIndexEnabled;
+import static io.trino.plugin.hudi.HudiUtil.getFieldFromSchema;
import static java.util.Objects.requireNonNull;
/**
@@ -49,36 +52,37 @@ public class IndexSupportFactory
* Creates the most suitable HudiIndexSupport strategy, considering configuration.
* Uses Supplier-based lazy instantiation combined with config checks.
*
- * @param schemaTableName The table schema and name
+ * @param hudiTableHandle The hudi table handle
* @param lazyMetaClient The Hudi table metadata client that is lazily instantiated.
* @param tupleDomain The query predicates.
* @param session Session containing session properties, which is required to control index behaviours for testing/debugging
* @return An Optional containing the chosen HudiIndexSupport strategy, or empty if none are applicable or enabled.
*/
public static Optional<HudiIndexSupport> createIndexSupport(
- SchemaTableName schemaTableName,
+ HudiTableHandle hudiTableHandle,
Lazy<HoodieTableMetaClient> lazyMetaClient,
Lazy<HoodieTableMetadata> lazyTableMetadata,
TupleDomain<HiveColumnHandle> tupleDomain,
ConnectorSession session)
{
+ TupleDomain<String> transformedTupleDomain = transformTupleDomain(session, hudiTableHandle, tupleDomain);
+ SchemaTableName schemaTableName = hudiTableHandle.getSchemaTableName();
// Define strategies as Suppliers paired with their config (isEnabled) flag
// IMPORTANT: Order of strategy here determines which index implementation is preferred first
List<StrategyProvider> strategyProviders = List.of(
new StrategyProvider(
() -> isRecordLevelIndexEnabled(session),
- () -> new HudiRecordLevelIndexSupport(session, schemaTableName, lazyMetaClient, lazyTableMetadata, tupleDomain)),
+ () -> new HudiRecordLevelIndexSupport(session, schemaTableName, lazyMetaClient, lazyTableMetadata, transformedTupleDomain)),
new StrategyProvider(
() -> isSecondaryIndexEnabled(session),
- () -> new HudiSecondaryIndexSupport(session, schemaTableName, lazyMetaClient, lazyTableMetadata, tupleDomain)),
+ () -> new HudiSecondaryIndexSupport(session, schemaTableName, lazyMetaClient, lazyTableMetadata, transformedTupleDomain)),
new StrategyProvider(
() -> isColumnStatsIndexEnabled(session),
- () -> new HudiColumnStatsIndexSupport(session, schemaTableName, lazyMetaClient, lazyTableMetadata, tupleDomain)),
+ () -> new HudiColumnStatsIndexSupport(session, schemaTableName, lazyMetaClient, lazyTableMetadata, transformedTupleDomain)),
new StrategyProvider(
() -> isNoOpIndexEnabled(session),
() -> new HudiNoOpIndexSupport(schemaTableName, lazyMetaClient)));
- TupleDomain<String> transformedTupleDomain = tupleDomain.transformKeys(HiveColumnHandle::getName);
for (StrategyProvider provider : strategyProviders) {
// Check if the strategy is enabled via config before instantiating
if (provider.isEnabled()) {
@@ -105,22 +109,33 @@ public class IndexSupportFactory
}
public static Optional<HudiPartitionStatsIndexSupport> createPartitionStatsIndexSupport(
- SchemaTableName schemaTableName,
+ HudiTableHandle hudiTableHandle,
Lazy<HoodieTableMetaClient> lazyMetaClient,
Lazy<HoodieTableMetadata> lazyTableMetadata,
TupleDomain<HiveColumnHandle> tupleDomain,
ConnectorSession session)
{
+ TupleDomain<String> transformedTupleDomain = transformTupleDomain(session, hudiTableHandle, tupleDomain);
+
StrategyProvider partitionStatsStrategy = new StrategyProvider(
- () -> isPartitionStatsIndexEnabled(session), () -> new HudiPartitionStatsIndexSupport(session, schemaTableName, lazyMetaClient, lazyTableMetadata, tupleDomain));
+ () -> isPartitionStatsIndexEnabled(session), () -> new HudiPartitionStatsIndexSupport(session, hudiTableHandle.getSchemaTableName(), lazyMetaClient, lazyTableMetadata, transformedTupleDomain));
- TupleDomain<String> transformedTupleDomain = tupleDomain.transformKeys(HiveColumnHandle::getName);
if (partitionStatsStrategy.isEnabled() && partitionStatsStrategy.getStrategy().canApply(transformedTupleDomain)) {
return Optional.of((HudiPartitionStatsIndexSupport) partitionStatsStrategy.getStrategy());
}
return Optional.empty();
}
+ private static TupleDomain<String> transformTupleDomain(ConnectorSession session, HudiTableHandle hudiTableHandle, TupleDomain<HiveColumnHandle> tupleDomain)
+ {
+ if (isResolveColumnNameCasingEnabled(session)) {
+ // if column case reconciliation is enabled, transform the tuple domain keys to match the column names from the Hudi table.
+ return tupleDomain.transformKeys(hiveColumnHandle ->
+ getFieldFromSchema(hiveColumnHandle.getName(), hudiTableHandle.getTableSchema()).name());
+ }
+ return tupleDomain.transformKeys(HiveColumnHandle::getName);
+ }
+
/**
* Helper class to pair the configuration check with the strategy supplier to allow for lazy initialization.
*/
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
index 7c1ae6606edb..bddc67f9cf74 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
@@ -34,7 +34,6 @@ import io.trino.plugin.hudi.query.index.IndexSupportFactory;
import io.trino.spi.TrinoException;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.ConnectorSplit;
-import io.trino.spi.connector.SchemaTableName;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.Option;
@@ -115,11 +114,10 @@ public class HudiBackgroundSplitLoader
this.enableMetadataTable = enableMetadataTable;
this.executor = requireNonNull(executor, "executor is null");
this.errorListener = requireNonNull(errorListener, "errorListener is null");
- SchemaTableName schemaTableName = tableHandle.getSchemaTableName();
this.lazyTableMetadata = lazyTableMetadata;
this.lazyMetaClient = Lazy.lazily(tableHandle::getMetaClient);
this.partitionIndexSupportOpt = enableMetadataTable ?
- IndexSupportFactory.createPartitionStatsIndexSupport(schemaTableName, lazyMetaClient, lazyTableMetadata, tableHandle.getRegularPredicates(), session) : Optional.empty();
+ IndexSupportFactory.createPartitionStatsIndexSupport(tableHandle, Lazy.lazily(tableHandle::getMetaClient), lazyTableMetadata, tableHandle.getRegularPredicates(), session) : Optional.empty();
this.isMetadataPartitionListingEnabled = isMetadataPartitionListingEnabled(session);
}
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
index 91b38f593e19..2fc020020252 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
@@ -63,6 +63,7 @@ import java.util.Map;
import static com.google.common.base.Verify.verify;
import static io.airlift.slice.Slices.utf8Slice;
import static io.trino.plugin.hudi.HudiUtil.constructSchema;
+import static io.trino.plugin.hudi.HudiUtil.getFieldFromSchema;
import static io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
import static io.trino.spi.StandardErrorCode.NUMERIC_VALUE_OUT_OF_RANGE;
import static io.trino.spi.type.BigintType.BIGINT;
@@ -149,7 +150,7 @@ public class HudiAvroSerializer
}
else {
// Record may not be projected, get index from it
- int fieldPosInSchema = record.getSchema().getField(columnHandle.getName()).pos();
+ int fieldPosInSchema = getFieldFromSchema(columnHandle.getName(), record.getSchema()).pos();
appendTo(columnTypes.get(channel), record.get(fieldPosInSchema), output);
}
}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/SessionBuilder.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/SessionBuilder.java
index 6a7d80e98e39..aa26d2d35f8e 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/SessionBuilder.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/SessionBuilder.java
@@ -23,6 +23,7 @@ import static io.trino.plugin.hudi.HudiSessionProperties.DYNAMIC_FILTERING_WAIT_
import static io.trino.plugin.hudi.HudiSessionProperties.METADATA_TABLE_ENABLED;
import static io.trino.plugin.hudi.HudiSessionProperties.PARTITION_STATS_INDEX_ENABLED;
import static io.trino.plugin.hudi.HudiSessionProperties.QUERY_PARTITION_FILTER_REQUIRED;
+import static io.trino.plugin.hudi.HudiSessionProperties.RECORD_INDEX_WAIT_TIMEOUT;
import static io.trino.plugin.hudi.HudiSessionProperties.RECORD_LEVEL_INDEX_ENABLED;
import static io.trino.plugin.hudi.HudiSessionProperties.SECONDARY_INDEX_ENABLED;
import static io.trino.plugin.hudi.HudiSessionProperties.SECONDARY_INDEX_WAIT_TIMEOUT;
@@ -134,4 +135,9 @@ public class SessionBuilder
{
return setCatalogProperty(SECONDARY_INDEX_WAIT_TIMEOUT, durationProp);
}
+
+ public SessionBuilder withRecordIndexTimeout(String durationProp)
+ {
+ return setCatalogProperty(RECORD_INDEX_WAIT_TIMEOUT, durationProp);
+ }
}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
index 1308cff269a9..ea35183ec4ed 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
@@ -38,9 +38,11 @@ import static io.trino.filesystem.tracing.CacheFileSystemTraceUtils.getCacheOper
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_FG_PT_V8_MOR;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.DATA;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.INDEX_DEFINITION;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.LOG;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE_PROPERTIES;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TABLE_PROPERTIES;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TIMELINE;
import static io.trino.testing.MultisetAssertions.assertMultisetsEqual;
import static java.util.stream.Collectors.toCollection;
@@ -83,8 +85,12 @@ public class TestHudiAlluxioCacheFileOperations
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("Alluxio.readCached", DATA), 2)
.addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 27)
+ .addCopies(new FileOperation("Alluxio.readCached", TIMELINE), 4)
+ .addCopies(new FileOperation("Alluxio.readCached", LOG), 15)
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
.addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 10)
+ .addCopies(new FileOperation("InputFile.length", TIMELINE), 2)
+ .addCopies(new FileOperation("InputFile.length", LOG), 1)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
.add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
@@ -95,8 +101,12 @@ public class TestHudiAlluxioCacheFileOperations
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("Alluxio.readCached", DATA), 2)
.addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 27)
+ .addCopies(new FileOperation("Alluxio.readCached", TIMELINE), 4)
+ .addCopies(new FileOperation("Alluxio.readCached", LOG), 15)
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
.addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 10)
+ .addCopies(new FileOperation("InputFile.length", TIMELINE), 2)
+ .addCopies(new FileOperation("InputFile.length", LOG), 1)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
.add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
@@ -116,8 +126,12 @@ public class TestHudiAlluxioCacheFileOperations
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("Alluxio.readCached", DATA), 6)
.addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 288)
+ .addCopies(new FileOperation("Alluxio.readCached", TIMELINE), 8)
+ .addCopies(new FileOperation("Alluxio.readCached", LOG), 30)
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 39)
.addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 93)
+ .addCopies(new FileOperation("InputFile.length", TIMELINE), 4)
+ .addCopies(new FileOperation("InputFile.length", LOG), 2)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 5)
.addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 3)
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 5)
@@ -127,8 +141,12 @@ public class TestHudiAlluxioCacheFileOperations
ImmutableMultiset.<FileOperation>builder()
.addCopies(new FileOperation("Alluxio.readCached", DATA), 6)
.addCopies(new FileOperation("Alluxio.readCached", METADATA_TABLE), 215)
+ .addCopies(new FileOperation("Alluxio.readCached", TIMELINE), 8)
+ .addCopies(new FileOperation("Alluxio.readCached", LOG), 30)
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 29)
.addCopies(new FileOperation("InputFile.length", METADATA_TABLE), 69)
+ .addCopies(new FileOperation("InputFile.length", TIMELINE), 4)
+ .addCopies(new FileOperation("InputFile.length", LOG), 2)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
.addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
.addCopies(new FileOperation("InputFile.newStream", TABLE_PROPERTIES), 4)
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
index 7e605a310a98..545693f1ea07 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
@@ -58,7 +58,8 @@ public class TestHudiConfig
.setRecordIndexWaitTimeout(Duration.valueOf("2s"))
.setSecondaryIndexWaitTimeout(Duration.valueOf("2s"))
.setMetadataPartitionListingEnabled(true)
- .setMetadataCacheEnabled(true));
+ .setMetadataCacheEnabled(true)
+ .setResolveColumnNameCasingEnabled(true));
}
@Test
@@ -92,6 +93,7 @@ public class TestHudiConfig
.put("hudi.index.secondary-index.wait-timeout", "1s")
.put("hudi.metadata.cache.enabled", "false")
.put("hudi.metadata.partition-listing.enabled", "false")
+ .put("hudi.table.resolve-column-name-casing.enabled", "false")
.buildOrThrow();
HudiConfig expected = new HudiConfig()
@@ -121,7 +123,8 @@ public class TestHudiConfig
.setRecordIndexWaitTimeout(Duration.valueOf("1s"))
.setSecondaryIndexWaitTimeout(Duration.valueOf("1s"))
.setMetadataPartitionListingEnabled(false)
- .setMetadataCacheEnabled(false);
+ .setMetadataCacheEnabled(false)
+ .setResolveColumnNameCasingEnabled(false);
assertFullMapping(properties, expected);
}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
index 3f6d97f16919..3f2cb7d77173 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
@@ -35,9 +35,11 @@ import static io.trino.filesystem.tracing.CacheFileSystemTraceUtils.isTrinoSchem
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_FG_PT_V8_MOR;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.DATA;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.INDEX_DEFINITION;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.LOG;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE_PROPERTIES;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TABLE_PROPERTIES;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TIMELINE;
import static io.trino.testing.MultisetAssertions.assertMultisetsEqual;
import static java.util.stream.Collectors.toCollection;
@@ -74,6 +76,8 @@ public class TestHudiMemoryCacheFileOperations
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 2)
.addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 4)
.addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 6)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", TIMELINE), 2)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", LOG), 1)
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
.add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
@@ -86,6 +90,8 @@ public class TestHudiMemoryCacheFileOperations
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 2)
.addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 4)
.addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 6)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", TIMELINE), 2)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", LOG), 1)
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 4)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
.add(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
@@ -107,6 +113,8 @@ public class TestHudiMemoryCacheFileOperations
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 6)
.addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 39)
.addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 54)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", TIMELINE), 4)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", LOG), 2)
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 39)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 5)
.addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 3)
@@ -118,6 +126,8 @@ public class TestHudiMemoryCacheFileOperations
.addCopies(new FileOperation("FileSystemCache.cacheInput", DATA), 6)
.addCopies(new FileOperation("FileSystemCache.cacheLength", METADATA_TABLE), 29)
.addCopies(new FileOperation("FileSystemCache.cacheStream", METADATA_TABLE), 40)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", TIMELINE), 4)
+ .addCopies(new FileOperation("FileSystemCache.cacheStream", LOG), 2)
.addCopies(new FileOperation("InputFile.lastModified", METADATA_TABLE), 29)
.addCopies(new FileOperation("InputFile.newStream", INDEX_DEFINITION), 4)
.addCopies(new FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
index 416914d3f7e2..58541011c238 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
@@ -35,9 +35,11 @@ import static io.trino.filesystem.tracing.CacheFileSystemTraceUtils.isTrinoSchem
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_FG_PT_V8_MOR;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.DATA;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.INDEX_DEFINITION;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.LOG;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.METADATA_TABLE_PROPERTIES;
import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TABLE_PROPERTIES;
+import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TIMELINE;
import static io.trino.testing.MultisetAssertions.assertMultisetsEqual;
import static java.util.stream.Collectors.toCollection;
@@ -76,6 +78,8 @@ public class TestHudiNoCacheFileOperations
.addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 4)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 6)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TIMELINE), 2)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", LOG), 1)
.add(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
.build());
@@ -88,6 +92,8 @@ public class TestHudiNoCacheFileOperations
.addCopies(new FileOperationUtils.FileOperation("InputFile.length", METADATA_TABLE), 4)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", INDEX_DEFINITION), 2)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 6)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TIMELINE), 2)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", LOG), 1)
.add(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 2)
.build());
@@ -111,6 +117,8 @@ public class TestHudiNoCacheFileOperations
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 54)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 3)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 5)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TIMELINE), 4)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", LOG), 2)
.build());
assertFileSystemAccesses(query,
@@ -122,6 +130,8 @@ public class TestHudiNoCacheFileOperations
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE), 40)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", METADATA_TABLE_PROPERTIES), 2)
.addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TABLE_PROPERTIES), 4)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", TIMELINE), 4)
+ .addCopies(new FileOperationUtils.FileOperation("InputFile.newStream", LOG), 2)
.build());
}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
index a8f90f826d05..3c98a85f5b2a 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
@@ -72,7 +72,10 @@ import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COMPREHENSIVE_TYPES_V6_MOR;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COMPREHENSIVE_TYPES_V8_MOR;
+import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COW_PT_TABLE_WITH_FIELD_NAMES_IN_CAPS;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COW_PT_TBL;
+import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COW_TABLE_WITH_FIELD_NAMES_IN_CAPS;
+import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COW_TABLE_WITH_MULTI_KEYS_AND_FIELD_NAMES_IN_CAPS;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_CUSTOM_KEYGEN_PT_V8_MOR;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_PT_V8_MOR;
import static io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_NON_EXTRACTABLE_PARTITION_PATH;
@@ -681,6 +684,81 @@ public class TestHudiSmokeTest
assertThat(prunedSplits).isEqualTo(1);
}
+ @ParameterizedTest
+ @EnumSource(
+ value = ResourceHudiTablesInitializer.TestingTable.class,
+ names = {"HUDI_COW_TABLE_WITH_FIELD_NAMES_IN_CAPS", "HUDI_MOR_TABLE_WITH_FIELD_NAMES_IN_CAPS"})
+ public void testFileSkippingWithColumnNameUsingUppercaseLetters(ResourceHudiTablesInitializer.TestingTable table)
+ {
+ Session session = SessionBuilder.from(getSession())
+ .withMdtEnabled(true)
+ .withColStatsIndexEnabled(true)
+ .withColumnStatsTimeout("5s")
+ .withRecordLevelIndexEnabled(false)
+ .withSecondaryIndexEnabled(false)
+ .withPartitionStatsIndexEnabled(false)
+ .build();
+ MaterializedResult prunedRes = getQueryRunner().execute(session, "SELECT * FROM " + table + " WHERE name='Alice'");
+ MaterializedResult totalRes = getQueryRunner().execute(session, "SELECT * FROM " + table);
+ int totalSplits = totalRes.getStatementStats().get().getTotalSplits();
+ int totalRows = totalRes.getRowCount();
+ int prunedSplits = prunedRes.getStatementStats().get().getTotalSplits();
+ int prunedRows = prunedRes.getRowCount();
+ assertThat(prunedSplits).isLessThan(totalSplits);
+ // With colstats file skipping, only 1 split should be returned
+ assertThat(prunedSplits).isEqualTo(1);
+ assertThat(totalRows).isEqualTo(2);
+ assertThat(prunedRows).isEqualTo(1);
+ }
+
+ @Test
+ public void testRLIWithColumnNameUsingUppercaseLetters()
+ {
+ Session session = SessionBuilder.from(getSession())
+ .withMdtEnabled(true)
+ .withColStatsIndexEnabled(false)
+ .withRecordLevelIndexEnabled(true)
+ .withRecordIndexTimeout("1s")
+ .withSecondaryIndexEnabled(false)
+ .withPartitionStatsIndexEnabled(false)
+ .build();
+ MaterializedResult totalRes = getQueryRunner().execute(session, "SELECT * FROM " + HUDI_COW_TABLE_WITH_FIELD_NAMES_IN_CAPS);
+ MaterializedResult prunedRes = getQueryRunner().execute(session, "SELECT * FROM " + HUDI_COW_TABLE_WITH_FIELD_NAMES_IN_CAPS + " WHERE id='1'");
+ int totalSplits = totalRes.getStatementStats().get().getTotalSplits();
+ int totalRows = totalRes.getRowCount();
+ int prunedSplits = prunedRes.getStatementStats().get().getTotalSplits();
+ int prunedRows = prunedRes.getRowCount();
+ assertThat(prunedSplits).isLessThan(totalSplits);
+ // With record index file skipping, only 1 split should be returned
+ assertThat(prunedSplits).isEqualTo(1);
+ assertThat(totalRows).isEqualTo(2);
+ assertThat(prunedRows).isEqualTo(1);
+ }
+
+ @Test
+ public void testMultiKeyRLIWithColumnNameUsingUppercaseLetters()
+ {
+ Session session = SessionBuilder.from(getSession())
+ .withMdtEnabled(true)
+ .withColStatsIndexEnabled(false)
+ .withRecordLevelIndexEnabled(true)
+ .withRecordIndexTimeout("1s")
+ .withSecondaryIndexEnabled(false)
+ .withPartitionStatsIndexEnabled(false)
+ .build();
+ MaterializedResult totalRes = getQueryRunner().execute(session, "SELECT * FROM " + HUDI_COW_TABLE_WITH_MULTI_KEYS_AND_FIELD_NAMES_IN_CAPS);
+ MaterializedResult prunedRes = getQueryRunner().execute(session, "SELECT * FROM " + HUDI_COW_TABLE_WITH_MULTI_KEYS_AND_FIELD_NAMES_IN_CAPS + " WHERE id='1' and age=30");
+ int totalSplits = totalRes.getStatementStats().get().getTotalSplits();
+ int totalRows = totalRes.getRowCount();
+ int prunedSplits = prunedRes.getStatementStats().get().getTotalSplits();
+ int prunedRows = prunedRes.getRowCount();
+ assertThat(prunedSplits).isLessThan(totalSplits);
+ // With record index file skipping, only 1 split should be returned
+ assertThat(prunedSplits).isEqualTo(1);
+ assertThat(totalRows).isEqualTo(2);
+ assertThat(prunedRows).isEqualTo(1);
+ }
+
@ParameterizedTest
@EnumSource(
value = ResourceHudiTablesInitializer.TestingTable.class,
@@ -763,6 +841,23 @@ public class TestHudiSmokeTest
assertThat(prunedSplits).isEqualTo(expectedSplits);
}
+ @Test
+ public void testPartitionStatsIndexPartitionPruningWithColumnNameUsingUppercaseLetters()
+ {
+ Session session = SessionBuilder.from(getSession())
+ .withMdtEnabled(true)
+ .withColStatsIndexEnabled(false)
+ .withRecordLevelIndexEnabled(false)
+ .withSecondaryIndexEnabled(false)
+ .withPartitionStatsIndexEnabled(true)
+ .build();
+ MaterializedResult prunedRes = getQueryRunner().execute(session, "SELECT * FROM " + HUDI_COW_PT_TABLE_WITH_FIELD_NAMES_IN_CAPS
+ + " WHERE country = 'US'");
+ int prunedSplits = prunedRes.getStatementStats().get().getTotalSplits();
+
+ assertThat(prunedSplits).isEqualTo(2);
+ }
+
@ParameterizedTest
@EnumSource(
value = ResourceHudiTablesInitializer.TestingTable.class,
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java
index 19e7f7fdf1ec..531ea5ed5c61 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java
@@ -176,13 +176,14 @@ public class TestHudiSplitFactory
private static HudiTableHandle createTableHandle()
{
return new HudiTableHandle(
- "test_schema",
- "test_table",
- "/test/path",
- HoodieTableType.MERGE_ON_READ,
- ImmutableList.of(),
- TupleDomain.all(),
- TupleDomain.all());
+ "test_schema",
+ "test_table",
+ "/test/path",
+ HoodieTableType.MERGE_ON_READ,
+ ImmutableList.of(),
+ TupleDomain.all(),
+ TupleDomain.all(),
+ "");
}
private static FileSlice createFileSlice(DataSize baseFileSize, Option<DataSize> logFileSize)
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
index da8e7e19c15c..420e10e1ada6 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
@@ -334,6 +334,10 @@ public class ResourceHudiTablesInitializer
{
HUDI_NON_PART_COW(nonPartitionRegularColumns()),
HUDI_TRIPS_COW_V8(tripsRegularColumns()),
+ HUDI_COW_TABLE_WITH_FIELD_NAMES_IN_CAPS(hudiTableWithFieldNamesInCapsRegularColumns()),
+ HUDI_COW_PT_TABLE_WITH_FIELD_NAMES_IN_CAPS(hudiTableWithFieldNamesInCapsRegularColumns(), hudiTableWithFieldNamesInCapsPartitionColumns(), hudiTableWithFieldNamesInCapsPartitions(), false), // delete
+ HUDI_COW_TABLE_WITH_MULTI_KEYS_AND_FIELD_NAMES_IN_CAPS(hudiTableWithFieldNamesInCapsRegularColumns()),
+ HUDI_MOR_TABLE_WITH_FIELD_NAMES_IN_CAPS(hudiTableWithFieldNamesInCapsRegularColumns(), ImmutableList.of(), ImmutableMap.of(), true),
HUDI_COW_PT_TBL(multiPartitionRegularColumns(), multiPartitionColumns(), multiPartitions(), false),
STOCK_TICKS_COW(stockTicksRegularColumns(), stockTicksPartitionColumns(), stockTicksPartitions(), false),
STOCK_TICKS_MOR(stockTicksRegularColumns(), stockTicksPartitionColumns(), stockTicksPartitions(), false),
@@ -438,6 +442,32 @@ public class ResourceHudiTablesInitializer
column("hh", HIVE_STRING));
}
+ // Table schema has capitalized column names, but the catalog returns them in lowercase.
+ // Using lowercase here to match the catalog for testing.
+ private static List<Column> hudiTableWithFieldNamesInCapsRegularColumns()
+ {
+ return ImmutableList.of(
+ column("id", HIVE_STRING),
+ column("name", HIVE_STRING),
+ column("age", HIVE_INT));
+ }
+
+ // The actual Hudi table has "Country" as the partition field name, but the catalog provides it in lowercase.
+ // Using lowercase here to stay consistent with the catalog for testing.
+ private static Map<String, String> hudiTableWithFieldNamesInCapsPartitions()
+ {
+ return ImmutableMap.of(
+ "country=IND", "IND",
+ "country=US", "US");
+ }
+
+ // The actual Hudi table has "Country" as the partition field name, but the catalog provides it in lowercase.
+ // Using lowercase here to stay consistent with the catalog for testing.
+ private static List<Column> hudiTableWithFieldNamesInCapsPartitionColumns()
+ {
+ return ImmutableList.of(column("country", HIVE_STRING));
+ }
+
private static List<Column> tripsRegularColumns()
{
return ImmutableList.of(
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_pt_table_with_field_names_in_caps.md b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_pt_table_with_field_names_in_caps.md
new file mode 100644
index 000000000000..39e06481bcd7
--- /dev/null
+++ b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_pt_table_with_field_names_in_caps.md
@@ -0,0 +1,57 @@
+## Create script
+
+Structure of table:
+- COW partitioned table with field names containing caps
+- No log files
+
+```scala
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.Row
+
+val schema = StructType(Seq(
+ StructField("Id", StringType, nullable = false),
+ StructField("Name", StringType, nullable = true),
+ StructField("Age", IntegerType, nullable = true),
+ StructField("Country", StringType, nullable = true)
+))
+
+
+var data = Seq(
+ Row("1", "Alice", 30, "IND"),
+ Row("2", "Bob", 25, "US")
+)
+
+var df = spark.createDataFrame(
+ spark.sparkContext.parallelize(data),
+ schema
+)
+
+df.show()
+
+var basePath = "file:///tmp/hudi_cow_pt_table_with_field_names_in_caps/"
+
+df.write.format("hudi").mode("Append").option("hoodie.table.name", "hudi_cow_pt_table_with_field_names_in_caps").option("hoodie.datasource.write.table.type", "COPY_ON_WRITE").option("hoodie.datasource.write.recordkey.field","Id").option("hoodie.datasource.write.operation","bulk_insert").option("hoodie.metadata.index.column.stats.enable", "true").option("hoodie.metadata.record.index.enable", "true").option("hoodie.datasource.write.secondarykey.column", "Name").option("hoodie.datasource.wr [...]
+
+data = Seq(
+ Row("3", "Charlie", 30, "IND"),
+ Row("4", "David", 25, "US")
+)
+
+df = spark.createDataFrame(
+ spark.sparkContext.parallelize(data),
+ schema
+)
+
+df.show()
+
+df.write.format("hudi").mode("Append")
+ .option("hoodie.table.name", "hudi_cow_pt_table_with_field_names_in_caps")
+ .option("hoodie.datasource.write.table.type", "COPY_ON_WRITE")
+ .option("hoodie.datasource.write.recordkey.field","Id")
+ .option("hoodie.datasource.write.operation","bulk_insert")
+ .option("hoodie.metadata.index.column.stats.enable", "true")
+ .option("hoodie.metadata.record.index.enable", "true")
+ .option("hoodie.datasource.write.secondarykey.column", "Name")
+ .option("hoodie.datasource.write.partitionpath.field", "Country")
+ .save(basePath)
+```
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_pt_table_with_field_names_in_caps.zip b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_pt_table_with_field_names_in_caps.zip
new file mode 100644
index 000000000000..5ab36564e196
Binary files /dev/null and b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_pt_table_with_field_names_in_caps.zip differ
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_field_names_in_caps.md b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_field_names_in_caps.md
new file mode 100644
index 000000000000..4f1c017fedd6
--- /dev/null
+++ b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_field_names_in_caps.md
@@ -0,0 +1,41 @@
+## Create script
+
+Structure of table:
+- COW table with field names containing caps
+- No log files
+
+```scala
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.Row
+
+val schema = StructType(Seq(
+ StructField("Id", StringType, nullable = false),
+ StructField("Name", StringType, nullable = true),
+ StructField("Age", IntegerType, nullable = true)
+))
+
+
+val data = Seq(
+ Row("1", "Alice", 30),
+ Row("2", "Bob", 25)
+)
+
+val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(data),
+ schema
+)
+
+df.show()
+
+var basePath = "file:///tmp/hudi_cow_table_with_field_names_in_caps/"
+
+df.write.format("hudi").mode("Append")
+ .option("hoodie.table.name", "hudi_cow_table_with_field_names_in_caps")
+ .option("hoodie.datasource.write.table.type", "COPY_ON_WRITE")
+ .option("hoodie.datasource.write.recordkey.field","Id")
+ .option("hoodie.datasource.write.operation","bulk_insert")
+ .option("hoodie.metadata.index.column.stats.enable", "true")
+ .option("hoodie.metadata.record.index.enable", "true")
+ .option("hoodie.datasource.write.secondarykey.column", "Name")
+ .save(basePath)
+```
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_field_names_in_caps.zip b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_field_names_in_caps.zip
new file mode 100644
index 000000000000..ede87b71413a
Binary files /dev/null and b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_field_names_in_caps.zip differ
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_multi_keys_and_field_names_in_caps.md b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_multi_keys_and_field_names_in_caps.md
new file mode 100644
index 000000000000..9a7ccd9eaf57
--- /dev/null
+++ b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_multi_keys_and_field_names_in_caps.md
@@ -0,0 +1,41 @@
+## Create script
+
+Structure of table:
+- COW table with multiple record keys and field names containing caps
+- No log files
+
+```scala
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.Row
+
+val schema = StructType(Seq(
+ StructField("Id", StringType, nullable = false),
+ StructField("Name", StringType, nullable = true),
+ StructField("Age", IntegerType, nullable = true)
+))
+
+
+val data = Seq(
+ Row("1", "Alice", 30),
+ Row("2", "Bob", 25)
+)
+
+val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(data),
+ schema
+)
+
+df.show()
+
+var basePath = "file:///tmp/hudi_cow_table_with_multi_keys_and_field_names_in_caps/"
+
+df.write.format("hudi").mode("Append")
+ .option("hoodie.table.name", "hudi_cow_table_with_multi_keys_and_field_names_in_caps")
+ .option("hoodie.datasource.write.table.type", "COPY_ON_WRITE")
+ .option("hoodie.datasource.write.recordkey.field","Id,Age")
+ .option("hoodie.datasource.write.operation","bulk_insert")
+ .option("hoodie.metadata.index.column.stats.enable", "true")
+ .option("hoodie.metadata.record.index.enable", "true")
+ .option("hoodie.datasource.write.secondarykey.column", "Name")
+ .save(basePath)
+```
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_multi_keys_and_field_names_in_caps.zip b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_multi_keys_and_field_names_in_caps.zip
new file mode 100644
index 000000000000..e2b9df2100dc
Binary files /dev/null and b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_cow_table_with_multi_keys_and_field_names_in_caps.zip differ
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_mor_table_with_field_names_in_caps.md b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_mor_table_with_field_names_in_caps.md
new file mode 100644
index 000000000000..f73eb1591343
--- /dev/null
+++ b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_mor_table_with_field_names_in_caps.md
@@ -0,0 +1,41 @@
+## Create script
+
+Structure of table:
+- MOR table with field names containing caps
+- No log files
+
+```scala
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.Row
+
+val schema = StructType(Seq(
+ StructField("Id", StringType, nullable = false),
+ StructField("Name", StringType, nullable = true),
+ StructField("Age", IntegerType, nullable = true)
+))
+
+
+val data = Seq(
+ Row("1", "Alice", 30),
+ Row("2", "Bob", 25)
+)
+
+val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(data),
+ schema
+)
+
+df.show()
+
+var basePath = "file:///tmp/hudi_mor_table_with_field_names_in_caps/"
+
+df.write.format("hudi").mode("Append")
+ .option("hoodie.table.name", "hudi_mor_table_with_field_names_in_caps")
+ .option("hoodie.datasource.write.table.type", "MERGE_ON_READ")
+ .option("hoodie.datasource.write.recordkey.field","Id")
+ .option("hoodie.datasource.write.operation","bulk_insert")
+ .option("hoodie.metadata.index.column.stats.enable", "true")
+ .option("hoodie.metadata.record.index.enable", "true")
+ .option("hoodie.datasource.write.secondarykey.column", "Name")
+ .save(basePath)
+```
diff --git a/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_mor_table_with_field_names_in_caps.zip b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_mor_table_with_field_names_in_caps.zip
new file mode 100644
index 000000000000..bb4f513aff9d
Binary files /dev/null and b/hudi-trino-plugin/src/test/resources/hudi-testing-data/hudi_mor_table_with_field_names_in_caps.zip differ
(hudi) 12/13: [ENG-31282][Trino] Workers should use latest commit time from table handle (#61)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit b1494e866489376fac1672679d86a11bfe828dfb
Author: vamsikarnika <va...@gmail.com>
AuthorDate: Mon Aug 25 18:56:40 2025 +0530
[ENG-31282][Trino] Workers should use latest commit time from table handle (#61)
(cherry picked from commit 47844ce82a7a685c6fecc56f45b2987f3c03f1b8)
---
.../java/io/trino/plugin/hudi/HudiMetadata.java | 4 +-
.../trino/plugin/hudi/HudiPageSourceProvider.java | 3 +-
.../java/io/trino/plugin/hudi/HudiTableHandle.java | 63 ++++++++++++++++------
.../plugin/hudi/split/TestHudiSplitFactory.java | 3 +-
4 files changed, 53 insertions(+), 20 deletions(-)
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiMetadata.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiMetadata.java
index 02df12bab2dd..917c94ea43d9 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiMetadata.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiMetadata.java
@@ -166,8 +166,8 @@ public class HudiMetadata
Optional.of(Lazy.lazily(() -> getLatestTableSchema(lazyMetaClient.get(), tableName.getTableName()))) : Optional.empty();
return new HudiTableHandle(
- Optional.of(table),
- Optional.of(lazyMetaClient),
+ table,
+ lazyMetaClient,
tableName.getSchemaName(),
tableName.getTableName(),
table.getStorage().getLocation(),
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
index 7d3812e2d71b..9b8411fdf907 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
@@ -219,7 +219,6 @@ public class HudiPageSourceProvider
// TODO: Move this into HudiTableHandle
HoodieTableMetaClient metaClient = buildTableMetaClient(
fileSystemFactory.create(session), hudiTableHandle.getSchemaTableName().toString(), hudiTableHandle.getBasePath());
- String latestCommitTime = metaClient.getCommitsTimeline().lastInstant().get().requestedTime();
HudiTrinoReaderContext readerContext = new HudiTrinoReaderContext(
dataPageSource,
@@ -237,7 +236,7 @@ public class HudiPageSourceProvider
readerContext,
new HudiTrinoStorage(fileSystemFactory.create(session), new TrinoStorageConfiguration()),
hudiTableHandle.getBasePath(),
- latestCommitTime,
+ hudiTableHandle.getLatestCommitTime(),
convertToFileSlice(hudiSplit, hudiTableHandle.getBasePath()),
dataSchema,
requestedSchema,
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiTableHandle.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiTableHandle.java
index 6ff4198c5646..75ae962286a4 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiTableHandle.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiTableHandle.java
@@ -34,6 +34,7 @@ import org.apache.hudi.util.Lazy;
import java.util.List;
import java.util.Optional;
import java.util.Set;
+import java.util.function.Supplier;
import static com.google.common.base.Preconditions.checkArgument;
import static io.trino.spi.connector.SchemaTableName.schemaTableName;
@@ -56,7 +57,7 @@ public class HudiTableHandle
// Coordinator-only
private final transient Optional<Table> table;
private final transient Optional<Lazy<HoodieTableMetaClient>> lazyMetaClient;
- private final transient Optional<Lazy<String>> lazyLatestCommitTime;
+ private final transient Lazy<String> lazyLatestCommitTime;
@JsonCreator
public HudiTableHandle(
@@ -67,15 +68,16 @@ public class HudiTableHandle
@JsonProperty("partitionColumns") List<HiveColumnHandle> partitionColumns,
@JsonProperty("partitionPredicates") TupleDomain<HiveColumnHandle> partitionPredicates,
@JsonProperty("regularPredicates") TupleDomain<HiveColumnHandle> regularPredicates,
- @JsonProperty("tableSchemaStr") String tableSchemaStr)
+ @JsonProperty("tableSchemaStr") String tableSchemaStr,
+ @JsonProperty("latestCommitTime") String latestCommitTime)
{
this(Optional.empty(), Optional.empty(), schemaName, tableName, basePath, tableType, partitionColumns, ImmutableSet.of(),
- partitionPredicates, regularPredicates, buildTableSchema(tableSchemaStr));
+ partitionPredicates, regularPredicates, buildTableSchema(tableSchemaStr), () -> latestCommitTime);
}
public HudiTableHandle(
- Optional<Table> table,
- Optional<Lazy<HoodieTableMetaClient>> lazyMetaClient,
+ Table table,
+ Lazy<HoodieTableMetaClient> lazyMetaClient,
String schemaName,
String tableName,
String basePath,
@@ -86,15 +88,46 @@ public class HudiTableHandle
TupleDomain<HiveColumnHandle> regularPredicates,
Optional<Lazy<Schema>> hudiTableSchema)
{
- this.table = requireNonNull(table, "table is null");
- this.lazyMetaClient = requireNonNull(lazyMetaClient, "lazyMetaClient is null");
- this.lazyLatestCommitTime = Optional.of(Lazy.lazily(() ->
- getMetaClient().getActiveTimeline()
+ this(
+ Optional.of(table),
+ Optional.of(lazyMetaClient),
+ schemaName,
+ tableName,
+ basePath,
+ tableType,
+ partitionColumns,
+ constraintColumns,
+ partitionPredicates,
+ regularPredicates,
+ hudiTableSchema,
+ () -> lazyMetaClient
+ .get()
+ .getActiveTimeline()
.getCommitsTimeline()
.filterCompletedInstants()
.lastInstant()
.map(HoodieInstant::requestedTime)
- .orElseThrow(() -> new TrinoException(HudiErrorCode.HUDI_NO_VALID_COMMIT, "Table has no valid commits"))));
+ .orElseThrow(() -> new TrinoException(
+ HudiErrorCode.HUDI_NO_VALID_COMMIT,
+ "Table has no valid commits")));
+ }
+
+ HudiTableHandle(
+ Optional<Table> table,
+ Optional<Lazy<HoodieTableMetaClient>> lazyMetaClient,
+ String schemaName,
+ String tableName,
+ String basePath,
+ HoodieTableType tableType,
+ List<HiveColumnHandle> partitionColumns,
+ Set<HiveColumnHandle> constraintColumns,
+ TupleDomain<HiveColumnHandle> partitionPredicates,
+ TupleDomain<HiveColumnHandle> regularPredicates,
+ Optional<Lazy<Schema>> hudiTableSchema,
+ Supplier<String> latestCommitTimeSupplier)
+ {
+ this.table = requireNonNull(table, "table is null");
+ this.lazyMetaClient = requireNonNull(lazyMetaClient, "lazyMetaClient is null");
this.schemaName = requireNonNull(schemaName, "schemaName is null");
this.tableName = requireNonNull(tableName, "tableName is null");
this.basePath = requireNonNull(basePath, "basePath is null");
@@ -104,6 +137,7 @@ public class HudiTableHandle
this.partitionPredicates = requireNonNull(partitionPredicates, "partitionPredicates is null");
this.regularPredicates = requireNonNull(regularPredicates, "regularPredicates is null");
this.hudiTableSchema = requireNonNull(hudiTableSchema, "hudiTableSchema is null");
+ this.lazyLatestCommitTime = Lazy.lazily(latestCommitTimeSupplier);
}
/**
@@ -144,12 +178,10 @@ public class HudiTableHandle
return lazyMetaClient.get().get();
}
+ @JsonProperty
public String getLatestCommitTime()
{
- checkArgument(lazyLatestCommitTime.isPresent(),
- "getLatestCommitTime() called on a table handle that has no Hudi meta-client; "
- + "this is likely because it is called on the worker.");
- return lazyLatestCommitTime.get().get();
+ return lazyLatestCommitTime.get();
}
@JsonProperty
@@ -237,7 +269,8 @@ public class HudiTableHandle
constraintColumns,
partitionPredicates.intersect(partitionTupleDomain),
regularPredicates.intersect(regularTupleDomain),
- hudiTableSchema);
+ hudiTableSchema,
+ this::getLatestCommitTime);
}
@Override
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java
index 531ea5ed5c61..9c4f4a6176fc 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/split/TestHudiSplitFactory.java
@@ -183,7 +183,8 @@ public class TestHudiSplitFactory
ImmutableList.of(),
TupleDomain.all(),
TupleDomain.all(),
- "");
+ "",
+ "101");
}
private static FileSlice createFileSlice(DataSize baseFileSize, Option<DataSize> logFileSize)
(hudi) 09/13: [ENG-31327] [Trino] Fix flaky tests due to table stats computation lagging behind query execution (#62)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 5ef46667a6d39f3e7f5f352bfab658790baf05f1
Author: vamsikarnika <va...@gmail.com>
AuthorDate: Wed Aug 20 00:01:19 2025 +0530
[ENG-31327] [Trino] Fix flaky tests due to table stats computation lagging behind query execution (#62)
* Fix flaky tests
* Added comment
---------
Co-authored-by: Vamsi <va...@onehouse.ai>
(cherry picked from commit ae88db81bc30bbcb5d33d5378468f3afdfa90561)
---
.../io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java | 5 +++++
.../io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java | 5 +++++
.../java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java | 7 +++++++
3 files changed, 17 insertions(+)
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
index 82e03799d1ed..81254ca23f72 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
@@ -75,6 +75,7 @@ public class TestHudiAlluxioCacheFileOperations
@Test
public void testSelectWithFilter()
+ throws InterruptedException
{
@Language("SQL") String query = "SELECT * FROM " + HUDI_MULTI_FG_PT_V8_MOR + " WHERE country='SG'";
assertFileSystemAccesses(
@@ -104,6 +105,7 @@ public class TestHudiAlluxioCacheFileOperations
@Test
public void testJoin()
+ throws InterruptedException
{
@Language("SQL") String query = "SELECT t1.id, t1.name, t1.price, t1.ts FROM " +
HUDI_MULTI_FG_PT_V8_MOR + " t1 " +
@@ -134,9 +136,12 @@ public class TestHudiAlluxioCacheFileOperations
}
private void assertFileSystemAccesses(@Language("SQL") String query, Multiset<FileOperation> expectedCacheAccesses)
+ throws InterruptedException
{
DistributedQueryRunner queryRunner = getDistributedQueryRunner();
queryRunner.executeWithPlan(queryRunner.getDefaultSession(), query);
+ // Allow time for table stats computation to finish before validation.
+ Thread.sleep(1000L);
assertMultisetsEqual(getFileOperations(queryRunner), expectedCacheAccesses);
}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
index 5799ce17893c..2df228e7bb9e 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
@@ -65,6 +65,7 @@ public class TestHudiMemoryCacheFileOperations
@Test
public void testSelectWithFilter()
+ throws InterruptedException
{
@Language("SQL") String query = "SELECT * FROM " + HUDI_MULTI_FG_PT_V8_MOR + " WHERE country='SG'";
assertFileSystemAccesses(
@@ -94,6 +95,7 @@ public class TestHudiMemoryCacheFileOperations
@Test
public void testJoin()
+ throws InterruptedException
{
@Language("SQL") String query = "SELECT t1.id, t1.name, t1.price, t1.ts FROM " +
HUDI_MULTI_FG_PT_V8_MOR + " t1 " +
@@ -124,9 +126,12 @@ public class TestHudiMemoryCacheFileOperations
}
private void assertFileSystemAccesses(@Language("SQL") String query, Multiset<FileOperation> expectedCacheAccesses)
+ throws InterruptedException
{
DistributedQueryRunner queryRunner = getDistributedQueryRunner();
queryRunner.executeWithPlan(queryRunner.getDefaultSession(), query);
+ // Allow time for table stats computation to finish before validation.
+ Thread.sleep(1000L);
assertMultisetsEqual(getFileOperations(queryRunner), expectedCacheAccesses);
}
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
index 0786e7497455..b91a4b91c892 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiNoCacheFileOperations.java
@@ -26,6 +26,7 @@ import org.intellij.lang.annotations.Language;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.parallel.Execution;
import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
import java.util.Map;
@@ -40,6 +41,7 @@ import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TABLE_PROPER
import static io.trino.testing.MultisetAssertions.assertMultisetsEqual;
import static java.util.stream.Collectors.toCollection;
+@ResourceLock("HUDI_CACHE_SYSTEM")
@Execution(ExecutionMode.SAME_THREAD)
public class TestHudiNoCacheFileOperations
extends AbstractTestQueryFramework
@@ -63,6 +65,7 @@ public class TestHudiNoCacheFileOperations
@Test
public void testSelectWithFilter()
+ throws InterruptedException
{
@Language("SQL") String query = "SELECT * FROM " + HUDI_MULTI_FG_PT_V8_MOR + " WHERE country='SG'";
assertFileSystemAccesses(
@@ -92,6 +95,7 @@ public class TestHudiNoCacheFileOperations
@Test
public void testJoin()
+ throws InterruptedException
{
@Language("SQL") String query = "SELECT t1.id, t1.name, t1.price, t1.ts FROM " +
HUDI_MULTI_FG_PT_V8_MOR + " t1 " +
@@ -122,9 +126,12 @@ public class TestHudiNoCacheFileOperations
}
private void assertFileSystemAccesses(@Language("SQL") String query, Multiset<FileOperationUtils.FileOperation> expectedCacheAccesses)
+ throws InterruptedException
{
DistributedQueryRunner queryRunner = getDistributedQueryRunner();
queryRunner.executeWithPlan(queryRunner.getDefaultSession(), query);
+ // Allow time for table stats computation to finish before validation.
+ Thread.sleep(1000L);
assertMultisetsEqual(getFileOperations(queryRunner), expectedCacheAccesses);
}
(hudi) 05/13: [MINOR] Added optimizations for HudiColumnStatsIndexSupport (#47)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d34aaa3a1b9633d960a327f180fd5250af8d68c5
Author: voonhous <vo...@gmail.com>
AuthorDate: Wed Jul 30 08:48:56 2025 +0800
[MINOR] Added optimizations for HudiColumnStatsIndexSupport (#47)
Co-authored-by: Y Ethan Guo <et...@gmail.com>
(cherry picked from commit c81def9fd40eea39996ac126f6afdb83cfcbfbca)
---
.../query/index/HudiColumnStatsIndexSupport.java | 62 ++++++++++++----------
.../index/HudiPartitionStatsIndexSupport.java | 19 ++++---
2 files changed, 47 insertions(+), 34 deletions(-)
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java
index 72e46693ee92..1f30f3d21743 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiColumnStatsIndexSupport.java
@@ -42,6 +42,7 @@ import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.metadata.HoodieTableMetadataUtil;
import org.apache.hudi.util.Lazy;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -49,7 +50,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import java.util.function.Function;
import java.util.stream.Collectors;
import static io.trino.parquet.predicate.PredicateUtils.isStatisticsOverflow;
@@ -68,7 +68,8 @@ public class HudiColumnStatsIndexSupport
extends HudiBaseIndexSupport
{
private static final Logger log = Logger.get(HudiColumnStatsIndexSupport.class);
- private final CompletableFuture<Optional<Map<String, Map<String, HoodieMetadataColumnStats>>>> statsByFileNameFuture;
+ // file name -> column name -> domain with column stats
+ private final CompletableFuture<Optional<Map<String, Map<String, Domain>>>> domainsWithStatsFuture;
protected final TupleDomain<String> regularColumnPredicates;
private final List<String> regularColumns;
private final Duration columnStatsWaitTimeout;
@@ -84,24 +85,29 @@ public class HudiColumnStatsIndexSupport
super(log, schemaTableName, lazyMetaClient);
this.columnStatsWaitTimeout = getColumnStatsWaitTimeout(session);
this.regularColumnPredicates = regularColumnPredicates.transformKeys(HiveColumnHandle::getName);
- this.regularColumns = this.regularColumnPredicates
- .getDomains().get().entrySet().stream().map(Map.Entry::getKey).collect(Collectors.toList());
- if (regularColumnPredicates.isAll() || !regularColumnPredicates.getDomains().isPresent()) {
- this.statsByFileNameFuture = CompletableFuture.completedFuture(Optional.empty());
+ this.regularColumns = this.regularColumnPredicates.getDomains()
+ .map(domains -> new ArrayList<>(domains.keySet()))
+ .orElse(new ArrayList<>());
+ if (regularColumnPredicates.isAll() || regularColumnPredicates.getDomains().isEmpty()) {
+ this.domainsWithStatsFuture = CompletableFuture.completedFuture(Optional.empty());
}
else {
// Get filter columns
List<String> encodedTargetColumnNames = regularColumns
.stream()
.map(col -> new ColumnIndexID(col).asBase64EncodedString()).collect(Collectors.toList());
- statsByFileNameFuture = CompletableFuture.supplyAsync(() -> {
+
+ Map<String, Type> columnTypes = regularColumnPredicates.getDomains().get().entrySet().stream()
+ .collect(Collectors.toMap(entry -> entry.getKey().getName(), entry -> entry.getValue().getType()));
+
+ domainsWithStatsFuture = CompletableFuture.supplyAsync(() -> {
HoodieTimer timer = HoodieTimer.start();
if (!lazyMetaClient.get().getTableConfig().getMetadataPartitions()
.contains(HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS)) {
return Optional.empty();
}
- Map<String, Map<String, HoodieMetadataColumnStats>> statsByFileName =
+ Map<String, Map<String, Domain>> domainsWithStats =
lazyTableMetadata.get().getRecordsByKeyPrefixes(encodedTargetColumnNames,
HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS, true)
.collectAsList()
@@ -112,10 +118,12 @@ public class HudiColumnStatsIndexSupport
HoodieMetadataColumnStats::getFileName,
Collectors.toMap(
HoodieMetadataColumnStats::getColumnName,
- Function.identity())));
- log.debug("Column stats lookup took %s ms and identified %d relevant file IDs.", timer.endTimer(), statsByFileName.size());
+ // Pre-compute the Domain object for each HoodieMetadataColumnStats
+ stats -> getDomainFromColumnStats(stats.getColumnName(), columnTypes.get(stats.getColumnName()), stats))));
+
+ log.debug("Column stats lookup took %s ms and identified %d relevant file IDs.", timer.endTimer(), domainsWithStats.size());
- return Optional.of(statsByFileName);
+ return Optional.of(domainsWithStats);
});
}
this.futureStartTimeMs = System.currentTimeMillis();
@@ -125,10 +133,10 @@ public class HudiColumnStatsIndexSupport
public boolean shouldSkipFileSlice(FileSlice slice)
{
try {
- if (statsByFileNameFuture.isDone()) {
- Optional<Map<String, Map<String, HoodieMetadataColumnStats>>> statsOpt = statsByFileNameFuture.get();
- return statsOpt
- .map(stats -> shouldSkipFileSlice(slice, stats, regularColumnPredicates, regularColumns))
+ if (domainsWithStatsFuture.isDone()) {
+ Optional<Map<String, Map<String, Domain>>> domainsWithStatsOpt = domainsWithStatsFuture.get();
+ return domainsWithStatsOpt
+ .map(domainsWithStats -> shouldSkipFileSlice(slice, domainsWithStats, regularColumnPredicates, regularColumns))
.orElse(false);
}
@@ -140,8 +148,8 @@ public class HudiColumnStatsIndexSupport
// If still within the timeout window, wait up to the remaining time
long remainingMs = Math.max(0, columnStatsWaitTimeout.toMillis() - elapsedMs);
- Optional<Map<String, Map<String, HoodieMetadataColumnStats>>> statsOpt =
- statsByFileNameFuture.get(remainingMs, TimeUnit.MILLISECONDS);
+ Optional<Map<String, Map<String, Domain>>> statsOpt =
+ domainsWithStatsFuture.get(remainingMs, TimeUnit.MILLISECONDS);
return statsOpt
.map(stats -> shouldSkipFileSlice(slice, stats, regularColumnPredicates, regularColumns))
@@ -192,22 +200,22 @@ public class HudiColumnStatsIndexSupport
// TODO: Move helper functions below to TupleDomain/DomainUtils
private static boolean shouldSkipFileSlice(
FileSlice fileSlice,
- Map<String, Map<String, HoodieMetadataColumnStats>> statsByFileName,
+ Map<String, Map<String, Domain>> domainsWithStats,
TupleDomain<String> regularColumnPredicates,
List<String> regularColumns)
{
String fileSliceName = fileSlice.getBaseFile().map(BaseFile::getFileName).orElse("");
// If no stats exist for this specific file, we cannot prune it.
- if (!statsByFileName.containsKey(fileSliceName)) {
+ if (!domainsWithStats.containsKey(fileSliceName)) {
return false;
}
- Map<String, HoodieMetadataColumnStats> stats = statsByFileName.get(fileSliceName);
- return !evaluateStatisticPredicate(regularColumnPredicates, stats, regularColumns);
+ Map<String, Domain> fileDomainsWithStats = domainsWithStats.get(fileSliceName);
+ return !evaluateStatisticPredicate(regularColumnPredicates, fileDomainsWithStats, regularColumns);
}
protected static boolean evaluateStatisticPredicate(
TupleDomain<String> regularColumnPredicates,
- Map<String, HoodieMetadataColumnStats> stats,
+ Map<String, Domain> domainsWithStats,
List<String> regularColumns)
{
if (regularColumnPredicates.isNone() || !regularColumnPredicates.getDomains().isPresent()) {
@@ -215,12 +223,12 @@ public class HudiColumnStatsIndexSupport
}
for (String regularColumn : regularColumns) {
Domain columnPredicate = regularColumnPredicates.getDomains().get().get(regularColumn);
- Optional<HoodieMetadataColumnStats> currentColumnStats = Optional.ofNullable(stats.get(regularColumn));
+ Optional<Domain> currentColumnStats = Optional.ofNullable(domainsWithStats.get(regularColumn));
if (currentColumnStats.isEmpty()) {
// No stats for column
}
else {
- Domain domain = getDomain(regularColumn, columnPredicate.getType(), currentColumnStats.get());
+ Domain domain = currentColumnStats.get();
if (columnPredicate.intersect(domain).isNone()) {
return false;
}
@@ -229,7 +237,7 @@ public class HudiColumnStatsIndexSupport
return true;
}
- private static Domain getDomain(String colName, Type type, HoodieMetadataColumnStats statistics)
+ static Domain getDomainFromColumnStats(String colName, Type type, HoodieMetadataColumnStats statistics)
{
if (statistics == null) {
return Domain.all(type);
@@ -243,7 +251,7 @@ public class HudiColumnStatsIndexSupport
!(statistics.getMaxValue() instanceof GenericRecord)) {
return Domain.all(type);
}
- return getDomain(colName, type, ((GenericRecord) statistics.getMinValue()).get(0),
+ return getDomainFromColumnStats(colName, type, ((GenericRecord) statistics.getMinValue()).get(0),
((GenericRecord) statistics.getMaxValue()).get(0), hasNullValue);
}
@@ -251,7 +259,7 @@ public class HudiColumnStatsIndexSupport
* Get a domain for the ranges defined by each pair of elements from {@code minimums} and {@code maximums}.
* Both arrays must have the same length.
*/
- private static Domain getDomain(String colName, Type type, Object minimum, Object maximum, boolean hasNullValue)
+ private static Domain getDomainFromColumnStats(String colName, Type type, Object minimum, Object maximum, boolean hasNullValue)
{
try {
if (type.equals(BOOLEAN)) {
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiPartitionStatsIndexSupport.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiPartitionStatsIndexSupport.java
index c2e98820fb6f..21dbdf57f262 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiPartitionStatsIndexSupport.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/query/index/HudiPartitionStatsIndexSupport.java
@@ -18,7 +18,9 @@ import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hudi.util.TupleDomainUtils;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.predicate.Domain;
import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.type.Type;
import org.apache.hudi.avro.model.HoodieMetadataColumnStats;
import org.apache.hudi.common.model.HoodieIndexDefinition;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -32,7 +34,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.function.Function;
import java.util.stream.Collectors;
import static io.trino.plugin.hudi.util.TupleDomainUtils.hasSimpleNullCheck;
@@ -70,8 +71,11 @@ public class HudiPartitionStatsIndexSupport
.stream()
.map(col -> new ColumnIndexID(col).asBase64EncodedString()).toList();
- // Map of partition stats keyed by partition name
- Map<String, Map<String, HoodieMetadataColumnStats>> statsByPartitionName = lazyMetadataTable.get().getRecordsByKeyPrefixes(
+ Map<String, Type> columnTypes = regularColumnPredicates.getDomains().get().entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getType()));
+
+ // Map of domains with partition stats keyed by partition name and column name
+ Map<String, Map<String, Domain>> domainsWithStats = lazyMetadataTable.get().getRecordsByKeyPrefixes(
encodedTargetColumnNames,
HoodieTableMetadataUtil.PARTITION_NAME_PARTITION_STATS, true)
.collectAsList()
@@ -82,14 +86,15 @@ public class HudiPartitionStatsIndexSupport
HoodieMetadataColumnStats::getFileName,
Collectors.toMap(
HoodieMetadataColumnStats::getColumnName,
- Function.identity())));
+ // Pre-compute the Domain object for each HoodieMetadataColumnStats
+ stats -> getDomainFromColumnStats(stats.getColumnName(), columnTypes.get(stats.getColumnName()), stats))));
// For each partition, determine if it should be kept based on stats availability and predicate evaluation
List<String> prunedPartitions = allPartitions.stream()
.filter(partition -> {
// Check if stats exist for this partition
- Map<String, HoodieMetadataColumnStats> partitionStats = statsByPartitionName.get(partition);
- if (partitionStats == null) {
+ Map<String, Domain> partitionDomainsWithStats = domainsWithStats.get(partition);
+ if (partitionDomainsWithStats == null) {
// Partition has no stats in the index, keep it
return true;
}
@@ -99,7 +104,7 @@ public class HudiPartitionStatsIndexSupport
// Important: If some columns in encodedTargetColumnNames is not available in partition stats, partition will not be pruned iff all available predicate
// evaluates to true. Since we cannot determine if the predicate will evaluate to true or not on the missing stat, adopt conservative measure to true,
// i.e. to not prune
- return evaluateStatisticPredicate(filteredRegularPredicates, partitionStats, regularColumns);
+ return evaluateStatisticPredicate(filteredRegularPredicates, partitionDomainsWithStats, regularColumns);
}
})
.collect(Collectors.toList());
(hudi) 03/13: [ENG-29373] Fix split generation parallelism for a non-partitioned table (#48)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 0c33fd3b3d7e938ba6710269681077235642292b
Author: vamsikarnika <va...@gmail.com>
AuthorDate: Thu Jul 17 09:05:52 2025 +0530
[ENG-29373] Fix split generation parallelism for a non-partitioned table (#48)
Co-authored-by: Vamsi <va...@onehouse.ai>
(cherry picked from commit 03898bda6b63c64885d735dedc7ffd6446a70924)
---
.../src/main/java/io/trino/plugin/hudi/HudiSplitSource.java | 4 +---
.../trino/plugin/hudi/split/HudiBackgroundSplitLoader.java | 13 +++++++++----
2 files changed, 10 insertions(+), 7 deletions(-)
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
index bfa356fc5c27..f7e773d983f2 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
@@ -16,7 +16,6 @@ package io.trino.plugin.hudi;
import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.Futures;
-import io.airlift.concurrent.BoundedExecutor;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
import io.trino.filesystem.cache.CachingHostAddressProvider;
@@ -62,7 +61,6 @@ import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static io.airlift.concurrent.MoreFutures.toCompletableFuture;
import static io.trino.plugin.hudi.HudiErrorCode.HUDI_CANNOT_OPEN_SPLIT;
import static io.trino.plugin.hudi.HudiSessionProperties.getMinimumAssignedSplitWeight;
-import static io.trino.plugin.hudi.HudiSessionProperties.getSplitGeneratorParallelism;
import static io.trino.plugin.hudi.HudiSessionProperties.getStandardSplitWeightSize;
import static io.trino.plugin.hudi.HudiSessionProperties.isHudiMetadataTableEnabled;
import static io.trino.plugin.hudi.HudiSessionProperties.isSizeBasedSplitWeightsEnabled;
@@ -118,7 +116,7 @@ public class HudiSplitSource
tableHandle,
hudiDirectoryLister,
queue,
- new BoundedExecutor(executor, getSplitGeneratorParallelism(session)),
+ executor,
createSplitWeightProvider(session),
lazyPartitions,
enableMetadataTable,
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
index df5436d1579b..c4c91ccadd71 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
@@ -15,6 +15,7 @@ package io.trino.plugin.hudi.split;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import io.airlift.concurrent.BoundedExecutor;
import io.airlift.log.Logger;
import io.trino.filesystem.cache.CachingHostAddressProvider;
import io.trino.metastore.Partition;
@@ -39,6 +40,7 @@ import java.util.Optional;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
@@ -55,7 +57,7 @@ public class HudiBackgroundSplitLoader
private final HudiTableHandle tableHandle;
private final HudiDirectoryLister hudiDirectoryLister;
private final AsyncQueue<ConnectorSplit> asyncQueue;
- private final Executor splitGeneratorExecutor;
+ private final Executor executor;
private final int splitGeneratorNumThreads;
private final HudiSplitFactory hudiSplitFactory;
private final Lazy<List<String>> lazyPartitions;
@@ -68,7 +70,7 @@ public class HudiBackgroundSplitLoader
HudiTableHandle tableHandle,
HudiDirectoryLister hudiDirectoryLister,
AsyncQueue<ConnectorSplit> asyncQueue,
- Executor splitGeneratorExecutor,
+ ExecutorService executor,
HudiSplitWeightProvider hudiSplitWeightProvider,
Lazy<Map<String, Partition>> lazyPartitionMap,
boolean enableMetadataTable,
@@ -79,11 +81,11 @@ public class HudiBackgroundSplitLoader
this.tableHandle = requireNonNull(tableHandle, "tableHandle is null");
this.hudiDirectoryLister = requireNonNull(hudiDirectoryLister, "hudiDirectoryLister is null");
this.asyncQueue = requireNonNull(asyncQueue, "asyncQueue is null");
- this.splitGeneratorExecutor = requireNonNull(splitGeneratorExecutor, "splitGeneratorExecutorService is null");
this.splitGeneratorNumThreads = getSplitGeneratorParallelism(session);
this.hudiSplitFactory = new HudiSplitFactory(tableHandle, hudiSplitWeightProvider, getTargetSplitSize(session), cachingHostAddressProvider);
this.lazyPartitions = Lazy.lazily(() -> requireNonNull(lazyPartitionMap, "partitions is null").get().keySet().stream().toList());
this.enableMetadataTable = enableMetadataTable;
+ this.executor = requireNonNull(executor, "executor is null");
this.errorListener = requireNonNull(errorListener, "errorListener is null");
SchemaTableName schemaTableName = tableHandle.getSchemaTableName();
this.partitionIndexSupportOpt = enableMetadataTable ?
@@ -118,7 +120,10 @@ public class HudiBackgroundSplitLoader
List<HudiPartitionInfoLoader> splitGenerators = new ArrayList<>();
List<ListenableFuture<Void>> futures = new ArrayList<>();
- for (int i = 0; i < splitGeneratorNumThreads; i++) {
+ int splitGeneratorParallelism = Math.min(splitGeneratorNumThreads, lazyPartitions.get().size());
+ Executor splitGeneratorExecutor = new BoundedExecutor(executor, splitGeneratorParallelism);
+
+ for (int i = 0; i < splitGeneratorParallelism; i++) {
HudiPartitionInfoLoader generator = new HudiPartitionInfoLoader(hudiDirectoryLister, tableHandle.getLatestCommitTime(), hudiSplitFactory,
asyncQueue, partitionQueue, useIndex);
splitGenerators.add(generator);
(hudi) 07/13: [ENG-30914] Fix flakiness when testing cache correctness (#60)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 4f4bfc0fd883175c42dd81d276e8d83dd1b181a1
Author: voonhous <vo...@gmail.com>
AuthorDate: Wed Aug 6 03:00:20 2025 +0800
[ENG-30914] Fix flakiness when testing cache correctness (#60)
(cherry picked from commit 2f3887d9368aba06b08bad0ebbac1229f14b2839)
---
.../java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java | 2 ++
.../test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java | 2 ++
.../java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java | 2 ++
3 files changed, 6 insertions(+)
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
index 7101cb897d77..82e03799d1ed 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
@@ -26,6 +26,7 @@ import org.intellij.lang.annotations.Language;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.parallel.Execution;
import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -43,6 +44,7 @@ import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TABLE_PROPER
import static io.trino.testing.MultisetAssertions.assertMultisetsEqual;
import static java.util.stream.Collectors.toCollection;
+@ResourceLock("HUDI_CACHE_SYSTEM")
@Execution(ExecutionMode.SAME_THREAD)
public class TestHudiAlluxioCacheFileOperations
extends AbstractTestQueryFramework
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java
index 0a082e1cb159..9c191a20e34c 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java
@@ -16,6 +16,7 @@ package io.trino.plugin.hudi;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.Closer;
import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.parallel.ResourceLock;
import java.io.IOException;
import java.nio.file.Files;
@@ -24,6 +25,7 @@ import java.nio.file.Path;
import static com.google.common.io.MoreFiles.deleteRecursively;
import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
+@ResourceLock("HUDI_CACHE_SYSTEM")
public class TestHudiAlluxioCachingSmokeTest
extends TestHudiSmokeTest
{
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
index 707a53b22ed4..5799ce17893c 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
@@ -26,6 +26,7 @@ import org.intellij.lang.annotations.Language;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.parallel.Execution;
import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
import java.util.Map;
@@ -40,6 +41,7 @@ import static io.trino.plugin.hudi.util.FileOperationUtils.FileType.TABLE_PROPER
import static io.trino.testing.MultisetAssertions.assertMultisetsEqual;
import static java.util.stream.Collectors.toCollection;
+@ResourceLock("HUDI_CACHE_SYSTEM")
@Execution(ExecutionMode.SAME_THREAD)
public class TestHudiMemoryCacheFileOperations
extends AbstractTestQueryFramework
(hudi) 01/13: [ENG-28391] Increase Default MaxOutstandingSplits and SplitLoaderParallelism (#45)
Posted by vh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vhs pushed a commit to branch HUDI-9767__upstream-trino-changes
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit ac351f1b72dd9f4d3442c9f395a776f413c1ff9a
Author: vamsikarnika <va...@gmail.com>
AuthorDate: Wed Jul 2 04:09:13 2025 +0530
[ENG-28391] Increase Default MaxOutstandingSplits and SplitLoaderParallelism (#45)
Co-authored-by: Vamsi <va...@onehouse.ai>
(cherry picked from commit ea7f22d0371173a31be0c693a24fa00b7374fe0f)
---
hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java | 4 ++--
.../src/test/java/io/trino/plugin/hudi/TestHudiConfig.java | 4 ++--
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
index 5138caea39c8..4a595f5044fe 100644
--- a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
+++ b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/HudiConfig.java
@@ -48,8 +48,8 @@ public class HudiConfig
private double minimumAssignedSplitWeight = 0.05;
private DataSize targetSplitSize = DataSize.of(128, MEGABYTE);
private int maxSplitsPerSecond = Integer.MAX_VALUE;
- private int maxOutstandingSplits = 1000;
- private int splitLoaderParallelism = 4;
+ private int maxOutstandingSplits = 10000;
+ private int splitLoaderParallelism = 10;
private int splitGeneratorParallelism = 4;
private long perTransactionMetastoreCacheMaximumSize = 2000;
private boolean queryPartitionFilterRequired;
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
index 5e07fedd920b..c618559cdf67 100644
--- a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
+++ b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/TestHudiConfig.java
@@ -43,8 +43,8 @@ public class TestHudiConfig
.setMinimumAssignedSplitWeight(0.05)
.setTargetSplitSize(DataSize.of(128, MEGABYTE))
.setMaxSplitsPerSecond(Integer.MAX_VALUE)
- .setMaxOutstandingSplits(1000)
- .setSplitLoaderParallelism(4)
+ .setMaxOutstandingSplits(10000)
+ .setSplitLoaderParallelism(10)
.setSplitGeneratorParallelism(4)
.setPerTransactionMetastoreCacheMaximumSize(2000)
.setQueryPartitionFilterRequired(false)