fix: cursor suggestions

Author: ulemons

services/libs/tinybird/pipes/cdp_segment_metrics_copy_pipe.pipe

@@ -1,20 +1,32 @@
 DESCRIPTION >
     Daily batch job that builds segment-level aggregate states from existing
     member and organization segment aggregate datasources.
-    Heavy DISTINCT logic lives here.
+    Activities are computed from activityRelations_enriched_deduplicated_ds using the latest snapshotId.
+    This version avoids CTEs and avoids *StateIf() combinators to keep AggregateFunction types consistent.
 
-NODE activitiesSegmentStates
+NODE activitiesAllTimeStates
 SQL >
-    -- Compute activity counters from event-level data (correct last-30-days behavior)
-    SELECT
-        segmentId,
-        countState() AS activitiesTotalState,
-        countStateIf(timestamp >= now() - INTERVAL 30 DAY) AS activitiesLast30DaysState
+    -- Compute total activities from event-level data (latest snapshot only).
+    -- Avoid countStateIf() to keep AggregateFunction(count) type consistent.
+    SELECT segmentId, countState() AS activitiesTotalState
+    FROM activityRelations_enriched_deduplicated_ds
+    WHERE
+        segmentId IS NOT NULL
+        AND segmentId != ''
+        AND snapshotId = (SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_ds)
+    GROUP BY segmentId
+
+NODE activitiesLast30States
+SQL >
+    -- Compute last-30-days activities from event-level data (latest snapshot only).
+    -- Use WHERE filtering instead of countStateIf() to avoid AggregateFunction(countIf) types.
+    SELECT segmentId, countState() AS activitiesLast30DaysState
     FROM activityRelations_enriched_deduplicated_ds
     WHERE
         segmentId IS NOT NULL
         AND segmentId != ''
-        AND snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds)
+        AND snapshotId = (SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_ds)
+        AND timestamp >= now() - INTERVAL 30 DAY
     GROUP BY segmentId
 
 NODE memberPerSegment
@@ -24,14 +36,21 @@ SQL >
     FROM cdp_member_segment_aggregates_ds
     GROUP BY segmentId, memberId
 
-NODE memberSegmentStates
+NODE memberAllTimeStates
 SQL >
-    -- Build segment-level activity + member DISTINCT states
-    SELECT
-        segmentId,
-        uniqCombinedState(memberId) AS membersUniqState,
-        uniqCombinedStateIf(memberId, lastActive >= now() - INTERVAL 30 DAY) AS membersLast30UniqState
+    -- Build segment-level DISTINCT member states (all time).
+    -- Avoid uniqCombinedStateIf() to keep AggregateFunction(uniqCombined, String) type consistent.
+    SELECT segmentId, uniqCombinedState(memberId) AS membersUniqState
+    FROM memberPerSegment
+    GROUP BY segmentId
+
+NODE memberLast30States
+SQL >
+    -- Build segment-level DISTINCT member states (last 30 days).
+    -- Use WHERE filtering instead of uniqCombinedStateIf() to avoid uniqCombinedIf types.
+    SELECT segmentId, uniqCombinedState(memberId) AS membersLast30UniqState
     FROM memberPerSegment
+    WHERE lastActive >= now() - INTERVAL 30 DAY
     GROUP BY segmentId
 
 NODE orgPerSegment
@@ -41,36 +60,55 @@ SQL >
     FROM cdp_organization_segment_aggregates_ds
     GROUP BY segmentId, organizationId
 
-NODE orgSegmentStates
+NODE orgAllTimeStates
 SQL >
-    -- Build segment-level organization DISTINCT states
-    SELECT
-        segmentId,
-        uniqCombinedState(organizationId) AS orgsUniqState,
-        uniqCombinedStateIf(
-            organizationId, lastActive >= now() - INTERVAL 30 DAY
-        ) AS orgsLast30UniqState
+    -- Build segment-level DISTINCT organization states (all time).
+    SELECT segmentId, uniqCombinedState(organizationId) AS orgsUniqState
     FROM orgPerSegment
     GROUP BY segmentId
 
+NODE orgLast30States
+SQL >
+    -- Build segment-level DISTINCT organization states (last 30 days).
+    -- Use WHERE filtering instead of uniqCombinedStateIf() to avoid uniqCombinedIf types.
+    SELECT segmentId, uniqCombinedState(organizationId) AS orgsLast30UniqState
+    FROM orgPerSegment
+    WHERE lastActive >= now() - INTERVAL 30 DAY
+    GROUP BY segmentId
+
 NODE segmentAggStates
 SQL >
     -- Attach hierarchy and snapshot date.
-    -- Coalesce NULL AggregateFunction states to empty states for segments without activity.
+    -- Coalesce NULL AggregateFunction states to empty states without using *StateIf combinators.
+    -- Empty states are generated by aggregating over an empty input (system.one WHERE 0).
     SELECT
         toDate(now()) AS snapshotDate,
         s.id AS segmentId,
         s.parentId AS parentId,
         s.grandparentId AS grandparentId,
-        -- Activities (event-level truth)
-        ifNull(a.activitiesTotalState, countStateIf(0)) AS activitiesTotalState,
-        ifNull(a.activitiesLast30DaysState, countStateIf(0)) AS activitiesLast30DaysState,
-        -- Members DISTINCT states
-        ifNull(ms.membersUniqState, uniqCombinedStateIf('x', 0)) AS membersUniqState,
-        ifNull(ms.membersLast30UniqState, uniqCombinedStateIf('x', 0)) AS membersLast30UniqState,
-        -- Organizations DISTINCT states
-        ifNull(os.orgsUniqState, uniqCombinedStateIf('x', 0)) AS orgsUniqState,
-        ifNull(os.orgsLast30UniqState, uniqCombinedStateIf('x', 0)) AS orgsLast30UniqState
+        -- Activities (AggregateFunction(count))
+        ifNull(
+            a.activitiesTotalState, (SELECT countState() FROM system.one WHERE 0)
+        ) AS activitiesTotalState,
+        ifNull(
+            al30.activitiesLast30DaysState, (SELECT countState() FROM system.one WHERE 0)
+        ) AS activitiesLast30DaysState,
+        -- Members (AggregateFunction(uniqCombined, String))
+        ifNull(
+            m.membersUniqState, (SELECT uniqCombinedState(CAST('' AS String)) FROM system.one WHERE 0)
+        ) AS membersUniqState,
+        ifNull(
+            ml30.membersLast30UniqState,
+            (SELECT uniqCombinedState(CAST('' AS String)) FROM system.one WHERE 0)
+        ) AS membersLast30UniqState,
+        -- Organizations (AggregateFunction(uniqCombined, String))
+        ifNull(
+            o.orgsUniqState, (SELECT uniqCombinedState(CAST('' AS String)) FROM system.one WHERE 0)
+        ) AS orgsUniqState,
+        ifNull(
+            ol30.orgsLast30UniqState,
+            (SELECT uniqCombinedState(CAST('' AS String)) FROM system.one WHERE 0)
+        ) AS orgsLast30UniqState
     FROM
         (
             SELECT id, parentId, grandparentId
@@ -83,9 +121,12 @@ SQL >
                 AND parentId != ''
                 AND grandparentId != ''
         ) AS s
-    LEFT JOIN activitiesSegmentStates AS a ON a.segmentId = s.id
-    LEFT JOIN memberSegmentStates AS ms ON ms.segmentId = s.id
-    LEFT JOIN orgSegmentStates AS os ON os.segmentId = s.id
+    LEFT JOIN activitiesAllTimeStates AS a ON a.segmentId = s.id
+    LEFT JOIN activitiesLast30States AS al30 ON al30.segmentId = s.id
+    LEFT JOIN memberAllTimeStates AS m ON m.segmentId = s.id
+    LEFT JOIN memberLast30States AS ml30 ON ml30.segmentId = s.id
+    LEFT JOIN orgAllTimeStates AS o ON o.segmentId = s.id
+    LEFT JOIN orgLast30States AS ol30 ON ol30.segmentId = s.id
 
 TYPE COPY
 TARGET_DATASOURCE cdp_segment_metrics_ds