Skip to content

Commit b2dda6d

Browse files
ferenc-csakyferenc-csaky
committed
Add test packages for iceberg-data-deduplication example
1 parent 8375e74 commit b2dda6d

14 files changed

Lines changed: 140 additions & 13 deletions

‎.gitignore‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
.idea
2+
*.iml
23
*/build
34
target
45
h2.db.mv.db
@@ -7,6 +8,11 @@ cache
78
*/myenv
89
warehouse
910

11+
# Performance benchmarking
12+
node_modules/
13+
iceberg-data/
14+
sqrl-iceberg-data/
15+
1016

1117
# Created by https://www.toptal.com/developers/gitignore/api/python
1218
# Edit at https://www.toptal.com/developers/gitignore?templates=python
@@ -224,8 +230,3 @@ Temporary Items
224230

225231
py-venv
226232
# End of https://www.toptal.com/developers/gitignore/api/macos
227-
228-
##performance benchmarking
229-
node_modules/
230-
iceberg-data/
231-
sqrl-iceberg-data/

‎iceberg-data-deduplication/data_compaction.sqrl‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
-- This script runs a deduplication compaction on Iceberg tables
22
-- It currently assumes a single primary partition column and a single time bucket partition column based on seconds
33

4-
IMPORT tables.{{tableName}}.*;
5-
IMPORT tables.compaction_tracker.*;
4+
IMPORT tables.data_{{env}}.*;
5+
IMPORT tables.tracker_{{env}}.*;
66
IMPORT udfs.read_partition_sizes;
77

88
-- Collect previously compacted partitions
@@ -73,7 +73,7 @@ _PartitionsToCompact.partition_id := CAST(partition_id AS {{partitionColType}});
7373

7474
_InputData :=
7575
SELECT /*+ BROADCAST(c) */ d.*
76-
FROM {{tableName}} AS d
76+
FROM DedupData AS d
7777
JOIN _PartitionsToCompact AS c
7878
ON d.{{partitionCol}} = c.partition_id
7979
AND d.{{timeBucketCol}} <= c.max_time_bucket;
@@ -82,7 +82,7 @@ _InputData.{{timeBucketCol}} := 0;
8282

8383
_DistInputData := DISTINCT _InputData ON {{distinctKeyCol}} ORDER BY {{timestampCol}} DESC;
8484

85-
INSERT OVERWRITE {{tableName}}
85+
INSERT OVERWRITE DedupData
8686
SELECT * FROM _DistInputData
8787
-- filter only when "deleteFlagCol" has a value
8888
{{#deleteFlagCol}}

iceberg-data-deduplication/data_compaction_package.json renamed to iceberg-data-deduplication/data_compaction_package_local.json

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
"script": {
55
"main": "data_compaction.sqrl",
66
"config": {
7+
"env": "local",
78
"warehouse": "sqrl-iceberg-data",
89
"catalogType": "hadoop",
910
"catalogName": "mycatalog",

‎iceberg-data-deduplication/data_compaction_package_test.json‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
{
2+
"version": "1",
3+
"enabled-engines": ["flink"],
4+
"script": {
5+
"main": "data_compaction.sqrl",
6+
"config": {
7+
"env": "test",
8+
"warehouse": "s3://sqrl-examples-data-bucket/datasqrl-examples/iceberg-canary",
9+
"catalogType": "glue",
10+
"catalogName": "canary",
11+
"databaseName": "canary",
12+
"tableName": "iceberg_dedup_test",
13+
"distinctKeyCol": "account_id",
14+
"partitionCol": "account_id",
15+
"partitionColType": "bigint",
16+
"timeBucketCol": "time_bucket",
17+
"timestampCol": "ts",
18+
"deleteFlagCol": null,
19+
"sortKey": null,
20+
"bufferSeconds": 2,
21+
"bucketSeconds": 5,
22+
"newDataNormalizerInMB": 1,
23+
"maxTotalPartitionSizeInMB": 10
24+
}
25+
},
26+
"engines": {
27+
"flink": {
28+
"config": {
29+
"execution.runtime-mode": "batch",
30+
"table.exec.source.idle-timeout": "1 ms"
31+
}
32+
}
33+
}
34+
}

‎iceberg-data-deduplication/data_delete.sqrl‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
IMPORT tables.compaction_tracker.*;
1+
IMPORT tables.tracker_{{env}}.*;
22
IMPORT udfs.delete_duplicated_data;
33

44
_DataToDelete :=

iceberg-data-deduplication/data_delete_package.json renamed to iceberg-data-deduplication/data_delete_package_local.json

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
"script": {
55
"main": "data_delete.sqrl",
66
"config": {
7+
"env": "local",
78
"warehouse": "sqrl-iceberg-data",
89
"catalogType": "hadoop",
910
"catalogName": "mycatalog",

‎iceberg-data-deduplication/data_delete_package_test.json‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
{
2+
"version": "1",
3+
"enabled-engines": ["flink"],
4+
"script": {
5+
"main": "data_delete.sqrl",
6+
"config": {
7+
"env": "test",
8+
"warehouse": "s3://sqrl-examples-data-bucket/datasqrl-examples/iceberg-canary",
9+
"catalogType": "glue",
10+
"catalogName": "canary",
11+
"databaseName": "canary",
12+
"commitWaitSec": 1,
13+
"stopDeleteFilter": ""
14+
}
15+
},
16+
"engines": {
17+
"flink": {
18+
"config": {
19+
"execution.runtime-mode": "batch"
20+
}
21+
}
22+
}
23+
}

‎iceberg-data-deduplication/datagen.sqrl‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
IMPORT tables.min_source AS _CDC;
2-
IMPORT tables.{{tableName}}.*;
2+
IMPORT tables.data_{{env}}.*;
33

44
CountRecords := SELECT COUNT(*) AS num_records FROM _CDC.SrcRecords;
55

66
EXPORT CountRecords TO logger.records;
77

8-
INSERT INTO {{tableName}} SELECT c.*, EXTRACT(SECOND FROM c.ts) + 1 AS time_bucket FROM _CDC.SrcRecords c;
8+
INSERT INTO DedupData SELECT c.*, EXTRACT(SECOND FROM c.ts) + 1 AS time_bucket FROM _CDC.SrcRecords c;

iceberg-data-deduplication/datagen_package.json renamed to iceberg-data-deduplication/datagen_package_local.json

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
"script": {
55
"main": "datagen.sqrl",
66
"config": {
7+
"env": "local",
78
"warehouse": "sqrl-iceberg-data",
89
"catalogType": "hadoop",
910
"catalogName": "mycatalog",

‎iceberg-data-deduplication/datagen_package_test.json‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
{
2+
"version": "1",
3+
"enabled-engines": ["flink"],
4+
"script": {
5+
"main": "datagen.sqrl",
6+
"config": {
7+
"env": "test",
8+
"warehouse": "s3://sqrl-examples-data-bucket/datasqrl-examples/iceberg-canary",
9+
"catalogType": "glue",
10+
"catalogName": "canary",
11+
"databaseName": "canary",
12+
"tableName": "iceberg_dedup_test"
13+
}
14+
},
15+
"engines": {
16+
"flink": {
17+
"config": {
18+
"execution.runtime-mode": "batch",
19+
"execution.checkpointing.interval": "8 s",
20+
"table.exec.source.idle-timeout": "1 ms"
21+
}
22+
}
23+
}
24+
}

0 commit comments

Comments
 (0)