Add configurable filter and sort at INSERT OVERWRITE in iceberg dedup example (#85)

Author: ferenc-csaky

iceberg-data-deduplication/dedup.sqrl

@@ -86,14 +86,22 @@ _CastDataToCompact.partition_id := CAST(partition_id AS {{partitionColType}});
 _InputData :=
 SELECT /*+ BROADCAST(c) */ d.*
 FROM {{tableName}} AS d
-        JOIN _CastDataToCompact AS c
-ON d.{{partitionCol}} = c.partition_id AND d.{{timeBucketCol}} <= c.max_time_bucket;
+JOIN _CastDataToCompact AS c
+  ON d.{{partitionCol}} = c.partition_id
+ AND d.{{timeBucketCol}} <= c.max_time_bucket;
 
 _InputData.{{timeBucketCol}} := 0;
 
 _DistInputData := DISTINCT _InputData ON {{distinctKeyCol}} ORDER BY {{timestampCol}} DESC;
 
-INSERT OVERWRITE {{tableName}} SELECT * FROM _DistInputData;
+INSERT OVERWRITE {{tableName}}
+    SELECT * FROM _DistInputData
+    -- filter only when "deleteFlagCol" has a value
+    {{#deleteFlagCol}}
+    WHERE {{deleteFlagCol}} IS TRUE
+    {{/deleteFlagCol}}
+    -- sort by partition column, and optionally any other column(s)
+    ORDER BY {{partitionCol}} {{#sortKey}}, {{sortKey}}{{/sortKey}};
 
 _CompactionResult :=
 SELECT table_name,

iceberg-data-deduplication/dedup_package.json

@@ -14,6 +14,8 @@
       "partitionColType": "bigint",
       "timeBucketCol": "time_bucket",
       "timestampCol": "ts",
+      "deleteFlagCol": null,
+      "sortKey": null,
       "bufferSeconds": 2,
       "bucketSeconds": 5,
       "newDataNormalizerInMB": 1,