feat(trino): resolve merge-required columns from the table schema

hudi-trino/pom.xml

@@ -212,6 +212,11 @@
             <artifactId>trino-hive</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-hive-formats</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>io.trino</groupId>
             <artifactId>trino-memory-context</artifactId>
@@ -416,12 +421,6 @@
             <scope>runtime</scope>
         </dependency>
 
-        <dependency>
-            <groupId>io.trino</groupId>
-            <artifactId>trino-hive-formats</artifactId>
-            <scope>runtime</scope>
-        </dependency>
-
         <dependency>
             <groupId>org.jetbrains</groupId>
             <artifactId>annotations</artifactId>

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

@@ -15,7 +15,7 @@
 
 import com.google.common.collect.ImmutableList;
 import io.trino.plugin.hive.HiveColumnHandle;
-import io.trino.plugin.hudi.util.SynthesizedColumnHandler;
+import io.trino.plugin.hudi.util.PrefilledColumnValues;
 import io.trino.spi.Page;
 import io.trino.spi.block.Block;
 import io.trino.spi.connector.ConnectorPageSource;
@@ -38,21 +38,22 @@ public class HudiBaseFileOnlyPageSource
 {
     private final ConnectorPageSource dataPageSource;
     private final List<HiveColumnHandle> allOutputColumns;
-    private final SynthesizedColumnHandler synthesizedColumnHandler;
-    // Maps output channel to physical source channel, or -1 if synthesized
+    private final PrefilledColumnValues prefilledColumnValues;
+    // Maps output channel to physical source channel, or -1 if prefilled
     private final int[] physicalSourceChannelMap;
 
     public HudiBaseFileOnlyPageSource(
             ConnectorPageSource dataPageSource,
             List<HiveColumnHandle> allOutputColumns,
             // Columns provided by dataPageSource
             List<HiveColumnHandle> dataColumns,
-            // Handler to manage synthesized/virtual in Hudi tables such as partition columns and metadata, i.e. file size (not hudi metadata)
-            SynthesizedColumnHandler synthesizedColumnHandler)
+            // Per-split constant values for columns not present in the data file, such as partition
+            // columns and Trino's hidden metadata columns, e.g. file size (not hudi metadata)
+            PrefilledColumnValues prefilledColumnValues)
     {
         this.dataPageSource = requireNonNull(dataPageSource, "dataPageSource is null");
         this.allOutputColumns = ImmutableList.copyOf(requireNonNull(allOutputColumns, "allOutputColumns is null"));
-        this.synthesizedColumnHandler = requireNonNull(synthesizedColumnHandler, "synthesizedColumnHandler is null");
+        this.prefilledColumnValues = requireNonNull(prefilledColumnValues, "prefilledColumnValues is null");
 
         // Create a mapping from the channel index in the output page to the channel index in the physicalDataPageSource's page
         this.physicalSourceChannelMap = new int[allOutputColumns.size()];
@@ -93,11 +94,6 @@ public SourcePage getNextSourcePage()
         }
 
         int positionCount = physicalSourcePage.getPositionCount();
-        if (positionCount == 0 && synthesizedColumnHandler.getSynthesizedColumnCount() == 0) {
-            // If only physical columns and page is empty
-            return physicalSourcePage;
-        }
-
         if (allOutputColumns.isEmpty()) {
             // Forward the zero-block page so positionCount survives -- new Page(new Block[0]) would infer positionCount=0.
             return physicalSourcePage;
@@ -110,8 +106,8 @@ public SourcePage getNextSourcePage()
                 outputBlocks[i] = physicalSourcePage.getBlock(physicalSourceChannelMap[i]);
             }
             else {
-                // Column is synthesized
-                outputBlocks[i] = synthesizedColumnHandler.createRleSynthesizedBlock(outputColumn, positionCount);
+                // Column is not in the data file; fill with the split's constant value
+                outputBlocks[i] = prefilledColumnValues.toRleBlock(outputColumn, positionCount);
             }
         }
         return SourcePage.create(new Page(outputBlocks));

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

@@ -14,9 +14,8 @@
 package io.trino.plugin.hudi;
 
 import io.trino.plugin.hive.HiveColumnHandle;
-import io.trino.plugin.hudi.reader.HudiTrinoReaderContext;
 import io.trino.plugin.hudi.util.HudiAvroSerializer;
-import io.trino.plugin.hudi.util.SynthesizedColumnHandler;
+import io.trino.plugin.hudi.util.PrefilledColumnValues;
 import io.trino.spi.Page;
 import io.trino.spi.PageBuilder;
 import io.trino.spi.TrinoException;
@@ -39,28 +38,23 @@
 public class HudiPageSource
         implements ConnectorPageSource
 {
-    HoodieFileGroupReader<IndexedRecord> fileGroupReader;
-    // TODO: Remove pageSource here, Hudi doesn't use this page source to read
-    ConnectorPageSource pageSource;
-    HudiTrinoReaderContext readerContext;
-    PageBuilder pageBuilder;
-    HudiAvroSerializer avroSerializer;
-    List<HiveColumnHandle> columnHandles;
-    ClosableIterator<IndexedRecord> recordIterator;
+    private final HoodieFileGroupReader<IndexedRecord> fileGroupReader;
+    // Reads flow through fileGroupReader; pageSource is kept for stats/isBlocked delegation
+    private final ConnectorPageSource pageSource;
+    private final PageBuilder pageBuilder;
+    private final HudiAvroSerializer avroSerializer;
+    private final ClosableIterator<IndexedRecord> recordIterator;
 
     public HudiPageSource(
             ConnectorPageSource pageSource,
             HoodieFileGroupReader<IndexedRecord> fileGroupReader,
-            HudiTrinoReaderContext readerContext,
             List<HiveColumnHandle> columnHandles,
-            SynthesizedColumnHandler synthesizedColumnHandler)
+            PrefilledColumnValues prefilledColumnValues)
     {
         this.pageSource = pageSource;
         this.fileGroupReader = fileGroupReader;
-        this.readerContext = readerContext;
-        this.columnHandles = columnHandles;
         this.pageBuilder = new PageBuilder(columnHandles.stream().map(HiveColumnHandle::getType).toList());
-        this.avroSerializer = new HudiAvroSerializer(columnHandles, synthesizedColumnHandler);
+        this.avroSerializer = new HudiAvroSerializer(columnHandles, prefilledColumnValues);
         try {
             this.recordIterator = fileGroupReader.getClosableIterator();
         }