Add support for data skipping using column stats

plugin/trino-hive/src/main/java/io/trino/plugin/hive/avro/AvroHiveFileUtils.java

@@ -67,7 +67,7 @@ public final class AvroHiveFileUtils
 {
     private final AtomicInteger recordNameSuffix = new AtomicInteger(0);
 
-    private AvroHiveFileUtils() {}
+    public AvroHiveFileUtils() {}
 
     // Lifted and shifted from org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils.determineSchemaOrThrowException
     public static Schema determineSchemaOrThrowException(TrinoFileSystem fileSystem, Map<String, String> properties)
@@ -146,7 +146,7 @@ private static Schema constructSchemaFromParts(List<String> columnNames, List<Hi
         return fieldBuilder.endRecord();
     }
 
-    private Schema avroSchemaForHiveType(HiveType hiveType)
+    public Schema avroSchemaForHiveType(HiveType hiveType)
     {
         Schema schema = switch (hiveType.getCategory()) {
             case PRIMITIVE -> createAvroPrimitive(hiveType);

plugin/trino-hudi/pom.xml

@@ -19,6 +19,13 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <!--Used to test execution in task executor after de-serializing-->
+            <groupId>com.esotericsoftware</groupId>
+            <artifactId>kryo</artifactId>
+            <version>4.0.2</version>
+        </dependency>
+
         <dependency>
             <groupId>com.google.errorprone</groupId>
             <artifactId>error_prone_annotations</artifactId>

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

@@ -31,12 +31,12 @@
 
 @DefunctConfig({
         "hudi.min-partition-batch-size",
-        "hudi.max-partition-batch-size",
-        "hudi.metadata-enabled",
+        "hudi.max-partition-batch-size"
 })
 public class HudiConfig
 {
     private List<String> columnsToHide = ImmutableList.of();
+    private boolean metadataEnabled = false;
     private boolean shouldUseParquetColumnNames = true;
     private boolean sizeBasedSplitWeightsEnabled = true;
     private DataSize standardSplitWeightSize = DataSize.of(128, MEGABYTE);
@@ -65,6 +65,19 @@ public HudiConfig setColumnsToHide(List<String> columnsToHide)
         return this;
     }
 
+    @Config("hudi.metadata-enabled")
+    @ConfigDescription("Fetch the list of file names and sizes from Hudi metadata table rather than storage.")
+    public HudiConfig setMetadataEnabled(boolean metadataEnabled)
+    {
+        this.metadataEnabled = metadataEnabled;
+        return this;
+    }
+
+    public boolean isMetadataEnabled()
+    {
+        return this.metadataEnabled;
+    }
+
     @Config("hudi.parquet.use-column-names")
     @ConfigDescription("Access Parquet columns using names from the file. If disabled, then columns are accessed using index."
             + "Only applicable to Parquet file format.")

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

@@ -32,6 +32,7 @@ public enum HudiErrorCode
     HUDI_FILESYSTEM_ERROR(7, EXTERNAL),
     HUDI_PARTITION_NOT_FOUND(8, EXTERNAL),
     // HUDI_UNSUPPORTED_TABLE_TYPE(9, EXTERNAL), // Unused. Could be mistaken with HUDI_UNKNOWN_TABLE_TYPE.
+    HUDI_NO_VALID_COMMIT(10, EXTERNAL)
 
     /**/;
 

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

@@ -16,13 +16,16 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
 import io.trino.filesystem.TrinoFileSystemFactory;
 import io.trino.metastore.Column;
 import io.trino.metastore.HiveMetastore;
 import io.trino.metastore.Table;
 import io.trino.metastore.TableInfo;
 import io.trino.plugin.base.classloader.ClassLoaderSafeSystemTable;
 import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hudi.storage.TrinoHudiStorage;
+import io.trino.plugin.hudi.storage.TrinoStorageConfiguration;
 import io.trino.spi.TrinoException;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
@@ -40,6 +43,9 @@
 import io.trino.spi.connector.TableNotFoundException;
 import io.trino.spi.predicate.TupleDomain;
 import io.trino.spi.type.TypeManager;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.storage.StoragePath;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -119,15 +125,20 @@ public HudiTableHandle getTableHandle(ConnectorSession session, SchemaTableName
             throw new TrinoException(UNSUPPORTED_TABLE_TYPE, format("Not a Hudi table: %s", tableName));
         }
         Location location = Location.of(table.get().getStorage().getLocation());
-        if (!hudiMetadataExists(fileSystemFactory.create(session), location)) {
+        TrinoFileSystem fileSystem = fileSystemFactory.create(session);
+        if (!hudiMetadataExists(fileSystem, location)) {
             throw new TrinoException(HUDI_BAD_DATA, "Location of table %s does not contain Hudi table metadata: %s".formatted(tableName, location));
         }
+        StoragePath metaLocation = new StoragePath(table.get().getStorage().getLocation(), HoodieTableMetaClient.METAFOLDER_NAME);
+        HoodieTableConfig tableConfig = new HoodieTableConfig(new TrinoHudiStorage(fileSystem, new TrinoStorageConfiguration()), metaLocation, null, null, null);
+        String preCombineField = tableConfig.getPreCombineField();
 
         return new HudiTableHandle(
                 tableName.getSchemaName(),
                 tableName.getTableName(),
                 table.get().getStorage().getLocation(),
                 COPY_ON_WRITE,
+                preCombineField,
                 getPartitionKeyColumnHandles(table.get(), typeManager),
                 TupleDomain.all(),
                 TupleDomain.all());

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

@@ -0,0 +1,143 @@
+/*
+ * 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 io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hudi.reader.TrinoHudiReaderContext;
+import io.trino.plugin.hudi.util.HudiAvroSerializer;
+import io.trino.plugin.hudi.util.SynthesizedColumnHandler;
+import io.trino.spi.Page;
+import io.trino.spi.PageBuilder;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.SourcePage;
+import io.trino.spi.metrics.Metrics;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hudi.common.table.read.HoodieFileGroupReader;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.OptionalLong;
+import java.util.concurrent.CompletableFuture;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public class HudiPageSource
+        implements ConnectorPageSource
+{
+    HoodieFileGroupReader<IndexedRecord> fileGroupReader;
+    ConnectorPageSource pageSource;
+    TrinoHudiReaderContext readerContext;
+    PageBuilder pageBuilder;
+    HudiAvroSerializer avroSerializer;
+    List<HiveColumnHandle> columnHandles;
+
+    public HudiPageSource(
+            ConnectorPageSource pageSource,
+            HoodieFileGroupReader<IndexedRecord> fileGroupReader,
+            TrinoHudiReaderContext readerContext,
+            List<HiveColumnHandle> columnHandles,
+            SynthesizedColumnHandler synthesizedColumnHandler)
+    {
+        this.pageSource = pageSource;
+        this.fileGroupReader = fileGroupReader;
+        this.initFileGroupReader();
+        this.readerContext = readerContext;
+        this.columnHandles = columnHandles;
+        this.pageBuilder = new PageBuilder(columnHandles.stream().map(HiveColumnHandle::getType).toList());
+        this.avroSerializer = new HudiAvroSerializer(columnHandles, synthesizedColumnHandler);
+    }
+
+    @Override
+    public long getCompletedBytes()
+    {
+        return pageSource.getCompletedBytes();
+    }
+
+    @Override
+    public OptionalLong getCompletedPositions()
+    {
+        return pageSource.getCompletedPositions();
+    }
+
+    @Override
+    public long getReadTimeNanos()
+    {
+        return pageSource.getReadTimeNanos();
+    }
+
+    @Override
+    public boolean isFinished()
+    {
+        try {
+            return !fileGroupReader.hasNext();
+        }
+        catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public SourcePage getNextSourcePage()
+    {
+        checkState(pageBuilder.isEmpty(), "PageBuilder is not empty at the beginning of a new page");
+        try {
+            while (fileGroupReader.hasNext()) {
+                avroSerializer.buildRecordInPage(pageBuilder, fileGroupReader.next());
+            }
+        }
+        catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+
+        Page newPage = pageBuilder.build();
+        pageBuilder.reset();
+        return SourcePage.create(newPage);
+    }
+
+    @Override
+    public long getMemoryUsage()
+    {
+        return pageSource.getMemoryUsage();
+    }
+
+    @Override
+    public void close()
+            throws IOException
+    {
+        fileGroupReader.close();
+        pageSource.close();
+    }
+
+    @Override
+    public CompletableFuture<?> isBlocked()
+    {
+        return pageSource.isBlocked();
+    }
+
+    @Override
+    public Metrics getMetrics()
+    {
+        return pageSource.getMetrics();
+    }
+
+    protected void initFileGroupReader()
+    {
+        try {
+            this.fileGroupReader.initRecordIterators();
+        }
+        catch (IOException e) {
+            throw new RuntimeException("Failed to initialize file group reader!", e);
+        }
+    }
+}