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

sabot/kernel/src/main/java/com/dremio/exec/store/iceberg/IcebergDeleteFileAggTableFunction.java

@@ -143,8 +143,9 @@ public VectorAccessible setup(VectorAccessible accessible) throws Exception {
     outputDeleteFiles = (ListVector) getVectorFromSchemaPath(outgoing, SystemSchemas.DELETE_FILES);
     outputSpecId = (IntVector) getVectorFromSchemaPath(outgoing, SystemSchemas.PARTITION_SPEC_ID);
 
-    currentDeleteFiles = ListVector.empty(SystemSchemas.DELETE_FILES, context.getAllocator());
+    currentDeleteFiles = (ListVector) outputDeleteFiles.getField().createVector(context.getAllocator());
     currentDeleteFilesWriter = currentDeleteFiles.getWriter();
+    currentDeleteFiles.allocateNew();
 
     state = State.START_NEW_AGG;
     return outgoing;
@@ -238,6 +239,8 @@ public void close() throws Exception {
   }
 
   private void copyInputToCurrent(int inputIndex) {
+    currentDeleteFiles.setValueCount(0);
+    currentDeleteFiles.reset();
     currentDataFilePath = inputDataFilePath.get(inputIndex);
     currentFileSize = inputFileSize.get(inputIndex);
     currentPartitionInfo = inputPartitionInfo.get(inputIndex);
@@ -265,6 +268,8 @@ private void copyCurrentToOutput(int outputIndex) {
     outputSpecId.setSafe(outputIndex, currentSpecID);
 
     currentDeleteFilesWriter.endList();
+
+    currentDeleteFiles.setValueCount(1);
     outputDeleteFiles.copyFromSafe(0, outputIndex, currentDeleteFiles);
 
     outgoing.setAllCount(outputIndex + 1);

sabot/kernel/src/main/java/com/dremio/exec/store/iceberg/IcebergScanPlanBuilder.java

@@ -673,6 +673,7 @@ public RelNode buildDataAndDeleteFileJoinAndAggregate(RelNode data, RelNode dele
     RelDataTypeField buildSeqNumField = getField(build, SystemSchemas.SEQUENCE_NUMBER);
     RelDataTypeField buildSpecIdField = getField(build, SystemSchemas.PARTITION_SPEC_ID);
     RelDataTypeField buildPartKeyField = getField(build, SystemSchemas.PARTITION_KEY);
+    RelDataTypeField buildPartFileContent = getField(build, SystemSchemas.FILE_CONTENT);
 
     // build the join condition:
     //   data.partitionSpecId == deletes.partitionSpecId &&
@@ -694,12 +695,48 @@ public RelNode buildDataAndDeleteFileJoinAndAggregate(RelNode data, RelNode dele
                 rexBuilder.makeInputRef(probePartKeyField.getType(), probePartKeyField.getIndex()),
                 rexBuilder.makeInputRef(
                     buildPartKeyField.getType(), probeFieldCount + buildPartKeyField.getIndex())));
+
+    RexNode fileContentIsEqualityDelete =
+        rexBuilder.makeCall(
+            SqlStdOperatorTable.AND,
+            rexBuilder.makeCall(
+                SqlStdOperatorTable.EQUALS,
+                rexBuilder.makeLiteral("EQUALITY_DELETES", buildPartFileContent.getType()),
+                rexBuilder.makeInputRef(buildPartFileContent.getType(), probeFieldCount + buildPartFileContent.getIndex())
+            ),
+            rexBuilder.makeCall(
+                SqlStdOperatorTable.LESS_THAN,
+                rexBuilder.makeInputRef(probeSeqNumField.getType(), probeSeqNumField.getIndex()),
+                rexBuilder.makeInputRef(
+                    buildSeqNumField.getType(), probeFieldCount + buildSeqNumField.getIndex()))
+        );
+
+    RexNode fileContentIsPositionalDelete=
+        rexBuilder.makeCall(
+            SqlStdOperatorTable.AND,
+            rexBuilder.makeCall(
+                SqlStdOperatorTable.EQUALS,
+                rexBuilder.makeLiteral("POSITION_DELETES", buildPartFileContent.getType()),
+                rexBuilder.makeInputRef(buildPartFileContent.getType(), probeFieldCount + buildPartFileContent.getIndex())
+            ),
+            rexBuilder.makeCall(
+                SqlStdOperatorTable.LESS_THAN_OR_EQUAL,
+                rexBuilder.makeInputRef(probeSeqNumField.getType(), probeSeqNumField.getIndex()),
+                rexBuilder.makeInputRef(
+                    buildSeqNumField.getType(), probeFieldCount + buildSeqNumField.getIndex()))
+        );
+
+
+
+
+    //RexNode extraJoinCondition = fileContentIsPositionalDelete;
+
     RexNode extraJoinCondition =
         rexBuilder.makeCall(
-            SqlStdOperatorTable.LESS_THAN_OR_EQUAL,
-            rexBuilder.makeInputRef(probeSeqNumField.getType(), probeSeqNumField.getIndex()),
-            rexBuilder.makeInputRef(
-                buildSeqNumField.getType(), probeFieldCount + buildSeqNumField.getIndex()));
+            SqlStdOperatorTable.OR,
+            fileContentIsEqualityDelete,
+            fileContentIsPositionalDelete);
+
 
     RelNode output =
         HashJoinPrel.create(

sabot/kernel/src/test/resources/iceberg/v2/upsert_table_with_equality_deletes/metadata/v1.metadata.json

@@ -0,0 +1,71 @@
+{
+  "format-version" : 2,
+  "table-uuid" : "4388d559-8bdf-4c73-9948-55f38e1a1110",
+  "location" : "file:///tmp/iceberg-test-tables/v2/upsert_table_with_equality_deletes",
+  "last-sequence-number" : 0,
+  "last-updated-ms" : 1750236420409,
+  "last-column-id" : 6,
+  "current-schema-id" : 0,
+  "schemas" : [ {
+    "type" : "struct",
+    "schema-id" : 0,
+    "identifier-field-ids" : [ 1 ],
+    "fields" : [ {
+      "id" : 1,
+      "name" : "hero_id",
+      "required" : true,
+      "type" : "int"
+    }, {
+      "id" : 2,
+      "name" : "name",
+      "required" : false,
+      "type" : "string"
+    }, {
+      "id" : 3,
+      "name" : "real_name",
+      "required" : false,
+      "type" : "string"
+    }, {
+      "id" : 4,
+      "name" : "universe",
+      "required" : false,
+      "type" : "string"
+    }, {
+      "id" : 5,
+      "name" : "alignment",
+      "required" : false,
+      "type" : "string"
+    }, {
+      "id" : 6,
+      "name" : "first_appearance",
+      "required" : false,
+      "type" : "date"
+    } ]
+  } ],
+  "default-spec-id" : 0,
+  "partition-specs" : [ {
+    "spec-id" : 0,
+    "fields" : [ {
+      "name" : "hero_id",
+      "transform" : "identity",
+      "source-id" : 1,
+      "field-id" : 1000
+    } ]
+  } ],
+  "last-partition-id" : 1000,
+  "default-sort-order-id" : 0,
+  "sort-orders" : [ {
+    "order-id" : 0,
+    "fields" : [ ]
+  } ],
+  "properties" : {
+    "write.parquet.compression-codec" : "zstd",
+    "write.upsert.enabled" : "true"
+  },
+  "current-snapshot-id" : -1,
+  "refs" : { },
+  "snapshots" : [ ],
+  "statistics" : [ ],
+  "snapshot-log" : [ ],
+  "metadata-log" : [ ]
+}

sabot/kernel/src/test/resources/iceberg/v2/upsert_table_with_equality_deletes/metadata/v2.metadata.json

@@ -0,0 +1,107 @@
+{
+  "format-version" : 2,
+  "table-uuid" : "4388d559-8bdf-4c73-9948-55f38e1a1110",
+  "location" : "file:///tmp/iceberg-test-tables/v2/upsert_table_with_equality_deletes",
+  "last-sequence-number" : 1,
+  "last-updated-ms" : 1750236531649,
+  "last-column-id" : 6,
+  "current-schema-id" : 0,
+  "schemas" : [ {
+    "type" : "struct",
+    "schema-id" : 0,
+    "identifier-field-ids" : [ 1 ],
+    "fields" : [ {
+      "id" : 1,
+      "name" : "hero_id",
+      "required" : true,
+      "type" : "int"
+    }, {
+      "id" : 2,
+      "name" : "name",
+      "required" : false,
+      "type" : "string"
+    }, {
+      "id" : 3,
+      "name" : "real_name",
+      "required" : false,
+      "type" : "string"
+    }, {
+      "id" : 4,
+      "name" : "universe",
+      "required" : false,
+      "type" : "string"
+    }, {
+      "id" : 5,
+      "name" : "alignment",
+      "required" : false,
+      "type" : "string"
+    }, {
+      "id" : 6,
+      "name" : "first_appearance",
+      "required" : false,
+      "type" : "date"
+    } ]
+  } ],
+  "default-spec-id" : 0,
+  "partition-specs" : [ {
+    "spec-id" : 0,
+    "fields" : [ {
+      "name" : "hero_id",
+      "transform" : "identity",
+      "source-id" : 1,
+      "field-id" : 1000
+    } ]
+  } ],
+  "last-partition-id" : 1000,
+  "default-sort-order-id" : 0,
+  "sort-orders" : [ {
+    "order-id" : 0,
+    "fields" : [ ]
+  } ],
+  "properties" : {
+    "write.parquet.compression-codec" : "zstd",
+    "write.upsert.enabled" : "true"
+  },
+  "current-snapshot-id" : 1589939248529822747,
+  "refs" : {
+    "main" : {
+      "snapshot-id" : 1589939248529822747,
+      "type" : "branch"
+    }
+  },
+  "snapshots" : [ {
+    "sequence-number" : 1,
+    "snapshot-id" : 1589939248529822747,
+    "timestamp-ms" : 1750236531649,
+    "summary" : {
+      "operation" : "overwrite",
+      "flink.operator-id" : "8c01f44774945068107a18f5580b6e5f",
+      "flink.job-id" : "ecac75feda6c2ebf393166a2b152a30d",
+      "flink.max-committed-checkpoint-id" : "1",
+      "added-data-files" : "4",
+      "added-equality-delete-files" : "4",
+      "added-delete-files" : "4",
+      "added-records" : "4",
+      "added-files-size" : "8839",
+      "added-equality-deletes" : "4",
+      "changed-partition-count" : "4",
+      "total-records" : "4",
+      "total-files-size" : "8839",
+      "total-data-files" : "4",
+      "total-delete-files" : "4",
+      "total-position-deletes" : "0",
+      "total-equality-deletes" : "4"
+    },
+    "manifest-list" : "file:/tmp/iceberg-test-tables/v2/upsert_table_with_equality_deletes/metadata/snap-1589939248529822747-1-b2022892-bf74-4bbd-9657-117185b216c7.avro",
+    "schema-id" : 0
+  } ],
+  "statistics" : [ ],
+  "snapshot-log" : [ {
+    "timestamp-ms" : 1750236531649,
+    "snapshot-id" : 1589939248529822747
+  } ],
+  "metadata-log" : [ {
+    "timestamp-ms" : 1750236420409,
+    "metadata-file" : "file:/tmp/iceberg-test-tables/v2/upsert_table_with_equality_deletes/metadata/v1.metadata.json"
+  } ]
+}