use V2 iceberg sink

Author: ferenc-csaky

sqrl-planner/src/main/java/com/datasqrl/config/CompilerConfigImpl.java

@@ -40,11 +40,6 @@ public boolean compileFlinkPlan() {
     return sqrlConfig.asBool("compile-flink-plan").get();
   }
 
-  @Override
-  public boolean addIcebergSerializationConfig() {
-    return sqrlConfig.asBool("add-iceberg-serialization-config").get();
-  }
-
   @Override
   public CostModel getCostModel() {
     var costModel = sqrlConfig.as("cost-model", Type.class).get();

sqrl-planner/src/main/java/com/datasqrl/config/PackageJson.java

@@ -54,8 +54,6 @@ interface CompilerConfig {
 
     boolean compileFlinkPlan();
 
-    boolean addIcebergSerializationConfig();
-
     CostModel getCostModel();
 
     ExplainConfig getExplain();

sqrl-planner/src/main/java/com/datasqrl/planner/FlinkPhysicalPlan.java

@@ -34,7 +34,6 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.sql.parser.ddl.SqlCreateFunction;
 import org.apache.flink.sql.parser.ddl.SqlCreateTable;
 import org.apache.flink.sql.parser.ddl.SqlTableOption;
@@ -56,12 +55,6 @@
 @Builder
 public class FlinkPhysicalPlan implements EnginePhysicalPlan {
 
-  private static final List<String> ICEBERG_SERIALIZATION_CONFIG =
-      List.of(
-          "org.apache.iceberg.SerializableByteBufferMap: {type: kryo, kryo-type: registered, class: com.esotericsoftware.kryo.serializers.JavaSerializer}",
-          "org.apache.iceberg.GenericDataFile: {type: kryo, kryo-type: registered, class: com.esotericsoftware.kryo.serializers.JavaSerializer}",
-          "org.apache.iceberg.io.WriteResult: {type: kryo, kryo-type: registered, class: com.esotericsoftware.kryo.serializers.JavaSerializer}");
-
   List<String> flinkSql;
   Set<String> connectors;
   Set<String> formats;
@@ -97,11 +90,9 @@ public static class Builder {
     private final Set<String> fullyResolvedFunctions = new HashSet<>();
     private final List<List<RichSqlInsert>> statementSets = new ArrayList<>();
 
-    private final boolean addIcebergSerializationConfig;
     private Configuration config;
 
-    public Builder(Configuration config, boolean addIcebergSerializationConfig) {
-      this.addIcebergSerializationConfig = addIcebergSerializationConfig;
+    public Builder(Configuration config) {
       this.config = config.clone();
       nextBatch();
     }
@@ -180,15 +171,9 @@ public FlinkPhysicalPlan build(Optional<CompiledPlan> compiledPlan) {
                   plan.explain(
                       ExplainFormat.TEXT, ExplainDetail.CHANGELOG_MODE, ExplainDetail.PLAN_ADVICE));
 
-      if (connectors.contains(IcebergEngineFactory.ENGINE_NAME) && addIcebergSerializationConfig) {
-        // We need to enforce the Kryo JavaSerializer for some built-in Iceberg classes
-        var updatedSerConf =
-            ImmutableList.<String>builder()
-                .addAll(config.get(PipelineOptions.SERIALIZATION_CONFIG, new ArrayList<>()))
-                .addAll(ICEBERG_SERIALIZATION_CONFIG)
-                .build();
-
-        config.set(PipelineOptions.SERIALIZATION_CONFIG, updatedSerConf);
+      if (connectors.contains(IcebergEngineFactory.ENGINE_NAME)) {
+        // Make sure we use the V2 sink
+        config.setString("table.exec.iceberg.use-v2-sink", "true");
       }
 
       return new FlinkPhysicalPlan(

sqrl-planner/src/main/java/com/datasqrl/planner/Sqrl2FlinkSQLTranslator.java

@@ -202,7 +202,7 @@ public Sqrl2FlinkSQLTranslator(
           jarUrls.stream().map(URL::toString).collect(Collectors.toList()));
     }
 
-    this.planBuilder = new Builder(config.clone(), compilerConfig.addIcebergSerializationConfig());
+    this.planBuilder = new Builder(config.clone());
 
     if (executionMode == RuntimeExecutionMode.STREAMING) {
       planBuilder.addInferredConfig(flink.getStreamingSpecificConfig());

sqrl-planner/src/main/resources/default-package.json

@@ -5,7 +5,6 @@
     "logger": "print",
     "extended-scalar-types": true,
     "compile-flink-plan": true,
-    "add-iceberg-serialization-config": true,
     "cost-model": "DEFAULT",
     "explain": {
       "text": true,

sqrl-planner/src/main/resources/jsonSchema/packageSchema.json

@@ -39,9 +39,6 @@
         "compile-flink-plan": {
           "type": "boolean"
         },
-        "add-iceberg-serialization-config": {
-          "type": "boolean"
-        },
         "cost-model": {
           "type": "string",
           "enum": ["DEFAULT", "READ", "WRITE"],

sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/FullUseCaseIT.java

@@ -61,7 +61,7 @@ public class FullUseCaseIT extends AbstractFullUseCaseTest {
   @Disabled("Intended for manual usage")
   @Test
   void specificUseCase() {
-    var pkg = USE_CASES.resolve("duckdb").resolve("package.json");
+    var pkg = USE_CASES.resolve("jwt-authorized").resolve("package.json");
 
     var param = new UseCaseParam(pkg);
     fullUseCaseTest(param);