Improve Hudi Perf

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

@@ -590,9 +590,11 @@ private List<DeltaLakeColumnHandle> getColumns(MetadataEntry deltaMetadata)
     @Override
     public TableStatistics getTableStatistics(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
+        LOG.warn("inside getTableStatistics");
         if (!isTableStatisticsEnabled(session)) {
             return TableStatistics.empty();
         }
+        LOG.warn("fetching stats from metastore");
         return metastore.getTableStatistics(session, (DeltaLakeTableHandle) tableHandle);
     }
 
@@ -2612,6 +2614,7 @@ public static TupleDomain<DeltaLakeColumnHandle> createStatisticsPredicate(
             List<DeltaLakeColumnMetadata> schema,
             List<String> canonicalPartitionColumns)
     {
+        LOG.warn("inside createStatisticsPredicate");
         return addFileEntry.getStats()
                 .map(deltaLakeFileStatistics -> withColumnDomains(
                         schema.stream()

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeSplitManager.java

@@ -14,6 +14,7 @@
 package io.trino.plugin.deltalake;
 
 import com.google.common.collect.ImmutableList;
+import io.airlift.log.Logger;
 import io.airlift.units.DataSize;
 import io.trino.plugin.base.classloader.ClassLoaderSafeConnectorSplitSource;
 import io.trino.plugin.deltalake.metastore.DeltaLakeMetastore;
@@ -66,6 +67,8 @@
 public class DeltaLakeSplitManager
         implements ConnectorSplitManager
 {
+    private static final Logger log = Logger.get(DeltaLakeSplitManager.class);
+
     private final TypeManager typeManager;
     private final BiFunction<ConnectorSession, HiveTransactionHandle, DeltaLakeMetastore> metastoreProvider;
     private final ExecutorService executor;
@@ -152,6 +155,7 @@ private Stream<DeltaLakeSplit> getSplits(
                 .filter(column -> predicatedColumnNames.contains(column.getName())) // DeltaLakeColumnMetadata.name is lowercase
                 .collect(toImmutableList());
 
+        log.warn(">>> creating delta splits");
         return validDataFiles.stream()
                 .flatMap(addAction -> {
                     if (tableHandle.getAnalyzeHandle().isPresent() && !tableHandle.getAnalyzeHandle().get().isInitialAnalyze() && !addAction.isDataChange()) {
@@ -198,6 +202,7 @@ private Stream<DeltaLakeSplit> getSplits(
                         }
                     }
 
+                    log.warn(">>> Creating delta splits for stats: " + statisticsPredicate);
                     return splitsForFile(
                             session,
                             addAction,

plugin/trino-hudi/pom.xml

@@ -110,11 +110,6 @@
             <artifactId>guice</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>javax.annotation</groupId>
-            <artifactId>javax.annotation-api</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>javax.inject</groupId>
             <artifactId>javax.inject</artifactId>
@@ -130,7 +125,59 @@
             <artifactId>joda-time</artifactId>
         </dependency>
 
+        <!--<dependency>
+            <groupId>org.apache.hbase</groupId>
+            <artifactId>hbase-common</artifactId>
+            <version>2.4.9</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>*</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>-->
+
+        <!--<dependency>
+            <groupId>org.lz4</groupId>
+            <artifactId>lz4-java</artifactId>
+            <version>1.8.0</version>
+            <scope>provided</scope>
+        </dependency>-->
+
         <dependency>
+            <groupId>org.apache.hudi</groupId>
+            <artifactId>hudi-trino-bundle</artifactId>
+            <version>${dep.hudi.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.protobuf</groupId>
+                    <artifactId>protobuf-java</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>commons-lang</groupId>
+                    <artifactId>commons-lang</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hudi</groupId>
+                    <artifactId>hudi-common</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hudi</groupId>
+                    <artifactId>hudi-hadoop-mr-bundle</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.parquet</groupId>
+                    <artifactId>parquet-avro</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.avro</groupId>
+                    <artifactId>avro</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <!--<dependency>
             <groupId>org.apache.hudi</groupId>
             <artifactId>hudi-common</artifactId>
             <version>${dep.hudi.version}</version>
@@ -216,7 +263,7 @@
                     <artifactId>*</artifactId>
                 </exclusion>
             </exclusions>
-        </dependency>
+        </dependency>-->
 
         <dependency>
             <groupId>org.weakref</groupId>
@@ -403,6 +450,10 @@
                     <groupId>io.dropwizard.metrics</groupId>
                     <artifactId>*</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>org.apache.logging.log4j</groupId>
+                    <artifactId>log4j-api</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
 

plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/ForHudiBackgroundSplitLoader.java

@@ -0,0 +1,32 @@
+/*
+ * 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 javax.inject.Qualifier;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+import static java.lang.annotation.ElementType.FIELD;
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.ElementType.PARAMETER;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+@Retention(RUNTIME)
+@Target({FIELD, PARAMETER, METHOD})
+@Qualifier
+public @interface ForHudiBackgroundSplitLoader
+{
+}

plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/ForHudiSplitSource.java

@@ -0,0 +1,32 @@
+/*
+ * 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 javax.inject.Qualifier;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+import static java.lang.annotation.ElementType.FIELD;
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.ElementType.PARAMETER;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+@Retention(RUNTIME)
+@Target({FIELD, PARAMETER, METHOD})
+@Qualifier
+public @interface ForHudiSplitSource
+{
+}

plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/HudiConfig.java

@@ -17,6 +17,7 @@
 import com.google.common.collect.ImmutableList;
 import io.airlift.configuration.Config;
 import io.airlift.configuration.ConfigDescription;
+import io.airlift.configuration.LegacyConfig;
 import io.airlift.units.DataSize;
 
 import javax.validation.constraints.DecimalMax;
@@ -42,10 +43,15 @@ public class HudiConfig
     private int minPartitionBatchSize = 10;
     private int maxPartitionBatchSize = 100;
     private boolean sizeBasedSplitWeightsEnabled = true;
-    private DataSize standardSplitWeightSize = DataSize.of(128, MEGABYTE);
+    private DataSize standardSplitWeightSize = DataSize.of(64, MEGABYTE);
     private double minimumAssignedSplitWeight = 0.05;
     private int maxSplitsPerSecond = Integer.MAX_VALUE;
     private int maxOutstandingSplits = 1000;
+    private boolean tableStatisticsEnabled;
+    private int splitLoaderParallelism = 2;
+    private int splitGeneratorParallelism = 16;
+    private int partitionScannerParallelism = 16;
+    private long perTransactionMetastoreCacheMaximumSize = 1000;
 
     public List<String> getColumnsToHide()
     {
@@ -99,7 +105,7 @@ public HudiConfig setMinPartitionBatchSize(int minPartitionBatchSize)
     }
 
     @Min(1)
-    @Max(100)
+    @Max(1000)
     public int getMinPartitionBatchSize()
     {
         return minPartitionBatchSize;
@@ -114,7 +120,7 @@ public HudiConfig setMaxPartitionBatchSize(int maxPartitionBatchSize)
     }
 
     @Min(1)
-    @Max(1000)
+    @Max(10000)
     public int getMaxPartitionBatchSize()
     {
         return maxPartitionBatchSize;
@@ -178,7 +184,6 @@ public HudiConfig setMaxSplitsPerSecond(int maxSplitsPerSecond)
         return this;
     }
 
-    @Min(1)
     public int getMaxOutstandingSplits()
     {
         return maxOutstandingSplits;
@@ -191,4 +196,74 @@ public HudiConfig setMaxOutstandingSplits(int maxOutstandingSplits)
         this.maxOutstandingSplits = maxOutstandingSplits;
         return this;
     }
+
+    @Config("hudi.table-statistics-enabled")
+    @ConfigDescription("Enable use of table statistics")
+    public HudiConfig setTableStatisticsEnabled(boolean tableStatisticsEnabled)
+    {
+        this.tableStatisticsEnabled = tableStatisticsEnabled;
+        return this;
+    }
+
+    public boolean isTableStatisticsEnabled()
+    {
+        return tableStatisticsEnabled;
+    }
+
+    @Min(1)
+    public int getSplitGeneratorParallelism()
+    {
+        return splitGeneratorParallelism;
+    }
+
+    @Config("hudi.split-generator-parallelism")
+    @ConfigDescription("Number of threads to generate splits from partitions.")
+    public HudiConfig setSplitGeneratorParallelism(int splitGeneratorParallelism)
+    {
+        this.splitGeneratorParallelism = splitGeneratorParallelism;
+        return this;
+    }
+
+    @Min(1)
+    public int getSplitLoaderParallelism()
+    {
+        return splitLoaderParallelism;
+    }
+
+    @Config("hudi.split-loader-parallelism")
+    @ConfigDescription("Number of threads to run background split loader. "
+            + "A single background split loader is needed per query.")
+    public HudiConfig setSplitLoaderParallelism(int splitLoaderParallelism)
+    {
+        this.splitLoaderParallelism = splitLoaderParallelism;
+        return this;
+    }
+
+    @Config("hudi.partition-scanner-parallelism")
+    @ConfigDescription("Number of threads to use for partition scanners")
+    public HudiConfig setPartitionScannerParallelism(int partitionScannerParallelism)
+    {
+        this.partitionScannerParallelism = partitionScannerParallelism;
+        return this;
+    }
+
+    @Min(1)
+    public int getPartitionScannerParallelism()
+    {
+        return partitionScannerParallelism;
+    }
+
+    @Min(1)
+    public long getPerTransactionMetastoreCacheMaximumSize()
+    {
+        return perTransactionMetastoreCacheMaximumSize;
+    }
+
+    @LegacyConfig("hive.per-transaction-metastore-cache-maximum-size")
+    @Config("delta.per-transaction-metastore-cache-maximum-size")
+    public HudiConfig setPerTransactionMetastoreCacheMaximumSize(long perTransactionMetastoreCacheMaximumSize)
+    {
+        this.perTransactionMetastoreCacheMaximumSize = perTransactionMetastoreCacheMaximumSize;
+        return this;
+    }
 }