Add code improvements (#409)

* Add code improvements * Add code improvements
memiiso · Sep 4, 2024 · 6e553bb · 6e553bb
1 parent 3f7e469
commit 6e553bb
Show file tree

Hide file tree

Showing 7 changed files with 26 additions and 20 deletions.
diff --git a/debezium-server-iceberg-sink/pom.xml b/debezium-server-iceberg-sink/pom.xml
@@ -133,6 +133,11 @@
             <artifactId>iceberg-azure-bundle</artifactId>
             <version>${version.iceberg}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.iceberg</groupId>
+            <artifactId>iceberg-bundled-guava</artifactId>
+            <version>${version.iceberg}</version>
+        </dependency>
         <!-- Google -->
         <dependency>
             <groupId>com.google.cloud.bigdataoss</groupId>

diff --git a/...berg-sink/src/main/java/io/debezium/server/iceberg/tableoperator/BaseDeltaTaskWriter.java b/...berg-sink/src/main/java/io/debezium/server/iceberg/tableoperator/BaseDeltaTaskWriter.java
@@ -12,7 +12,7 @@
 import org.apache.iceberg.types.TypeUtil;
 
 import java.io.IOException;
-import java.util.List;
+import java.util.Set;
 
 import static io.debezium.server.iceberg.tableoperator.IcebergTableOperator.opFieldName;
 
@@ -32,7 +32,7 @@ abstract class BaseDeltaTaskWriter extends BaseTaskWriter<Record> {
                       FileIO io,
                       long targetFileSize,
                       Schema schema,
-                      List<Integer> equalityFieldIds,
+                      Set<Integer> equalityFieldIds,
                       boolean upsert,
                       boolean upsertKeepDeletes) {
     super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
@@ -59,16 +59,18 @@ public void write(Record row) throws IOException {
           "This field is required when updating or deleting data, when running in upsert mode."
       );
     }
-    if (upsert && !opFieldValue.equals("c")) {// anything which not an insert is upsert
-      writer.delete(row);
-    }
-    // if its deleted row and upsertKeepDeletes = true then add deleted record to target table
-    // else deleted records are deleted from target table
-    if (
-        upsertKeepDeletes
-            || !(opFieldValue.equals("d")))// anything which not an insert is upsert
-    {
+    if (!upsert) {
+      // APPEND ONLY MODE!!
       writer.write(row);
+    } else {
+      // UPSERT MODE
+      if (!opFieldValue.equals("c")) {// anything which not created is deleted first
+        writer.delete(row);
+      }
+      // when upsertKeepDeletes = FALSE we dont keep deleted record
+      if (upsertKeepDeletes || !opFieldValue.equals("d")) {
+        writer.write(row);
+      }
     }
   }
 

diff --git a/...ink/src/main/java/io/debezium/server/iceberg/tableoperator/IcebergTableWriterFactory.java b/...ink/src/main/java/io/debezium/server/iceberg/tableoperator/IcebergTableWriterFactory.java
@@ -2,8 +2,7 @@
 
 import io.debezium.server.iceberg.IcebergUtil;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Set;
 
 import jakarta.enterprise.context.Dependent;
 import org.apache.iceberg.FileFormat;
@@ -37,8 +36,7 @@ public BaseTaskWriter<Record> create(Table icebergTable) {
     GenericAppenderFactory appenderFactory = IcebergUtil.getTableAppender(icebergTable);
     OutputFileFactory fileFactory = IcebergUtil.getTableOutputFileFactory(icebergTable, format);
     // equality Field Ids
-    List<Integer> equalityFieldIds = new ArrayList<>(icebergTable.schema().identifierFieldIds());
-
+    Set<Integer> equalityFieldIds = icebergTable.schema().identifierFieldIds();
     BaseTaskWriter<Record> writer;
 
     // 1. TABLE DONT HAVE identifierFieldIds

diff --git a/...g-sink/src/main/java/io/debezium/server/iceberg/tableoperator/PartitionedDeltaWriter.java b/...g-sink/src/main/java/io/debezium/server/iceberg/tableoperator/PartitionedDeltaWriter.java
@@ -2,8 +2,8 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
-import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.PartitionKey;
@@ -29,7 +29,7 @@ class PartitionedDeltaWriter extends BaseDeltaTaskWriter {
                          FileIO io,
                          long targetFileSize,
                          Schema schema,
-                         List<Integer> equalityFieldIds,
+                         Set<Integer> equalityFieldIds,
                          boolean upsert,
                          boolean upsertKeepDeletes) {
     super(spec, format, appenderFactory, fileFactory, io, targetFileSize, schema, equalityFieldIds, upsert, upsertKeepDeletes);

diff --git a/...sink/src/main/java/io/debezium/server/iceberg/tableoperator/UnpartitionedDeltaWriter.java b/...sink/src/main/java/io/debezium/server/iceberg/tableoperator/UnpartitionedDeltaWriter.java
@@ -1,7 +1,7 @@
 package io.debezium.server.iceberg.tableoperator;
 
 import java.io.IOException;
-import java.util.List;
+import java.util.Set;
 
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.PartitionSpec;
@@ -21,7 +21,7 @@ class UnpartitionedDeltaWriter extends BaseDeltaTaskWriter {
                            FileIO io,
                            long targetFileSize,
                            Schema schema,
-                           List<Integer> equalityFieldIds,
+                           Set<Integer> equalityFieldIds,
                            boolean upsert,
                            boolean upsertKeepDeletes) {
     super(spec, format, appenderFactory, fileFactory, io, targetFileSize, schema, equalityFieldIds, upsert, upsertKeepDeletes);

diff --git a/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/TestConfigSource.java b/debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/TestConfigSource.java
@@ -75,6 +75,7 @@ public TestConfigSource() {
     config.put("debezium.source.table.whitelist", "inventory.*");
     config.put("%postgresql.debezium.source.replica.identity.autoset.values", "inventory.*:FULL");
 
+    config.put("quarkus.devservices.enabled", "false");
     config.put("quarkus.log.level", "WARN");
     config.put("quarkus.log.category.\"org.apache.spark\".level", "WARN");
     config.put("quarkus.log.category.\"org.apache.hadoop\".level", "ERROR");

diff --git a/pom.xml b/pom.xml
@@ -31,7 +31,7 @@
         <version.assembly.plugin>3.7.1</version.assembly.plugin>
         <!-- Use same version as iceberg https://github.com/apache/iceberg/blob/main/gradle/libs.versions.toml#L53-->
         <version.jackson>2.14.2</version.jackson>
-        <version.iceberg>1.6.0</version.iceberg>
+        <version.iceberg>1.6.1</version.iceberg>
         <!-- Following two properties defines which version of iceberg-spark-runtime is used -->
         <!-- Example https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-spark-runtime-3.4_2.13/1.4.3 -->
         <version.spark.major>4.0</version.spark.major>