Fix table_changes incorrect results when querying cow tables

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/ForIcebergCopyOnWriteTableChangesProcessor.java

@@ -0,0 +1,29 @@
+/*
+ * 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.iceberg;
+
+import com.google.inject.BindingAnnotation;
+
+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})
+@BindingAnnotation
+public @interface ForIcebergCopyOnWriteTableChangesProcessor {}

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergConfig.java

@@ -95,6 +95,7 @@ public class IcebergConfig
     private Set<String> queryPartitionFilterRequiredSchemas = ImmutableSet.of();
     private int splitManagerThreads = Math.min(Runtime.getRuntime().availableProcessors() * 2, 32);
     private int planningThreads = Math.min(Runtime.getRuntime().availableProcessors(), 16);
+    private int tableChangesProcessorThreads = Math.min(Runtime.getRuntime().availableProcessors(), 16);
     private int fileDeleteThreads = Runtime.getRuntime().availableProcessors() * 2;
     private List<String> allowedExtraProperties = ImmutableList.of();
     private boolean incrementalRefreshEnabled = true;
@@ -566,6 +567,20 @@ public IcebergConfig setPlanningThreads(String planningThreads)
         return this;
     }
 
+    @Min(0)
+    public int getTableChangesProcessorThreads()
+    {
+        return tableChangesProcessorThreads;
+    }
+
+    @Config("iceberg.table-changes-processor-threads")
+    @ConfigDescription("Number of threads to use for table changes processing for Copy-On-Write tables")
+    public IcebergConfig setTableChangesProcessorThreads(String tableChangesProcessorThreads)
+    {
+        this.tableChangesProcessorThreads = ThreadCount.valueOf(tableChangesProcessorThreads).getThreadCount();
+        return this;
+    }
+
     @Min(0)
     public int getFileDeleteThreads()
     {

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergExecutorModule.java

@@ -40,6 +40,7 @@ public void configure(Binder binder)
         closingBinder(binder).registerExecutor(Key.get(ListeningExecutorService.class, ForIcebergSplitSource.class));
         closingBinder(binder).registerExecutor(Key.get(ExecutorService.class, ForIcebergSplitManager.class));
         closingBinder(binder).registerExecutor(Key.get(ExecutorService.class, ForIcebergPlanning.class));
+        closingBinder(binder).registerExecutor(Key.get(ExecutorService.class, ForIcebergCopyOnWriteTableChangesProcessor.class));
     }
 
     @Singleton
@@ -96,4 +97,17 @@ public ExecutorService createFileDeleteExecutor(CatalogName catalogName, Iceberg
                 config.getFileDeleteThreads(),
                 daemonThreadsNamed("iceberg-file-delete-" + catalogName + "-%s"));
     }
+
+    @Provides
+    @Singleton
+    @ForIcebergCopyOnWriteTableChangesProcessor
+    public ExecutorService createIcebergCopyOnWriteTableChangesExecutor(CatalogName catalogName, IcebergConfig config)
+    {
+        if (config.getPlanningThreads() == 0) {
+            return newDirectExecutorService();
+        }
+        return newFixedThreadPool(
+                config.getTableChangesProcessorThreads(),
+                daemonThreadsNamed("iceberg-copy-on-write-table-changes-" + catalogName + "-%s"));
+    }
 }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergUtil.java

@@ -66,6 +66,7 @@
 import org.apache.iceberg.ManifestReader;
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowLevelOperationMode;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.SnapshotUpdate;
@@ -187,11 +188,13 @@
 import static java.util.Comparator.comparing;
 import static java.util.Locale.ENGLISH;
 import static java.util.Objects.requireNonNull;
+import static org.apache.iceberg.RowLevelOperationMode.MERGE_ON_READ;
 import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION;
 import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES;
 import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
 import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
 import static org.apache.iceberg.TableProperties.FORMAT_VERSION;
+import static org.apache.iceberg.TableProperties.MERGE_MODE;
 import static org.apache.iceberg.TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED;
 import static org.apache.iceberg.TableProperties.METADATA_PREVIOUS_VERSIONS_MAX;
 import static org.apache.iceberg.TableProperties.OBJECT_STORE_ENABLED;
@@ -1271,4 +1274,10 @@ public static ManifestReader<? extends ContentFile<?>> readerForManifest(Manifes
             case DELETES -> ManifestFiles.readDeleteManifest(manifest, fileIO, specsById);
         };
     }
+
+    public static RowLevelOperationMode rowLevelOperationMode(Table table)
+    {
+        // Trino uses MoR by default unlike Spark
+        return RowLevelOperationMode.fromName(table.properties().getOrDefault(MERGE_MODE, MERGE_ON_READ.modeName()));
+    }
 }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/functions/tablechanges/AbstractTableChangesSplitSource.java

@@ -0,0 +1,112 @@
+/*
+ * 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.iceberg.functions.tablechanges;
+
+import io.trino.plugin.iceberg.IcebergFileFormat;
+import io.trino.plugin.iceberg.PartitionData;
+import io.trino.spi.SplitWeight;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.ConnectorSplit;
+import io.trino.spi.connector.ConnectorSplitSource;
+import io.trino.spi.type.DateTimeEncoding;
+import org.apache.iceberg.AddedRowsScanTask;
+import org.apache.iceberg.ChangelogScanTask;
+import org.apache.iceberg.DeletedDataFileScanTask;
+import org.apache.iceberg.PartitionSpecParser;
+import org.apache.iceberg.SplittableScanTask;
+import org.apache.iceberg.Table;
+
+import java.util.Iterator;
+
+import static com.google.common.collect.Iterators.singletonIterator;
+import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
+import static io.trino.spi.type.TimeZoneKey.UTC_KEY;
+import static java.util.Objects.requireNonNull;
+
+public abstract class AbstractTableChangesSplitSource
+        implements ConnectorSplitSource
+{
+    private final Table icebergTable;
+
+    public AbstractTableChangesSplitSource(Table icebergTable)
+    {
+        this.icebergTable = requireNonNull(icebergTable, "icebergTable is null");
+    }
+
+    @SuppressWarnings("unchecked")
+    protected static Iterator<? extends ChangelogScanTask> splitIfPossible(ChangelogScanTask wholeFileScan, long targetSplitSize)
+    {
+        if (wholeFileScan instanceof AddedRowsScanTask) {
+            return ((SplittableScanTask<AddedRowsScanTask>) wholeFileScan).split(targetSplitSize).iterator();
+        }
+
+        if (wholeFileScan instanceof DeletedDataFileScanTask) {
+            return ((SplittableScanTask<DeletedDataFileScanTask>) wholeFileScan).split(targetSplitSize).iterator();
+        }
+
+        return singletonIterator(wholeFileScan);
+    }
+
+    protected ConnectorSplit toIcebergSplit(ChangelogScanTask task)
+    {
+        // TODO: Support DeletedRowsScanTask (requires https://github.com/apache/iceberg/pull/6182)
+        if (task instanceof AddedRowsScanTask addedRowsScanTask) {
+            return toSplit(addedRowsScanTask);
+        }
+        else if (task instanceof DeletedDataFileScanTask deletedDataFileScanTask) {
+            return toSplit(deletedDataFileScanTask);
+        }
+        else {
+            throw new TrinoException(NOT_SUPPORTED, "ChangelogScanTask type is not supported:" + task);
+        }
+    }
+
+    protected TableChangesInternalSplit toSplit(AddedRowsScanTask task)
+    {
+        return new TableChangesInternalSplit(
+                TableChangesInternalSplit.ChangeType.ADDED_FILE,
+                task.commitSnapshotId(),
+                DateTimeEncoding.packDateTimeWithZone(icebergTable.snapshot(task.commitSnapshotId()).timestampMillis(), UTC_KEY),
+                task.changeOrdinal(),
+                task.file().location(),
+                task.start(),
+                task.length(),
+                task.file().fileSizeInBytes(),
+                task.file().recordCount(),
+                IcebergFileFormat.fromIceberg(task.file().format()),
+                PartitionSpecParser.toJson(task.spec()),
+                PartitionData.toJson(task.file().partition()),
+                SplitWeight.standard(),
+                icebergTable.io().properties());
+    }
+
+    protected TableChangesInternalSplit toSplit(DeletedDataFileScanTask task)
+    {
+        return new TableChangesInternalSplit(
+                TableChangesInternalSplit.ChangeType.DELETED_FILE,
+                task.commitSnapshotId(),
+                DateTimeEncoding.packDateTimeWithZone(icebergTable.snapshot(task.commitSnapshotId()).timestampMillis(), UTC_KEY),
+                task.changeOrdinal(),
+                task.file().location(),
+                task.start(),
+                task.length(),
+                task.file().fileSizeInBytes(),
+                task.file().recordCount(),
+                IcebergFileFormat.fromIceberg(task.file().format()),
+                PartitionSpecParser.toJson(task.spec()),
+                PartitionData.toJson(task.file().partition()),
+                SplitWeight.standard(),
+                icebergTable.io().properties());
+    }
+}