feat: add index for dedup copy pipe (CM-1000) (#3900)

Signed-off-by: Umberto Sgueglia <ulemons92@gmail.com>
Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>

Author: ulemons

services/libs/tinybird/datasources/activities.datasource

@@ -48,6 +48,9 @@ SCHEMA >
     `updatedById` UUID `json:$.updatedById` DEFAULT toUUID('00000000-0000-0000-0000-000000000000'),
     `segmentId` LowCardinality(String) `json:$.segmentId` DEFAULT ''
 
+INDEXES >
+    idx_updatedAt updatedAt TYPE minmax GRANULARITY 1
+
 ENGINE ReplacingMergeTree
 ENGINE_PARTITION_KEY toYear(createdAt)
 ENGINE_SORTING_KEY id

services/libs/tinybird/datasources/activities_deduplicated_ds.datasource

@@ -35,6 +35,7 @@ SCHEMA >
 INDEXES >
     idx_body_ngram3 body TYPE ngrambf_v1(3, 2048, 6, 0) GRANULARITY 64
     idx_title_ngram3 title TYPE ngrambf_v1(3, 512, 6, 0) GRANULARITY 64
+    idx_updatedAt updatedAt TYPE minmax GRANULARITY 1
 
 ENGINE MergeTree
 ENGINE_PARTITION_KEY toYear(timestamp)

services/libs/tinybird/pipes/activities_deduplicated_copy_pipe_append_mode.pipe

@@ -15,7 +15,12 @@ SQL >
         a.url,
         a.updatedAt
     FROM activities a
-    WHERE a.updatedAt > (SELECT max("updatedAt") FROM activities_deduplicated_ds)
+    WHERE
+        a.updatedAt > (
+            SELECT greatest(max(updatedAt), now() - INTERVAL 3 HOUR)
+            FROM activities_deduplicated_ds
+            WHERE updatedAt > now() - INTERVAL 3 HOUR
+        )
 
 TYPE COPY
 TARGET_DATASOURCE activities_deduplicated_ds