Support 475 Iceberg Glue REST endpoint support

Author: miniway

core/trino-main/src/main/java/io/trino/operator/ChangeOnlyUpdatedColumnsMergeProcessor.java

@@ -14,15 +14,15 @@
 package io.trino.operator;
 
 import io.trino.spi.Page;
+import io.trino.spi.PageBlockUtil;
 import io.trino.spi.block.Block;
-import io.trino.spi.block.ColumnarRow;
 import io.trino.spi.block.RunLengthEncodedBlock;
 
 import java.util.ArrayList;
 import java.util.List;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static io.trino.spi.block.ColumnarRow.toColumnarRow;
+import static io.trino.spi.block.RowBlock.getRowFieldsFromBlock;
 import static io.trino.spi.predicate.Utils.nativeValueToBlock;
 import static io.trino.spi.type.TinyintType.TINYINT;
 import static java.util.Objects.requireNonNull;
@@ -67,26 +67,24 @@ public Page transformPage(Page inputPage)
         // TODO: Check with Karol to see if we can get empty pages
         checkArgument(positionCount > 0, "positionCount should be > 0, but is %s", positionCount);
 
-        ColumnarRow mergeRow = toColumnarRow(inputPage.getBlock(mergeRowChannel));
-        checkArgument(!mergeRow.mayHaveNull(), "The mergeRow may not have null rows");
-
-        // We've verified that the mergeRow block has no null rows, so it's okay to get the field blocks
-
-        List<Block> builder = new ArrayList<>(dataColumnChannels.size() + 3);
-
+        Block mergeRow = inputPage.getBlock(mergeRowChannel).getLoadedBlock();
+        List<Block> fields = getRowFieldsFromBlock(mergeRow);
+        List<Block> builder = new ArrayList<>(dataColumnChannels.size() + 4);
         for (int channel : dataColumnChannels) {
-            builder.add(mergeRow.getField(channel));
+            builder.add(fields.get(channel));
         }
-        Block operationChannelBlock = mergeRow.getField(mergeRow.getFieldCount() - 2);
+        Block operationChannelBlock = fields.get(fields.size() - 2);
         builder.add(operationChannelBlock);
+        Block caseNumberChannelBlock = fields.get(fields.size() - 1);
+        builder.add(caseNumberChannelBlock);
         builder.add(inputPage.getBlock(rowIdChannel));
         builder.add(RunLengthEncodedBlock.create(INSERT_FROM_UPDATE_BLOCK, positionCount));
 
         Page result = new Page(builder.toArray(Block[]::new));
 
         int defaultCaseCount = 0;
         for (int position = 0; position < positionCount; position++) {
-            if (TINYINT.getByte(operationChannelBlock, position) == DEFAULT_CASE_OPERATION_NUMBER) {
+            if (mergeRow.isNull(position)) {
                 defaultCaseCount++;
             }
         }
@@ -97,14 +95,14 @@ public Page transformPage(Page inputPage)
         int usedCases = 0;
         int[] positions = new int[positionCount - defaultCaseCount];
         for (int position = 0; position < positionCount; position++) {
-            if (TINYINT.getByte(operationChannelBlock, position) != DEFAULT_CASE_OPERATION_NUMBER) {
+            if (!mergeRow.isNull(position)) {
                 positions[usedCases] = position;
                 usedCases++;
             }
         }
 
         checkArgument(usedCases + defaultCaseCount == positionCount, "usedCases (%s) + defaultCaseCount (%s) != positionCount (%s)", usedCases, defaultCaseCount, positionCount);
 
-        return result.getPositions(positions, 0, usedCases);
+        return PageBlockUtil.getPositions(result, positions, 0, usedCases);
     }
 }

core/trino-main/src/main/java/io/trino/operator/DeleteAndInsertMergeProcessor.java

@@ -19,19 +19,19 @@
 import io.trino.spi.PageBuilder;
 import io.trino.spi.block.Block;
 import io.trino.spi.block.BlockBuilder;
-import io.trino.spi.block.ColumnarRow;
 import io.trino.spi.type.Type;
 
 import java.util.List;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Verify.verify;
-import static io.trino.spi.block.ColumnarRow.toColumnarRow;
+import static io.trino.spi.block.RowBlock.getRowFieldsFromBlock;
 import static io.trino.spi.connector.ConnectorMergeSink.DELETE_OPERATION_NUMBER;
 import static io.trino.spi.connector.ConnectorMergeSink.INSERT_OPERATION_NUMBER;
 import static io.trino.spi.connector.ConnectorMergeSink.UPDATE_DELETE_OPERATION_NUMBER;
 import static io.trino.spi.connector.ConnectorMergeSink.UPDATE_INSERT_OPERATION_NUMBER;
 import static io.trino.spi.connector.ConnectorMergeSink.UPDATE_OPERATION_NUMBER;
+import static io.trino.spi.type.IntegerType.INTEGER;
 import static io.trino.spi.type.TinyintType.TINYINT;
 import static java.util.Objects.requireNonNull;
 
@@ -100,8 +100,9 @@ public Page transformPage(Page inputPage)
         int originalPositionCount = inputPage.getPositionCount();
         checkArgument(originalPositionCount > 0, "originalPositionCount should be > 0, but is %s", originalPositionCount);
 
-        ColumnarRow mergeRow = toColumnarRow(inputPage.getBlock(mergeRowChannel));
-        Block operationChannelBlock = mergeRow.getField(mergeRow.getFieldCount() - 2);
+        Block mergeRow = inputPage.getBlock(mergeRowChannel);
+        List<Block> fields = getRowFieldsFromBlock(mergeRow);
+        Block operationChannelBlock = fields.get(fields.size() - 2);
 
         int updatePositions = 0;
         int insertPositions = 0;
@@ -123,6 +124,7 @@ public Page transformPage(Page inputPage)
         List<Type> pageTypes = ImmutableList.<Type>builder()
                 .addAll(dataColumnTypes)
                 .add(TINYINT)
+                .add(INTEGER)
                 .add(rowIdType)
                 .add(TINYINT)
                 .build();
@@ -137,7 +139,7 @@ public Page transformPage(Page inputPage)
                 }
                 // Insert and update because both create an insert row
                 if (operation == INSERT_OPERATION_NUMBER || operation == UPDATE_OPERATION_NUMBER) {
-                    addInsertRow(pageBuilder, mergeRow, position, operation != INSERT_OPERATION_NUMBER);
+                    addInsertRow(pageBuilder, fields, position, operation != INSERT_OPERATION_NUMBER);
                 }
             }
         }
@@ -170,33 +172,39 @@ private void addDeleteRow(PageBuilder pageBuilder, Page originalPage, int positi
         // Add the operation column == deleted
         TINYINT.writeLong(pageBuilder.getBlockBuilder(dataColumnChannels.size()), causedByUpdate ? UPDATE_DELETE_OPERATION_NUMBER : DELETE_OPERATION_NUMBER);
 
+        // Add the dummy case number, delete and insert won't use it, use -1 to mark it shouldn't be used
+        INTEGER.writeLong(pageBuilder.getBlockBuilder(dataColumnChannels.size() + 1), -1);
+
         // Copy row ID column
-        rowIdType.appendTo(originalPage.getBlock(rowIdChannel), position, pageBuilder.getBlockBuilder(dataColumnChannels.size() + 1));
+        rowIdType.appendTo(originalPage.getBlock(rowIdChannel), position, pageBuilder.getBlockBuilder(dataColumnChannels.size() + 2));
 
         // Write 0, meaning this row is not an insert derived from an update
-        TINYINT.writeLong(pageBuilder.getBlockBuilder(dataColumnChannels.size() + 2), 0);
+        TINYINT.writeLong(pageBuilder.getBlockBuilder(dataColumnChannels.size() + 3), 0);
 
         pageBuilder.declarePosition();
     }
 
-    private void addInsertRow(PageBuilder pageBuilder, ColumnarRow mergeCaseBlock, int position, boolean causedByUpdate)
+    private void addInsertRow(PageBuilder pageBuilder, List<Block> fields, int position, boolean causedByUpdate)
     {
         // Copy the values from the merge block
         for (int targetChannel : dataColumnChannels) {
             Type columnType = dataColumnTypes.get(targetChannel);
             BlockBuilder targetBlock = pageBuilder.getBlockBuilder(targetChannel);
             // The value comes from that column of the page
-            columnType.appendTo(mergeCaseBlock.getField(targetChannel), position, targetBlock);
+            columnType.appendTo(fields.get(targetChannel), position, targetBlock);
         }
 
         // Add the operation column == insert
         TINYINT.writeLong(pageBuilder.getBlockBuilder(dataColumnChannels.size()), causedByUpdate ? UPDATE_INSERT_OPERATION_NUMBER : INSERT_OPERATION_NUMBER);
 
+        // Add the dummy case number, delete and insert won't use it
+        INTEGER.writeLong(pageBuilder.getBlockBuilder(dataColumnChannels.size() + 1), 0);
+
         // Add null row ID column
-        pageBuilder.getBlockBuilder(dataColumnChannels.size() + 1).appendNull();
+        pageBuilder.getBlockBuilder(dataColumnChannels.size() + 2).appendNull();
 
         // Write 1 if this row is an insert derived from an update, 0 otherwise
-        TINYINT.writeLong(pageBuilder.getBlockBuilder(dataColumnChannels.size() + 2), causedByUpdate ? 1 : 0);
+        TINYINT.writeLong(pageBuilder.getBlockBuilder(dataColumnChannels.size() + 3), causedByUpdate ? 1 : 0);
 
         pageBuilder.declarePosition();
     }

core/trino-main/src/main/java/io/trino/sql/planner/QueryPlanner.java

@@ -556,6 +556,8 @@ public PlanNode plan(Delete node)
         List<Symbol> columnSymbols = columnSymbolsBuilder.build();
         Symbol operationSymbol = symbolAllocator.newSymbol("operation", TINYINT);
         assignmentsBuilder.put(operationSymbol, new GenericLiteral("TINYINT", String.valueOf(DELETE_OPERATION_NUMBER)));
+        Symbol caseNumberSymbol = symbolAllocator.newSymbol("case_number", INTEGER);
+        assignmentsBuilder.put(caseNumberSymbol, new GenericLiteral("INTEGER", String.valueOf(0)));
         Symbol projectedRowIdSymbol = symbolAllocator.newSymbol(rowIdSymbol.getName(), rowIdType);
         assignmentsBuilder.put(projectedRowIdSymbol, rowIdSymbol.toSymbolReference());
         assignmentsBuilder.put(symbolAllocator.newSymbol("insert_from_update", TINYINT), new GenericLiteral("TINYINT", "0"));
@@ -929,11 +931,13 @@ private MergeWriterNode createMergePipeline(Table table, RelationPlan relationPl
         }
 
         Symbol operationSymbol = symbolAllocator.newSymbol("operation", TINYINT);
+        Symbol caseNumberSymbol = symbolAllocator.newSymbol("case_number", INTEGER);
         Symbol insertFromUpdateSymbol = symbolAllocator.newSymbol("insert_from_update", TINYINT);
 
         List<Symbol> projectedSymbols = ImmutableList.<Symbol>builder()
                 .addAll(columnSymbols)
                 .add(operationSymbol)
+                .add(caseNumberSymbol)
                 .add(rowIdSymbol)
                 .add(insertFromUpdateSymbol)
                 .build();