Skip to content

Fixed handling of iceberg table created via flink iceberg sink in upsert mode - #98

Open
andrea-rockt wants to merge 1 commit into
dremio:masterfrom
andrea-rockt:feature/handle-equality-and-positional-deletes-flink-tables
Open

Fixed handling of iceberg table created via flink iceberg sink in upsert mode#98
andrea-rockt wants to merge 1 commit into
dremio:masterfrom
andrea-rockt:feature/handle-equality-and-positional-deletes-flink-tables

Conversation

@andrea-rockt

@andrea-rockt andrea-rockt commented Jun 19, 2025

Copy link
Copy Markdown

Issue summary

Dremio appears to not correctly handle tables written with the flink iceberg sink in upsert mode.

The flink iceberg sink utilizes a mix of positional deletes (for changes happening inside the same checkpoint interval to the same key) and equality deletes (only for the first change to a key during a checkpoint interval, to not scan the full table to generate positional delete )

Spark is able to read the table correctly

flink iceberg sink and spark iceberg reader are part of the official iceberg codebase so i expect coherency between the two implementations

The sample table used in this PR is provided in this pull request content

/sabot/kernel/src/test/resources/iceberg/v2/upsert_table_with_equality_deletes/

Libraries version

<dependency>
    <groupId>org.apache.iceberg</groupId>
    <artifactId>iceberg-flink-runtime-1.17</artifactId>
    <version>1.4.3</version>
</dependency>
<dependency>
    <groupId>com.ververica</groupId>
    <artifactId>flink-connector-postgres-cdc</artifactId>
    <version>2.4.2</version>
</dependency>

The problem happens with all the releases of flink iceberg runtime

      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.5.5
      /_/

org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.2

Writer application

package it.agilelab.banking.poc.cdc;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class SqlApi {


  public static void main(String[] args) throws Exception {

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    env.enableCheckpointing(120000, CheckpointingMode.EXACTLY_ONCE);

    StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);


    tableEnvironment.executeSql("CREATE CATALOG my_catalog WITH (\n" +
                                "'type'='iceberg'," +
                                "'catalog-type'='hadoop'," +
                                "'warehouse'='file:///tmp/iceberg-test-tables'," +
                                "'property-version'='1'\n" +
        ");\n").await();


    tableEnvironment.executeSql("-- register a PostgreSQL table 'shipments' in Flink SQL\n" +
        "CREATE TABLE hero(\n" +
        "  hero_id INT,\n" +
        "  name STRING,\n" +
        "  real_name STRING,\n" +
        "  universe STRING,\n" +
        "  alignment STRING,\n" +
        "  first_appearance DATE, \n" +
        "  PRIMARY KEY (hero_id) NOT ENFORCED\n" +
        ") WITH (\n" +
        "  'connector' = 'postgres-cdc',\n" +
        "  'hostname' = 'localhost',\n" +
        "  'port' = '5432',\n" +
        "  'username' = 'DATAPRODUCTREPLICATION',\n" +
        "  'password' = 'dataproductreplication',\n" +
        "  'database-name' = 'DATAPRODUCT',\n" +
        "  'schema-name' = 'domain_dataproduct_v0',\n" +
        "  'table-name' = 'hero',\n" +
        "  'decoding.plugin.name' = 'pgoutput',\n" +
        "  'changelog-mode' = 'upsert',\n" +
        "  'slot.name' = 'flink_debezium',\n" +
        "  'debezium.publication.name' = 'flink_cdc_publication',\n" +
        "  'debezium.slot.drop.on.stop' = 'false',\n" +
        "  'debezium.publication.autocreate.mode' = 'disabled'\n" +
        ")");




    tableEnvironment.executeSql("CREATE TABLE if not exists my_catalog.v2.upsert_table_with_equality_deletes(\n" +
        "  hero_id INT,\n" +
        "  name STRING,\n" +
        "  real_name STRING,\n" +
        "  universe STRING,\n" +
        "  alignment STRING,\n" +
        "  first_appearance DATE,\n" +
        "  PRIMARY KEY (hero_id) NOT ENFORCED\n" +
        ") PARTITIONED BY (hero_id) " +
                                "WITH (\n" +
       "  'write.upsert.enabled' = 'true',\n" +
        "  'format-version' = '2'\n" +
        ")").await();


    tableEnvironment.executeSql("INSERT INTO my_catalog.v2.upsert_table_with_equality_deletes\n" +
        "SELECT * FROM hero").await();
    env.execute();
  }
}

Table construction timeline

t1/flink - Initial load of table
t2/flink - Performs checkpoint 1
t3/spark - query snapshot

sql("select * from my_catalog.v2.upsert_table_with_equality_deletes 
VERSION AS OF 1589939248529822747 order by hero_id").show()

+-------+--------+-----------+--------+---------+----------------+
|hero_id|    name|  real_name|universe|alignment|first_appearance|
+-------+--------+-----------+--------+---------+----------------+
|      1|  Batman|Bruce Wayne|      DC|     good|      1939-05-01|
|      2|Superman| Clark Kent|      DC|     good|      1938-06-01|
|      3|Iron Man| Tony Stark|  Marvel|     good|      1963-03-01|
|      4|Deadpool|Wade Wilson|  Marvel|  neutral|      1991-02-01|
+-------+--------+-----------+--------+---------+----------------+

t4/database - deadpool updated 3 times
t5/flink - replicate changes to iceberg table
t6/flink - Performs checkpoint 2
t7/spark - Query snapshot of checkpoint 2

sql("select * from my_catalog.v2.upsert_table_with_equality_deletes 
VERSION AS OF 8118271012486886609 order by hero_id").show()
+-------+--------+--------------------+--------+---------+----------------+
|hero_id|    name|           real_name|universe|alignment|first_appearance|
+-------+--------+--------------------+--------+---------+----------------+
|      1|  Batman|         Bruce Wayne|      DC|     good|      1939-05-01|
|      2|Superman|          Clark Kent|      DC|     good|      1938-06-01|
|      3|Iron Man|          Tony Stark|  Marvel|     good|      1963-03-01|
|      4|Deadpool|Wade Wilson Update 3|  Marvel|  neutral|      1991-02-01|
+-------+--------+--------------------+--------+---------+----------------+

t8/database - Removed batman
t9/database - Inserted Aquaman
t10/database - Updated tony stark
t11/flink - Replicate changes
t12/flink - Perform checkpoint 3
t13/spark - Query snapshot after checkpoint

scala> sql("select * from my_catalog.v2.upsert_table_with_equality_deletes 
VERSION AS OF 5502512143601116985 order by hero_id").show()
+-------+--------+--------------------+--------+---------+----------------+
|hero_id|    name|           real_name|universe|alignment|first_appearance|
+-------+--------+--------------------+--------+---------+----------------+
|      2|Superman|          Clark Kent|      DC|     good|      1938-06-01|
|      3|Iron Man| Tony Stark Update 4|  Marvel|     good|      1963-03-01|
|      4|Deadpool|Wade Wilson Update 3|  Marvel|  neutral|      1991-02-01|
|      5| Aquaman|             Aquaman|      DC|     good|      2025-06-10|
+-------+--------+--------------------+--------+---------+----------------+

This state matches the database state

Accessing the table via dremio

Dremio versions tried (docker deployment)

  • dremio/dremio-oss:25.1.0
  • dremio/dremio-oss:25.1.1
  • dremio/dremio-oss:25.2.0
  • dremio/dremio-oss:26.0.0
SELECT * FROM "iceberg-test-tables".v2.upsert_table_with_equality_deletes

Physical plan

00-00    Screen : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {120.9 rows, 524.9036 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2188
00-01      Project(Fragment=[$0], Records=[$1], Path=[$2], Metadata=[$3], Partition=[$4], FileSize=[$5], IcebergMetadata=[$6], fileschema=[$7], PartitionData=[$8], OperationType=[$9], PartitionValue=[$10], RejectedRecords=[$11], ReferencedDataFiles=[$12]) : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {120.0 rows, 524.0036 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2187
00-02        WriterCommitter(final=[/opt/dremio/data/dist/results/17ae8fe9-df5f-1a7a-5ea9-adc87a794e00]) : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {111.0 rows, 524.00243 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2186
00-03          Writer : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {102.0 rows, 515.00243 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2185
00-04            Project(hero_id=[$0], name=[$1], real_name=[$2], universe=[$3], alignment=[$4], first_appearance=[$5]) : rowType = RecordType(INTEGER hero_id, VARCHAR(65536) name, VARCHAR(65536) real_name, VARCHAR(65536) universe, VARCHAR(65536) alignment, DATE first_appearance): rowcount = 9.0, cumulative cost = {93.0 rows, 506.00243 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2184
00-05              Project(hero_id=[$0], name=[$1], real_name=[$2], universe=[$3], alignment=[$4], first_appearance=[$5]) : rowType = RecordType(INTEGER hero_id, VARCHAR(65536) name, VARCHAR(65536) real_name, VARCHAR(65536) universe, VARCHAR(65536) alignment, DATE first_appearance): rowcount = 9.0, cumulative cost = {84.0 rows, 506.00189 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2183
00-06                TableFunction(columns=[`hero_id`, `name`, `real_name`, `universe`, `alignment`, `first_appearance`], Table Function Type=[DATA_FILE_SCAN], table=[ccc.tmp."iceberg-test-tables".v2.upsert_table_with_equality_deletes]) : rowType = RecordType(INTEGER hero_id, VARCHAR(65536) name, VARCHAR(65536) real_name, VARCHAR(65536) universe, VARCHAR(65536) alignment, DATE first_appearance): rowcount = 9.0, cumulative cost = {75.0 rows, 506.00135 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2182
00-07                  IcebergSplitGen(columns=[`splitsIdentity`, `splits`, `colIds`, `deleteFiles`], Table Function Type=[ICEBERG_SPLIT_GEN]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) ARRAY deleteFiles): rowcount = 7.0, cumulative cost = {66.0 rows, 497.00135 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2181
00-08                    IcebergDeleteFileAgg(columns=[`datafilePath`, `fileSize`, `partitionInfo`, `colIds`, `deleteFiles`, `partitionSpecId`], Table Function Type=[ICEBERG_DELETE_FILE_AGG]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) ARRAY deleteFiles, INTEGER partitionSpecId): rowcount = 7.0, cumulative cost = {59.0 rows, 490.00135 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2180
00-09                      Project(datafilePath=[$0], fileSize=[$1], partitionInfo=[$5], colIds=[$6], deleteFile=[$8], partitionSpecId=[$3]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, INTEGER partitionSpecId): rowcount = 9.0, cumulative cost = {52.0 rows, 483.00135 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2179
00-10                        HashJoin(condition=[AND(=($3, $12), =($4, $13))], joinType=[left], extraCondition=[<=($2, $11)]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, VARCHAR(65536) datafilePath0, BIGINT fileSize0, BIGINT sequenceNumber0, INTEGER partitionSpecId0, VARBINARY(65536) partitionKey0, VARBINARY(65536) partitionInfo0, VARBINARY(65536) colIds0, VARCHAR(65536) fileContent0): rowcount = 9.0, cumulative cost = {43.0 rows, 483.00081 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2168
00-12                          IcebergManifestScan(columns=[`datafilePath`, `fileSize`, `sequenceNumber`, `partitionSpecId`, `partitionKey`, `partitionInfo`, `colIds`, `fileContent`], manifestContent=[DATA]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent): rowcount = 7.0, cumulative cost = {8.0 rows, 8.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 2164
00-14                            IcebergManifestList(table=[ccc.tmp."iceberg-test-tables".v2.upsert_table_with_equality_deletes], snapshot=[8715012473742984723], columns=[`splitsIdentity`, `splits`, `colIds`], splits=[1], metadataFileLocation=[/tmp/iceberg-test-tables/v2/upsert_table_with_equality_deletes/metadata/v4.metadata.json], manifestContent=[DATA]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds): rowcount = 1.0, cumulative cost = {1.0 rows, 1.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 2163
00-11                          Project(deleteFile=[$0], datafilePath0=[$1], fileSize0=[$2], sequenceNumber0=[$3], partitionSpecId0=[$4], partitionKey0=[$5], partitionInfo0=[$6], colIds0=[$7], fileContent0=[$8]) : rowType = RecordType(RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, VARCHAR(65536) datafilePath0, BIGINT fileSize0, BIGINT sequenceNumber0, INTEGER partitionSpecId0, VARBINARY(65536) partitionKey0, VARBINARY(65536) partitionInfo0, VARBINARY(65536) colIds0, VARCHAR(65536) fileContent0): rowcount = 9.0, cumulative cost = {19.0 rows, 10.00081 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 2167
00-13                            IcebergManifestScan(columns=[`deleteFile`, `datafilePath`, `fileSize`, `sequenceNumber`, `partitionSpecId`, `partitionKey`, `partitionInfo`, `colIds`, `fileContent`], manifestContent=[DELETES]) : rowType = RecordType(RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent): rowcount = 9.0, cumulative cost = {10.0 rows, 10.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 2166
00-15                              IcebergManifestList(table=[ccc.tmp."iceberg-test-tables".v2.upsert_table_with_equality_deletes], snapshot=[8715012473742984723], columns=[`splitsIdentity`, `splits`, `colIds`], splits=[1], metadataFileLocation=[/tmp/iceberg-test-tables/v2/upsert_table_with_equality_deletes/metadata/v4.metadata.json], manifestContent=[DELETES]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds): rowcount = 1.0, cumulative cost = {1.0 rows, 1.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 2165

TABLE IS EMPTY, equality delete files are applied to all data files in snapshots

+-------+--------+--------------------+--------+---------+----------------+
|hero_id|    name|           real_name|universe|alignment|first_appearance|
+-------+--------+--------------------+--------+---------+----------------+
+-------+--------+--------------------+--------+---------+----------------+

Steps taken to produce this patch

Extracondition of hash join between data and delete files appears to be wrong.

Iceberg documentation says that delete files should be applied to datafiles where sequenceNumber is strictly less than dremio appears to use less than equal (it is a bit ambiguous in wording)

00-10                        HashJoin(condition=[AND(=($3, $12), =($4, $13))], joinType=[left], extraCondition=[<=($2, $11)]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, VARCHAR(65536) datafilePath0, BIGINT fileSize0, BIGINT sequenceNumber0, INTEGER partitionSpecId0, VARBINARY(65536) partitionKey0, VARBINARY(65536) partitionInfo0, VARBINARY(65536) colIds0, VARCHAR(65536) fileContent0): rowcount = 9.0, cumulative cost = {43.0 rows, 483.00081 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 2168

Dremio Patched to use less than operator (instead of less than equal) in extra hash join condition when associating datafiles to the respective equality and positional delete files

Screen : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {120.9 rows, 524.9036 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1314
00-01      Project(Fragment=[$0], Records=[$1], Path=[$2], Metadata=[$3], Partition=[$4], FileSize=[$5], IcebergMetadata=[$6], fileschema=[$7], PartitionData=[$8], OperationType=[$9], PartitionValue=[$10], RejectedRecords=[$11], ReferencedDataFiles=[$12]) : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {120.0 rows, 524.0036 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1313
00-02        WriterCommitter(final=[/Users/andreafonti/projects/github.com/dremio/dremio-oss/distribution/server/target/dremio-community-26.0.0-202504290223270716-afdd6663/dremio-community-26.0.0-202504290223270716-afdd6663/data/pdfs/results/17ae8e88-2e3e-ae35-97a1-217905b1c800]) : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {111.0 rows, 524.00243 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1312
00-03          Writer : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {102.0 rows, 515.00243 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1311
00-04            Project(hero_id=[$0], name=[$1], real_name=[$2], universe=[$3], alignment=[$4], first_appearance=[$5]) : rowType = RecordType(INTEGER hero_id, VARCHAR(65536) name, VARCHAR(65536) real_name, VARCHAR(65536) universe, VARCHAR(65536) alignment, DATE first_appearance): rowcount = 9.0, cumulative cost = {93.0 rows, 506.00243 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1310
00-05              Project(hero_id=[$0], name=[$1], real_name=[$2], universe=[$3], alignment=[$4], first_appearance=[$5]) : rowType = RecordType(INTEGER hero_id, VARCHAR(65536) name, VARCHAR(65536) real_name, VARCHAR(65536) universe, VARCHAR(65536) alignment, DATE first_appearance): rowcount = 9.0, cumulative cost = {84.0 rows, 506.00189 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1309
00-06                TableFunction(columns=[`hero_id`, `name`, `real_name`, `universe`, `alignment`, `first_appearance`], Table Function Type=[DATA_FILE_SCAN], table=["iceberg-test-tables".v2.upsert_table_with_equality_deletes]) : rowType = RecordType(INTEGER hero_id, VARCHAR(65536) name, VARCHAR(65536) real_name, VARCHAR(65536) universe, VARCHAR(65536) alignment, DATE first_appearance): rowcount = 9.0, cumulative cost = {75.0 rows, 506.00135 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1308
00-07                  IcebergSplitGen(columns=[`splitsIdentity`, `splits`, `colIds`, `deleteFiles`], Table Function Type=[ICEBERG_SPLIT_GEN]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) ARRAY deleteFiles): rowcount = 7.0, cumulative cost = {66.0 rows, 497.00135 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1307
00-08                    IcebergDeleteFileAgg(columns=[`datafilePath`, `fileSize`, `partitionInfo`, `colIds`, `deleteFiles`, `partitionSpecId`], Table Function Type=[ICEBERG_DELETE_FILE_AGG]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) ARRAY deleteFiles, INTEGER partitionSpecId): rowcount = 7.0, cumulative cost = {59.0 rows, 490.00135 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1306
00-09                      Project(datafilePath=[$0], fileSize=[$1], partitionInfo=[$5], colIds=[$6], deleteFile=[$8], partitionSpecId=[$3]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, INTEGER partitionSpecId): rowcount = 9.0, cumulative cost = {52.0 rows, 483.00135 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1305
00-10                        HashJoin(condition=[AND(=($3, $12), =($4, $13))], joinType=[left], extraCondition=[<($2, $11)]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, VARCHAR(65536) datafilePath0, BIGINT fileSize0, BIGINT sequenceNumber0, INTEGER partitionSpecId0, VARBINARY(65536) partitionKey0, VARBINARY(65536) partitionInfo0, VARBINARY(65536) colIds0, VARCHAR(65536) fileContent0): rowcount = 9.0, cumulative cost = {43.0 rows, 483.00081 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1294
00-12                          IcebergManifestScan(columns=[`datafilePath`, `fileSize`, `sequenceNumber`, `partitionSpecId`, `partitionKey`, `partitionInfo`, `colIds`, `fileContent`], manifestContent=[DATA]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent): rowcount = 7.0, cumulative cost = {8.0 rows, 8.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 1290
00-14                            IcebergManifestList(table=["iceberg-test-tables".v2.upsert_table_with_equality_deletes], snapshot=[8715012473742984723], columns=[`splitsIdentity`, `splits`, `colIds`], splits=[1], metadataFileLocation=[/tmp/iceberg-test-tables/v2/upsert_table_with_equality_deletes/metadata/v4.metadata.json], manifestContent=[DATA]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds): rowcount = 1.0, cumulative cost = {1.0 rows, 1.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 1289
00-11                          Project(deleteFile=[$0], datafilePath0=[$1], fileSize0=[$2], sequenceNumber0=[$3], partitionSpecId0=[$4], partitionKey0=[$5], partitionInfo0=[$6], colIds0=[$7], fileContent0=[$8]) : rowType = RecordType(RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, VARCHAR(65536) datafilePath0, BIGINT fileSize0, BIGINT sequenceNumber0, INTEGER partitionSpecId0, VARBINARY(65536) partitionKey0, VARBINARY(65536) partitionInfo0, VARBINARY(65536) colIds0, VARCHAR(65536) fileContent0): rowcount = 9.0, cumulative cost = {19.0 rows, 10.00081 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 1293
00-13                            IcebergManifestScan(columns=[`deleteFile`, `datafilePath`, `fileSize`, `sequenceNumber`, `partitionSpecId`, `partitionKey`, `partitionInfo`, `colIds`, `fileContent`], manifestContent=[DELETES]) : rowType = RecordType(RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent): rowcount = 9.0, cumulative cost = {10.0 rows, 10.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 1292
00-15                              IcebergManifestList(table=["iceberg-test-tables".v2.upsert_table_with_equality_deletes], snapshot=[8715012473742984723], columns=[`splitsIdentity`, `splits`, `colIds`], splits=[1], metadataFileLocation=[/tmp/iceberg-test-tables/v2/upsert_table_with_equality_deletes/metadata/v4.metadata.json], manifestContent=[DELETES]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds): rowcount = 1.0, cumulative cost = {1.0 rows, 1.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 1291

00-10                        HashJoin(condition=[AND(=($3, $12), =($4, $13))], joinType=[left], extraCondition=[<($2, $11)]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, VARCHAR(65536) datafilePath0, BIGINT fileSize0, BIGINT sequenceNumber0, INTEGER partitionSpecId0, VARBINARY(65536) partitionKey0, VARBINARY(65536) partitionInfo0, VARBINARY(65536) colIds0, VARCHAR(65536) fileContent0): rowcount = 9.0, cumulative cost = {43.0 rows, 483.00081 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1294

+-------+--------+--------------------+--------+---------+----------------+
|hero_id|    name|           real_name|universe|alignment|first_appearance|
+-------+--------+--------------------+--------+---------+----------------+
|      2|Superman|          Clark Kent|      DC|     good|      1938-06-01|
|      3|Iron Man| Tony Stark Update 4|  Marvel|     good|      1963-03-01|
|      4|Deadpool|Wade Wilson Update 1|  Marvel|  neutral|      1991-02-01|
|      4|Deadpool|Wade Wilson Update 2|  Marvel|  neutral|      1991-02-01|
|      4|Deadpool|Wade Wilson Update 3|  Marvel|  neutral|      1991-02-01|
|      5| Aquaman|             Aquaman|      DC|     good|      2025-06-10|
+-------+--------+--------------------+--------+---------+----------------+

Equality delete fils are now applied as expected but positional delete files get skipped

Dremio patched to use less than as hash join predicate on equality deletes and less_than_equal on positional deletes

00-00    Screen : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {129.9 rows, 614.90513 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1362
00-01      Project(Fragment=[$0], Records=[$1], Path=[$2], Metadata=[$3], Partition=[$4], FileSize=[$5], IcebergMetadata=[$6], fileschema=[$7], PartitionData=[$8], OperationType=[$9], PartitionValue=[$10], RejectedRecords=[$11], ReferencedDataFiles=[$12]) : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {129.0 rows, 614.00513 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1361
00-02        WriterCommitter(final=[/Users/andreafonti/projects/github.com/dremio/dremio-oss/distribution/server/target/dremio-community-26.0.0-202504290223270716-afdd6663/dremio-community-26.0.0-202504290223270716-afdd6663/data/pdfs/results/17ad17b4-9606-55ad-e7e8-ea712d06b700]) : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {120.0 rows, 614.00396 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1360
00-03          Writer : rowType = RecordType(VARCHAR(65536) Fragment, BIGINT Records, VARCHAR(65536) Path, VARBINARY(65536) Metadata, INTEGER Partition, BIGINT FileSize, VARBINARY(65536) IcebergMetadata, VARBINARY(65536) fileschema, VARBINARY(65536) ARRAY PartitionData, INTEGER OperationType, VARCHAR(65536) PartitionValue, BIGINT RejectedRecords, VARBINARY(65536) ReferencedDataFiles): rowcount = 9.0, cumulative cost = {111.0 rows, 605.00396 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 1359
00-04            Project(hero_id=[$0], name=[$1], real_name=[$2], universe=[$3], alignment=[$4], first_appearance=[$5]) : rowType = RecordType(INTEGER hero_id, VARCHAR(65536) name, VARCHAR(65536) real_name, VARCHAR(65536) universe, VARCHAR(65536) alignment, DATE first_appearance): rowcount = 9.0, cumulative cost = {102.0 rows, 596.00396 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 481
00-05              Project(hero_id=[$0], name=[$1], real_name=[$2], universe=[$3], alignment=[$4], first_appearance=[$5]) : rowType = RecordType(INTEGER hero_id, VARCHAR(65536) name, VARCHAR(65536) real_name, VARCHAR(65536) universe, VARCHAR(65536) alignment, DATE first_appearance): rowcount = 9.0, cumulative cost = {93.0 rows, 596.00342 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 480
00-06                TableFunction(columns=[`hero_id`, `name`, `real_name`, `universe`, `alignment`, `first_appearance`], Table Function Type=[DATA_FILE_SCAN], table=[ciao."iceberg-test-tables".v2.upsert_table_with_equality_deletes]) : rowType = RecordType(INTEGER hero_id, VARCHAR(65536) name, VARCHAR(65536) real_name, VARCHAR(65536) universe, VARCHAR(65536) alignment, DATE first_appearance): rowcount = 9.0, cumulative cost = {84.0 rows, 596.00288 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 479
00-07                  IcebergSplitGen(columns=[`splitsIdentity`, `splits`, `colIds`, `deleteFiles`], Table Function Type=[ICEBERG_SPLIT_GEN]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) ARRAY deleteFiles): rowcount = 7.0, cumulative cost = {75.0 rows, 587.00288 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 478
00-08                    IcebergDeleteFileAgg(columns=[`datafilePath`, `fileSize`, `partitionInfo`, `colIds`, `deleteFiles`, `partitionSpecId`], Table Function Type=[ICEBERG_DELETE_FILE_AGG]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) ARRAY deleteFiles, INTEGER partitionSpecId): rowcount = 7.0, cumulative cost = {68.0 rows, 580.00288 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 477
00-09                      Project(datafilePath=[$0], fileSize=[$1], partitionInfo=[$5], colIds=[$6], deleteFile=[$8], partitionSpecId=[$3]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, INTEGER partitionSpecId): rowcount = 9.0, cumulative cost = {61.0 rows, 573.00288 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 476
00-10                        Project(datafilePath=[$0], fileSize=[$1], sequenceNumber=[$2], partitionSpecId=[$3], partitionKey=[$4], partitionInfo=[$5], colIds=[$6], fileContent=[$7], deleteFile=[$8], datafilePath0=[$9], fileSize0=[$10], sequenceNumber0=[$11], partitionSpecId0=[$12], partitionKey0=[$13], partitionInfo0=[$14], colIds0=[$15], fileContent0=[$16]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, VARCHAR(65536) datafilePath0, BIGINT fileSize0, BIGINT sequenceNumber0, INTEGER partitionSpecId0, VARBINARY(65536) partitionKey0, VARBINARY(65536) partitionInfo0, VARBINARY(65536) colIds0, VARCHAR(65536) fileContent0): rowcount = 9.0, cumulative cost = {52.0 rows, 573.00234 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 475
00-11                          HashJoin(condition=[AND(=($3, $12), =($4, $13))], joinType=[left], extraCondition=[OR(AND($17, <($2, $11)), AND($18, <=($2, $11)))]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent, RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) $f0, VARCHAR(65536) $f1, BIGINT $f2, BIGINT $f3, INTEGER $f4, VARBINARY(65536) $f5, VARBINARY(65536) $f6, VARBINARY(65536) $f7, VARCHAR(65536) $f8, BOOLEAN $f9, BOOLEAN $f10): rowcount = 9.0, cumulative cost = {43.0 rows, 573.00081 cpu, 0.0 io, 0.0 network, 237.60000000000002 memory}, id = 463
00-13                            IcebergManifestScan(columns=[`datafilePath`, `fileSize`, `sequenceNumber`, `partitionSpecId`, `partitionKey`, `partitionInfo`, `colIds`, `fileContent`], manifestContent=[DATA]) : rowType = RecordType(VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent): rowcount = 7.0, cumulative cost = {8.0 rows, 8.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 459
00-15                              IcebergManifestList(table=[ciao."iceberg-test-tables".v2.upsert_table_with_equality_deletes], snapshot=[5502512143601116985], columns=[`splitsIdentity`, `splits`, `colIds`], splits=[1], metadataFileLocation=[/tmp/iceberg-test-tables/v2/upsert_table_with_equality_deletes/metadata/v4.metadata.json], manifestContent=[DATA]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds): rowcount = 1.0, cumulative cost = {1.0 rows, 1.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 458
00-12                            Project($f0=[$0], $f1=[$1], $f2=[$2], $f3=[$3], $f4=[$4], $f5=[$5], $f6=[$6], $f7=[$7], $f8=[$8], $f9=[=('EQUALITY_DELETES':VARCHAR(16), $8)], $f10=[=('POSITION_DELETES':VARCHAR(16), $8)]) : rowType = RecordType(RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) $f0, VARCHAR(65536) $f1, BIGINT $f2, BIGINT $f3, INTEGER $f4, VARBINARY(65536) $f5, VARBINARY(65536) $f6, VARBINARY(65536) $f7, VARCHAR(65536) $f8, BOOLEAN $f9, BOOLEAN $f10): rowcount = 9.0, cumulative cost = {19.0 rows, 82.00081 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 462
00-14                              IcebergManifestScan(columns=[`deleteFile`, `datafilePath`, `fileSize`, `sequenceNumber`, `partitionSpecId`, `partitionKey`, `partitionInfo`, `colIds`, `fileContent`], manifestContent=[DELETES]) : rowType = RecordType(RecordType(VARCHAR(65536) path, INTEGER fileContent, BIGINT recordCount, INTEGER ARRAY equalityIds) deleteFile, VARCHAR(65536) datafilePath, BIGINT fileSize, BIGINT sequenceNumber, INTEGER partitionSpecId, VARBINARY(65536) partitionKey, VARBINARY(65536) partitionInfo, VARBINARY(65536) colIds, VARCHAR(65536) fileContent): rowcount = 9.0, cumulative cost = {10.0 rows, 10.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 461
00-16                                IcebergManifestList(table=[ciao."iceberg-test-tables".v2.upsert_table_with_equality_deletes], snapshot=[5502512143601116985], columns=[`splitsIdentity`, `splits`, `colIds`], splits=[1], metadataFileLocation=[/tmp/iceberg-test-tables/v2/upsert_table_with_equality_deletes/metadata/v4.metadata.json], manifestContent=[DELETES]) : rowType = RecordType(RecordType(VARCHAR(65536) path, BIGINT offset, BIGINT length, BIGINT fileLength) splitsIdentity, VARBINARY(65536) splits, VARBINARY(65536) colIds): rowcount = 1.0, cumulative cost = {1.0 rows, 1.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 460

Dremio result matches the result of SPARK

+-------+--------+--------------------+--------+---------+----------------+
|hero_id|    name|           real_name|universe|alignment|first_appearance|
+-------+--------+--------------------+--------+---------+----------------+
|      2|Superman|          Clark Kent|      DC|     good|      1938-06-01|
|      3|Iron Man| Tony Stark Update 4|  Marvel|     good|      1963-03-01|
|      4|Deadpool|Wade Wilson Update 3|  Marvel|  neutral|      1991-02-01|
|      5| Aquaman|             Aquaman|      DC|     good|      2025-06-10|
+-------+--------+--------------------+--------+---------+----------------+

Additional notes

Once the hash join condition was changed to distinguish between positional and equality delete files the IcebergDeleteFileAggTableFunction broke, when copying data from the internal accumulator to the output vector the inner equalityIds list was populated with nulls instead of the correct values, unless the equalityIds length was the same for each entry (it is null for positional delete, no list at all)

I believe to have fixed the vectors handling and properly allocated memory for the accumulator vector but i don't have sufficient expertise to judge it

@Jordano-Dremio

Jordano-Dremio commented Jul 14, 2025 •
edited
Loading

Copy link
Copy Markdown

Hi @andrea-rockt. Thank you for reaching out and providing this extensive commit statement. Over the past 2-3 weeks our team has discussed this internally as we've coincidentally hit both of the bugs you state in this fix.

The sequence number issue is fixed now. The null equality Id's should be merged this week. I'll report back on whether we can have the equality id issue in 26.1.0 in prep for the next oss release.

Thanks again! We appreciate it.

@Jordano-Dremio

Jordano-Dremio commented Jul 17, 2025

Copy link
Copy Markdown

Hi @andrea-rockt , Updating once more to confirm that both of these bugs you've addressed will be part of 26.1.0 release. Thanks for the contribution!

Best,
Jordano

@andrea-rockt

andrea-rockt commented Jul 17, 2025

Copy link
Copy Markdown
Author

Thanks for the update @Jordano-Dremio much appreciated, thanks again for the support 💪

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@andrea-rockt @Jordano-Dremio