Support MOR snapshot query and data skipping

plugin/trino-hudi/pom.xml

@@ -132,88 +132,32 @@
 
         <dependency>
             <groupId>org.apache.hudi</groupId>
-            <artifactId>hudi-common</artifactId>
+            <artifactId>hudi-trino-bundle</artifactId>
             <version>${dep.hudi.version}</version>
             <exclusions>
                 <exclusion>
-                    <groupId>org.apache.hbase</groupId>
-                    <artifactId>hbase-server</artifactId>
+                    <groupId>com.google.protobuf</groupId>
+                    <artifactId>protobuf-java</artifactId>
                 </exclusion>
                 <exclusion>
-                    <groupId>org.apache.hbase</groupId>
-                    <artifactId>hbase-client</artifactId>
+                    <groupId>commons-lang</groupId>
+                    <artifactId>commons-lang</artifactId>
                 </exclusion>
                 <exclusion>
-                    <groupId>org.osgi</groupId>
-                    <artifactId>org.osgi.core</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.apache.orc</groupId>
-                    <artifactId>orc-core</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>com.fasterxml.jackson.core</groupId>
-                    <artifactId>jackson-annotations</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>com.fasterxml.jackson.core</groupId>
-                    <artifactId>jackson-databind</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.apache.httpcomponents</groupId>
-                    <artifactId>httpclient</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.apache.httpcomponents</groupId>
-                    <artifactId>fluent-hc</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.rocksdb</groupId>
-                    <artifactId>rocksdbjni</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>com.esotericsoftware</groupId>
-                    <artifactId>kryo-shaded</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.apache.hadoop</groupId>
-                    <artifactId>hadoop-client</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.apache.hadoop</groupId>
-                    <artifactId>hadoop-hdfs</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.apache.httpcomponents</groupId>
-                    <artifactId>httpcore</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.apache.hive</groupId>
-                    <artifactId>hive-exec</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.apache.hive</groupId>
-                    <artifactId>hive-jdbc</artifactId>
+                    <groupId>org.apache.hudi</groupId>
+                    <artifactId>hudi-common</artifactId>
                 </exclusion>
                 <exclusion>
-                    <groupId>com.github.ben-manes.caffeine</groupId>
-                    <artifactId>caffeine</artifactId>
+                    <groupId>org.apache.hudi</groupId>
+                    <artifactId>hudi-hadoop-mr-bundle</artifactId>
                 </exclusion>
                 <exclusion>
-                    <groupId>org.lz4</groupId>
-                    <artifactId>lz4-java</artifactId>
+                    <groupId>org.apache.parquet</groupId>
+                    <artifactId>parquet-avro</artifactId>
                 </exclusion>
-            </exclusions>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.hudi</groupId>
-            <artifactId>hudi-hadoop-mr</artifactId>
-            <version>${dep.hudi.version}</version>
-            <exclusions>
                 <exclusion>
-                    <groupId>*</groupId>
-                    <artifactId>*</artifactId>
+                    <groupId>org.apache.avro</groupId>
+                    <artifactId>avro</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,11 +17,11 @@
 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;
 import javax.validation.constraints.DecimalMin;
-import javax.validation.constraints.Max;
 import javax.validation.constraints.Min;
 import javax.validation.constraints.NotNull;
 
@@ -39,13 +39,15 @@ public class HudiConfig
     private List<String> columnsToHide = ImmutableList.of();
     private boolean metadataEnabled;
     private boolean shouldUseParquetColumnNames = true;
-    private int minPartitionBatchSize = 10;
-    private int maxPartitionBatchSize = 100;
     private boolean sizeBasedSplitWeightsEnabled = true;
     private DataSize standardSplitWeightSize = DataSize.of(128, MEGABYTE);
     private double minimumAssignedSplitWeight = 0.05;
     private int maxSplitsPerSecond = Integer.MAX_VALUE;
     private int maxOutstandingSplits = 1000;
+    private int splitLoaderParallelism = 2;
+    private int splitGeneratorParallelism = 16;
+    private long perTransactionMetastoreCacheMaximumSize = 2000;
+    private String fileSystemViewSpillableDirectory = "/tmp";
 
     public List<String> getColumnsToHide()
     {
@@ -90,36 +92,6 @@ public boolean getUseParquetColumnNames()
         return this.shouldUseParquetColumnNames;
     }
 
-    @Config("hudi.min-partition-batch-size")
-    @ConfigDescription("Minimum number of partitions returned in a single batch.")
-    public HudiConfig setMinPartitionBatchSize(int minPartitionBatchSize)
-    {
-        this.minPartitionBatchSize = minPartitionBatchSize;
-        return this;
-    }
-
-    @Min(1)
-    @Max(100)
-    public int getMinPartitionBatchSize()
-    {
-        return minPartitionBatchSize;
-    }
-
-    @Config("hudi.max-partition-batch-size")
-    @ConfigDescription("Maximum number of partitions returned in a single batch.")
-    public HudiConfig setMaxPartitionBatchSize(int maxPartitionBatchSize)
-    {
-        this.maxPartitionBatchSize = maxPartitionBatchSize;
-        return this;
-    }
-
-    @Min(1)
-    @Max(1000)
-    public int getMaxPartitionBatchSize()
-    {
-        return maxPartitionBatchSize;
-    }
-
     @Config("hudi.size-based-split-weights-enabled")
     @ConfigDescription("Unlike uniform splitting, size-based splitting ensures that each batch of splits has enough data to process. " +
             "By default, it is enabled to improve performance.")
@@ -191,4 +163,60 @@ public HudiConfig setMaxOutstandingSplits(int maxOutstandingSplits)
         this.maxOutstandingSplits = maxOutstandingSplits;
         return this;
     }
+
+    @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;
+    }
+
+    @Min(1)
+    public long getPerTransactionMetastoreCacheMaximumSize()
+    {
+        return perTransactionMetastoreCacheMaximumSize;
+    }
+
+    @LegacyConfig("hive.per-transaction-metastore-cache-maximum-size")
+    @Config("hudi.per-transaction-metastore-cache-maximum-size")
+    public HudiConfig setPerTransactionMetastoreCacheMaximumSize(long perTransactionMetastoreCacheMaximumSize)
+    {
+        this.perTransactionMetastoreCacheMaximumSize = perTransactionMetastoreCacheMaximumSize;
+        return this;
+    }
+
+    public String getFileSystemViewSpillableDirectory()
+    {
+        return fileSystemViewSpillableDirectory;
+    }
+
+    @Config("huhi.fs-view-spillable-dir")
+    @ConfigDescription("Path on storage to use when the file system view is held in a spillable map.")
+    public HudiConfig setFileSystemViewSpillableDirectory(String fileSystemViewSpillableDirectory)
+    {
+        this.fileSystemViewSpillableDirectory = fileSystemViewSpillableDirectory;
+        return this;
+    }
 }

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

@@ -28,7 +28,8 @@ public enum HudiErrorCode
     HUDI_MISSING_DATA(3, EXTERNAL),
     HUDI_CANNOT_OPEN_SPLIT(4, EXTERNAL),
     HUDI_UNSUPPORTED_FILE_FORMAT(5, EXTERNAL),
-    HUDI_CURSOR_ERROR(6, EXTERNAL);
+    HUDI_CURSOR_ERROR(6, EXTERNAL),
+    HUDI_NO_VALID_COMMIT(7, EXTERNAL);
 
     private final ErrorCode errorCode;