Finalize split iceberg deduplication example (#107)

Author: ferenc-csaky

.github/workflows/build.yml

@@ -88,7 +88,8 @@ jobs:
           - example: iceberg-dedup
             path: iceberg-data-deduplication
             test_commands: |
-              compile data_deduplication_package.json
+              compile data_compaction_package.json
+              compile data_delete_package.json
 
           - example: iot-sensor
             path: iot-sensor-metrics

…ta-deduplication/data_deduplication.sqrl …-data-deduplication/data_compaction.sqrliceberg-data-deduplication/data_deduplication.sqrl renamed to iceberg-data-deduplication/data_compaction.sqrl iceberg-data-deduplication/data_deduplication.sqrl renamed to iceberg-data-deduplication/data_compaction.sqrl

@@ -5,23 +5,38 @@ IMPORT tables.{{tableName}}.*;
 IMPORT tables.compaction_tracker.*;
 IMPORT udfs.read_partition_sizes;
 
+-- Collect previously compacted partitions
+_PrevCompactions :=
+SELECT partition_id,
+       MAX(partition_time_bucket) AS partition_time_bucket
+FROM CompactionTracker
+WHERE `action` = 'overwrite'
+GROUP BY partition_id;
+
 _AllPartitions :=
-SELECT partition_map['{{partitionCol}}'] AS partition_id,
-       CAST(partition_map['{{timeBucketCol}}'] AS BIGINT) AS time_bucket,
-       partition_size
-FROM TABLE(read_partition_sizes(
-        '{{warehouse}}', '{{catalogType}}', '{{catalogName}}', '{{databaseName}}', '{{tableName}}'));
+SELECT p.*
+FROM (
+  SELECT partition_map['{{partitionCol}}'] AS partition_id,
+         CAST(partition_map['{{timeBucketCol}}'] AS BIGINT) AS time_bucket,
+         partition_size
+  FROM TABLE(read_partition_sizes('{{warehouse}}', '{{catalogType}}', '{{catalogName}}', '{{databaseName}}', '{{tableName}}'))
+) p
+-- filter out partitions that have been compacted already
+LEFT JOIN _PrevCompactions pc ON pc.partition_id = p.partition_id
+WHERE pc.partition_id IS NULL
+   OR pc.partition_time_bucket < p.time_bucket;
 
 _Partitions :=
 SELECT * FROM _AllPartitions
-WHERE time_bucket <= FLOOR((CAST('${DEPLOYMENT_TIMESTAMP}' AS BIGINT) - {{bufferSeconds}}) / {{bucketSeconds}}) * {{bucketSeconds}};
+WHERE time_bucket <= FLOOR((CAST('${DEPLOYMENT_TIMESTAMP}' AS BIGINT) - {{bufferSeconds}} * 1000) / ({{bucketSeconds}} * 1000)) * {{bucketSeconds}} * 1000;
 
 _PartitionSizing :=
 SELECT partition_id,
+       time_bucket,
        SUM(CASE WHEN time_bucket = 0 THEN partition_size ELSE 0 END) AS base_size,
        SUM(CASE WHEN time_bucket > 0 THEN partition_size ELSE 0 END) AS new_size
 FROM _Partitions
-GROUP BY partition_id;
+GROUP BY partition_id, time_bucket;
 
 _PartitionSizing.total_size := base_size + new_size;
 _PartitionSizing.new_rel_percentage := CAST(new_size AS DOUBLE) / GREATEST(total_size, 1) * 100;
@@ -30,6 +45,7 @@ _PartitionSizing.score := new_rel_percentage + new_abs_percentage;
 
 _PartitionPriority :=
 SELECT partition_id,
+       time_bucket,
        new_size,
        total_size,
        new_rel_percentage,
@@ -45,9 +61,10 @@ _PartitionsToCompact :=
 SELECT DISTINCT '{{tableName}}' AS table_name,
                 '${DEPLOYMENT_ID}' AS job_id,
                 partition_id,
+                time_bucket,
                 new_rel_percentage,
                 new_abs_percentage,
-                FLOOR((CAST('${DEPLOYMENT_TIMESTAMP}' AS BIGINT) - {{bufferSeconds}}) / {{bucketSeconds}}) * {{bucketSeconds}} AS max_time_bucket
+                FLOOR((CAST('${DEPLOYMENT_TIMESTAMP}' AS BIGINT) - {{bufferSeconds}} * 1000) / ({{bucketSeconds}} * 1000)) * {{bucketSeconds}} * 1000 AS max_time_bucket
 FROM _PartitionPriority
 WHERE cumulative_total_size <= {{maxTotalPartitionSizeInMB}} * 1024 * 1024
    OR row_num <= 1;
@@ -69,7 +86,7 @@ INSERT OVERWRITE {{tableName}}
     SELECT * FROM _DistInputData
     -- filter only when "deleteFlagCol" has a value
     {{#deleteFlagCol}}
-    WHERE {{deleteFlagCol}} IS NOT TRUE
+    WHERE {{{deleteFlagCol}}} IS NOT TRUE
     {{/deleteFlagCol}}
     -- sort by partition column, and optionally any other column(s)
     ORDER BY {{partitionCol}} {{#sortKey}}, {{sortKey}}{{/sortKey}};
@@ -78,10 +95,12 @@ _CompactionResult :=
 SELECT CAST(table_name AS STRING),
        CAST(job_id AS STRING),
        CAST(partition_id AS STRING),
+       '{{partitionCol}}' AS partition_col_name,
+       time_bucket AS partition_time_bucket,
        new_rel_percentage,
        new_abs_percentage,
        max_time_bucket,
-       'overwrite' AS action,
+       'overwrite' AS `action`,
        NOW() AS action_time
 FROM _PartitionsToCompact;
 

…lication/data_deduplication_package.json …duplication/data_compaction_package.jsoniceberg-data-deduplication/data_deduplication_package.json renamed to iceberg-data-deduplication/data_compaction_package.json iceberg-data-deduplication/data_deduplication_package.json renamed to iceberg-data-deduplication/data_compaction_package.json

@@ -2,7 +2,7 @@
   "version": "1",
   "enabled-engines": ["flink"],
   "script": {
-    "main": "data_deduplication.sqrl",
+    "main": "data_compaction.sqrl",
     "config": {
       "warehouse": "sqrl-iceberg-data",
       "catalogType": "hadoop",

iceberg-data-deduplication/data_delete.sqrl

@@ -5,42 +5,50 @@ _DataToDelete :=
 SELECT table_name,
        job_id,
        partition_id,
+       partition_col_name,
        new_rel_percentage,
        new_abs_percentage,
        max_time_bucket
 FROM (
     SELECT table_name,
            job_id,
            partition_id,
+           partition_col_name,
            new_rel_percentage,
            new_abs_percentage,
            max_time_bucket,
-           action,
+           `action`,
            ROW_NUMBER() OVER (PARTITION BY table_name, partition_id ORDER BY action_time DESC) AS row_num
     FROM CompactionTracker
-    WHERE action_time < NOW() - INTERVAL '{{commit_wait_sec}}' SECOND AND action_time > NOW() - INTERVAL '24' HOUR -- time buffer to ensure commit happened and stop attempting to delete
+    WHERE action_time < NOW() - INTERVAL '{{commitWaitSec}}' SECOND {{stopDeleteFilter}} -- time buffer to ensure commit happened and stop attempting to delete based on config
 )
-WHERE row_num = 1 AND action <> 'delete'; -- compacted partitions haven't been deleted yet
+WHERE row_num = 1 AND `action` <> 'delete'; -- filter compacted partitions that haven't been deleted yet
 
---TODO: need extra safety: validate that the selected partitions should be deleted by checking that the last partition update time < partition 0 update time
+-- TODO: Validate that the selected partitions should be deleted by checking that the last partition update time < partition 0 update time
 
 _DeleteFnResult :=
-SELECT delete_duplicated_data(
-               '{{warehouse}}',
-               '{{catalogType}}',
-               '{{catalogName}}',
-               '{{databaseName}}',
-               MAX(table_name),
-               MAX(max_time_bucket),
-               COLLECT(MAP['{{partitionCol}}', partition_id])) AS res
+SELECT *, delete_duplicated_data(
+                '{{warehouse}}',
+                '{{catalogType}}',
+                '{{catalogName}}',
+                '{{databaseName}}',
+                table_name,
+                partition_col_name,
+                partition_id,
+                max_time_bucket) AS res
 FROM _DataToDelete;
 
 _DeleteResult :=
-SELECT
-    d.*,
-    CASE WHEN fn.res = true THEN 'delete' ELSE 'delete_failed' END AS action,
-        NOW() AS action_time
-FROM _DataToDelete AS d
-    CROSS JOIN (SELECT res FROM _DeleteFnResult LIMIT 1) AS fn;
+SELECT table_name,
+       job_id,
+       partition_id,
+       partition_col_name,
+       CAST(NULL AS BIGINT) AS partition_time_bucket,
+       new_rel_percentage,
+       new_abs_percentage,
+       max_time_bucket,
+       CASE WHEN res = true THEN 'delete' ELSE 'delete_failed' END AS `action`,
+       NOW() AS action_time
+FROM _DeleteFnResult;
 
 INSERT INTO CompactionTracker SELECT * FROM _DeleteResult;

iceberg-data-deduplication/data_delete_package.json

@@ -8,8 +8,8 @@
       "catalogType": "hadoop",
       "catalogName": "mycatalog",
       "databaseName": "mydatabase",
-      "partitionCol": "account_id",
-      "commit_wait_sec": 1
+      "commitWaitSec": 1,
+      "stopDeleteFilter": ""
     }
   },
   "engines": {

iceberg-data-deduplication/tables/compaction_tracker.sqrl

@@ -1,12 +1,14 @@
 CREATE TABLE `CompactionTracker` (
-    `table_name`         STRING NOT NULL,
-    `job_id`             STRING NOT NULL,
-    `partition_id`       STRING NOT NULL,
-    `new_rel_percentage` DOUBLE NOT NULL,
-    `new_abs_percentage` DOUBLE NOT NULL,
-    `max_time_bucket`    BIGINT NOT NULL,
-    `action`             STRING NOT NULL,
-    `action_time`        TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL
+    `table_name`            STRING NOT NULL,
+    `job_id`                STRING NOT NULL,
+    `partition_id`          STRING NOT NULL,
+    `partition_col_name`    STRING NOT NULL,
+    `partition_time_bucket` BIGINT,
+    `new_rel_percentage`    DOUBLE NOT NULL,
+    `new_abs_percentage`    DOUBLE NOT NULL,
+    `max_time_bucket`       BIGINT NOT NULL,
+    `action`                STRING NOT NULL,
+    `action_time`           TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL
 ) PARTITIONED BY (`table_name`)
 WITH (
     'connector' = 'iceberg',

iceberg-data-deduplication/udfs/myudf-0.1.0-SNAPSHOT.jar

@@ -1,13 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
 <project xmlns="http://maven.apache.org/POM/4.0.0"
-  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
 
   <groupId>com.myudf</groupId>
   <artifactId>myudf</artifactId>
   <version>0.1.0-SNAPSHOT</version>
 
   <properties>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <maven.compiler.source>17</maven.compiler.source>
     <maven.compiler.target>17</maven.compiler.target>
   </properties>
@@ -16,7 +18,7 @@
     <dependency>
       <groupId>org.projectlombok</groupId>
       <artifactId>lombok</artifactId>
-      <version>1.18.42</version>
+      <version>1.18.46</version>
       <scope>provided</scope>
     </dependency>
 
@@ -29,7 +31,7 @@
     <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-table-api-java-bridge</artifactId>
-      <version>1.19.3</version>
+      <version>2.2.0</version>
       <scope>provided</scope>
     </dependency>
 
@@ -42,9 +44,22 @@
 
     <dependency>
       <groupId>org.apache.iceberg</groupId>
-      <artifactId>iceberg-flink-runtime-1.19</artifactId>
-      <version>1.9.2</version>
+      <artifactId>iceberg-flink-runtime-2.0</artifactId>
+      <version>1.10.1</version>
       <scope>provided</scope>
     </dependency>
   </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-jar-plugin</artifactId>
+        <version>3.4.2</version>
+        <configuration>
+          <outputDirectory>${project.basedir}</outputDirectory>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
 </project>