Add test packages for iceberg-data-deduplication example

.github/workflows/build.yml

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

.gitignore

@@ -8,6 +8,11 @@ cache
 */myenv
 warehouse
 
+# Performance benchmarking
+node_modules/
+iceberg-data/
+sqrl-iceberg-data/
+
 
 # Created by https://www.toptal.com/developers/gitignore/api/python
 # Edit at https://www.toptal.com/developers/gitignore?templates=python
@@ -225,8 +230,3 @@ Temporary Items
 
 py-venv
 # End of https://www.toptal.com/developers/gitignore/api/macos
-
-##performance benchmarking
-node_modules/
-iceberg-data/
-sqrl-iceberg-data/

iceberg-data-deduplication/data_compaction.sqrl

@@ -1,8 +1,8 @@
 -- This script runs a deduplication compaction on Iceberg tables
 -- It currently assumes a single primary partition column and a single time bucket partition column based on seconds
 
-IMPORT tables.{{tableName}}.*;
-IMPORT tables.compaction_tracker.*;
+IMPORT tables.data_{{env}}.*;
+IMPORT tables.tracker_{{env}}.*;
 IMPORT udfs.read_partition_sizes;
 
 -- Collect previously compacted partitions
@@ -73,7 +73,7 @@ _PartitionsToCompact.partition_id := CAST(partition_id AS {{partitionColType}});
 
 _InputData :=
 SELECT /*+ BROADCAST(c) */ d.*
-FROM {{tableName}} AS d
+FROM DedupData AS d
 JOIN _PartitionsToCompact AS c
   ON d.{{partitionCol}} = c.partition_id
  AND d.{{timeBucketCol}} <= c.max_time_bucket;
@@ -82,7 +82,7 @@ _InputData.{{timeBucketCol}} := 0;
 
 _DistInputData := DISTINCT _InputData ON {{distinctKeyCol}} ORDER BY {{timestampCol}} DESC;
 
-INSERT OVERWRITE {{tableName}}
+INSERT OVERWRITE DedupData
     SELECT * FROM _DistInputData
     -- filter only when "deleteFlagCol" has a value
     {{#deleteFlagCol}}

iceberg-data-deduplication/data_compaction_package.json → iceberg-data-deduplication/data_compaction_package_local.json

@@ -4,6 +4,7 @@
   "script": {
     "main": "data_compaction.sqrl",
     "config": {
+      "env": "local",
       "warehouse": "sqrl-iceberg-data",
       "catalogType": "hadoop",
       "catalogName": "mycatalog",

iceberg-data-deduplication/data_compaction_package_test.json

@@ -0,0 +1,34 @@
+{
+  "version": "1",
+  "enabled-engines": ["flink"],
+  "script": {
+    "main": "data_compaction.sqrl",
+    "config": {
+      "env": "test",
+      "warehouse": "s3://sqrl-examples-data-bucket/datasqrl-examples/iceberg-canary",
+      "catalogType": "glue",
+      "catalogName": "canary",
+      "databaseName": "canary",
+      "tableName": "iceberg_dedup_test",
+      "distinctKeyCol": "account_id",
+      "partitionCol": "account_id",
+      "partitionColType": "bigint",
+      "timeBucketCol": "time_bucket",
+      "timestampCol": "ts",
+      "deleteFlagCol": null,
+      "sortKey": null,
+      "bufferSeconds": 2,
+      "bucketSeconds": 5,
+      "newDataNormalizerInMB": 1,
+      "maxTotalPartitionSizeInMB": 10
+    }
+  },
+  "engines": {
+    "flink": {
+      "config": {
+        "execution.runtime-mode": "batch",
+        "table.exec.source.idle-timeout": "1 ms"
+      }
+    }
+  }
+}

iceberg-data-deduplication/data_delete.sqrl

@@ -1,4 +1,4 @@
-IMPORT tables.compaction_tracker.*;
+IMPORT tables.tracker_{{env}}.*;
 IMPORT udfs.delete_duplicated_data;
 
 _DataToDelete :=

iceberg-data-deduplication/data_delete_package.json → iceberg-data-deduplication/data_delete_package_local.json

@@ -4,6 +4,7 @@
   "script": {
     "main": "data_delete.sqrl",
     "config": {
+      "env": "local",
       "warehouse": "sqrl-iceberg-data",
       "catalogType": "hadoop",
       "catalogName": "mycatalog",

iceberg-data-deduplication/data_delete_package_test.json

@@ -0,0 +1,23 @@
+{
+  "version": "1",
+  "enabled-engines": ["flink"],
+  "script": {
+    "main": "data_delete.sqrl",
+    "config": {
+      "env": "test",
+      "warehouse": "s3://sqrl-examples-data-bucket/datasqrl-examples/iceberg-canary",
+      "catalogType": "glue",
+      "catalogName": "canary",
+      "databaseName": "canary",
+      "commitWaitSec": 1,
+      "stopDeleteFilter": ""
+    }
+  },
+  "engines": {
+    "flink": {
+      "config": {
+        "execution.runtime-mode": "batch"
+      }
+    }
+  }
+}

iceberg-data-deduplication/datagen.sqrl

@@ -1,8 +1,8 @@
 IMPORT tables.min_source AS _CDC;
-IMPORT tables.{{tableName}}.*;
+IMPORT tables.data_{{env}}.*;
 
 CountRecords := SELECT COUNT(*) AS num_records FROM _CDC.SrcRecords;
 
 EXPORT CountRecords TO logger.records;
 
-INSERT INTO {{tableName}} SELECT c.*, EXTRACT(SECOND FROM c.ts) + 1 AS time_bucket FROM _CDC.SrcRecords c;
+INSERT INTO DedupData SELECT c.*, EXTRACT(SECOND FROM c.ts) + 1 AS time_bucket FROM _CDC.SrcRecords c;

iceberg-data-deduplication/datagen_package.json → iceberg-data-deduplication/datagen_package_local.json

@@ -4,6 +4,7 @@
   "script": {
     "main": "datagen.sqrl",
     "config": {
+      "env": "local",
       "warehouse": "sqrl-iceberg-data",
       "catalogType": "hadoop",
       "catalogName": "mycatalog",

iceberg-data-deduplication/datagen_package_test.json

@@ -0,0 +1,24 @@
+{
+  "version": "1",
+  "enabled-engines": ["flink"],
+  "script": {
+    "main": "datagen.sqrl",
+    "config": {
+      "env": "test",
+      "warehouse": "s3://sqrl-examples-data-bucket/datasqrl-examples/iceberg-canary",
+      "catalogType": "glue",
+      "catalogName": "canary",
+      "databaseName": "canary",
+      "tableName": "iceberg_dedup_test"
+    }
+  },
+  "engines": {
+    "flink": {
+      "config": {
+        "execution.runtime-mode": "batch",
+        "execution.checkpointing.interval": "8 s",
+        "table.exec.source.idle-timeout": "1 ms"
+      }
+    }
+  }
+}

iceberg-data-deduplication/tables/cdc_data.sqrl → iceberg-data-deduplication/tables/data_local.sqrl

@@ -1,4 +1,4 @@
-CREATE TABLE `cdc_data` (
+CREATE TABLE `DedupData` (
     `account_id`  BIGINT NOT NULL,
     `id`          BIGINT NOT NULL,
     `pt`          STRING NOT NULL,

iceberg-data-deduplication/tables/data_test.sqrl

@@ -0,0 +1,20 @@
+CREATE TABLE `DedupData` (
+    `account_id`  BIGINT NOT NULL,
+    `id`          BIGINT NOT NULL,
+    `pt`          STRING NOT NULL,
+    `op`          STRING NOT NULL,
+    `payload`     STRING NOT NULL,
+    `ts`          TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL,
+    `time_bucket` BIGINT NOT NULL,
+    PRIMARY KEY (`account_id`, `id`) NOT ENFORCED
+) PARTITIONED BY (`account_id`, `id`, `time_bucket`)
+WITH (
+    'connector' = 'iceberg',
+    'catalog-name' = '{{catalogName}}',
+    'catalog-database' = '{{databaseName}}',
+    'catalog-table' = '{{tableName}}',
+    'catalog-impl' = 'org.apache.iceberg.aws.glue.GlueCatalog',
+    'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO',
+    'warehouse' = '{{warehouse}}',
+    'format-version' = '2'
+);

iceberg-data-deduplication/tables/tracker_test.sqrl

@@ -0,0 +1,22 @@
+CREATE TABLE `CompactionTracker` (
+    `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',
+    'catalog-name' = '{{catalogName}}',
+    'catalog-database' = '{{databaseName}}',
+    'catalog-table' = 'compaction_tracker',
+    'catalog-impl' = 'org.apache.iceberg.aws.glue.GlueCatalog',
+    'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO',
+    'warehouse' = '{{warehouse}}',
+    'format-version' = '2'
+);