From 03ab90b42d970e8cf19a19e3d5aa74500748a192 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C5=A1per=20Grom?= Date: Mon, 17 Aug 2026 09:03:34 +0100 Subject: [PATCH 1/5] feat: route collection-scoped activity queries to single bucket (IN-1231) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Gašper Grom --- .../tinybird/pipes/activities_filtered.pipe | 10 ++++++- ...tyRelations_collection_bucket_routing.pipe | 26 +++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) create mode 100644 services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe diff --git a/services/libs/tinybird/pipes/activities_filtered.pipe b/services/libs/tinybird/pipes/activities_filtered.pipe index 53344642b7..45e9017f91 100644 --- a/services/libs/tinybird/pipes/activities_filtered.pipe +++ b/services/libs/tinybird/pipes/activities_filtered.pipe @@ -16,6 +16,8 @@ DESCRIPTION > - `activity_types`: Optional array of activity types (e.g., ['authored-commit', 'co-authored-commit']) - `includeCodeContributions`: Optional boolean to include code contribution activities. Defaults to 1. Set to 0 to exclude. Inherited from activityTypes_filtered. - `includeCollaborations`: Optional boolean to include or exclude collaboration activities. Inherited from activityTypes_filtered. + - `bucketId`: Optional pre-resolved collection bucket id (from `collection_buckets.pipe`). When given alongside + `collectionSlug`, routes directly to the single matching bucket table instead of scanning the 10-way union. - Response: `id` (activityId), `timestamp`, `type`, `platform`, `memberId`, `organizationId`, `segmentId`. - This pipe is consumed by many of downstream pipes and widgets across the platform for consistent activity filtering. - Performance is optimized through proper sorting keys on `segmentId`, `timestamp`, `type`, `platform`, and `memberId` in the source datasource. @@ -27,12 +29,18 @@ SQL > % SELECT activityId as id, timestamp, type, platform, memberId, organizationId, segmentId FROM - {% if defined(collectionSlug) %} activityRelations_collection_bucket_union + {% if defined(collectionSlug) and defined(bucketId) %} + activityRelations_collection_bucket_routing + {% elif defined(collectionSlug) %} activityRelations_collection_bucket_union {% else %} activityRelations_bucket_routing {% end %} as a where {% if defined(collectionSlug) %} segmentId IN (SELECT segmentId FROM segments_filtered_by_collection) + {% if defined(bucketId) %} + AND a.collectionSlug + = {{ String(collectionSlug, description="Collection slug for bucket-scoped reads", required=False) }} + {% end %} {% else %} segmentId = (SELECT segmentId FROM segments_filtered) {% end %} {% if defined(startDate) %} diff --git a/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe b/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe new file mode 100644 index 0000000000..6a8356d582 --- /dev/null +++ b/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe @@ -0,0 +1,26 @@ +DESCRIPTION > + - Routes a collection-scoped activity query directly to its single bucket table, mirroring + `activityRelations_bucket_routing.pipe`'s role for projects. + - Requires `bucketId` to already be resolved by the caller (via `collection_buckets.pipe`) + and passed in - this pipe only does the Jinja table-name branch, it does not compute the hash. + - Callers must still filter on `collectionSlug`, since a bucket can hold multiple curated + collections' activities. + +NODE activityRelations_collection_bucket_routing_2 +SQL > + % + SELECT selected_bucket.* + FROM + {% if bucketId == '0' %} activityRelations_collection_deduplicated_cleaned_bucket_0_ds + {% elif bucketId == '1' %} activityRelations_collection_deduplicated_cleaned_bucket_1_ds + {% elif bucketId == '2' %} activityRelations_collection_deduplicated_cleaned_bucket_2_ds + {% elif bucketId == '3' %} activityRelations_collection_deduplicated_cleaned_bucket_3_ds + {% elif bucketId == '4' %} activityRelations_collection_deduplicated_cleaned_bucket_4_ds + {% elif bucketId == '5' %} activityRelations_collection_deduplicated_cleaned_bucket_5_ds + {% elif bucketId == '6' %} activityRelations_collection_deduplicated_cleaned_bucket_6_ds + {% elif bucketId == '7' %} activityRelations_collection_deduplicated_cleaned_bucket_7_ds + {% elif bucketId == '8' %} activityRelations_collection_deduplicated_cleaned_bucket_8_ds + {% elif bucketId == '9' %} activityRelations_collection_deduplicated_cleaned_bucket_9_ds + -- fallback, should never happen + {% else %} activityRelations_collection_deduplicated_cleaned_bucket_0_ds + {% end %} as selected_bucket From a0fe5c1321cd7b4f0dacba18e9f0b324a3453c4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C5=A1per=20Grom?= Date: Mon, 17 Aug 2026 09:15:34 +0100 Subject: [PATCH 2/5] refactor: minimize collection bucket routing pipes (IN-1231) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Gašper Grom --- services/libs/tinybird/pipes/activities_filtered.pipe | 3 +-- .../activityRelations_collection_bucket_routing.pipe | 8 -------- 2 files changed, 1 insertion(+), 10 deletions(-) diff --git a/services/libs/tinybird/pipes/activities_filtered.pipe b/services/libs/tinybird/pipes/activities_filtered.pipe index 45e9017f91..5ce113599e 100644 --- a/services/libs/tinybird/pipes/activities_filtered.pipe +++ b/services/libs/tinybird/pipes/activities_filtered.pipe @@ -38,8 +38,7 @@ SQL > {% if defined(collectionSlug) %} segmentId IN (SELECT segmentId FROM segments_filtered_by_collection) {% if defined(bucketId) %} - AND a.collectionSlug - = {{ String(collectionSlug, description="Collection slug for bucket-scoped reads", required=False) }} + AND a.collectionSlug = {{ String(collectionSlug) }} {% end %} {% else %} segmentId = (SELECT segmentId FROM segments_filtered) {% end %} diff --git a/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe b/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe index 6a8356d582..cf91304032 100644 --- a/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe +++ b/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe @@ -1,11 +1,3 @@ -DESCRIPTION > - - Routes a collection-scoped activity query directly to its single bucket table, mirroring - `activityRelations_bucket_routing.pipe`'s role for projects. - - Requires `bucketId` to already be resolved by the caller (via `collection_buckets.pipe`) - and passed in - this pipe only does the Jinja table-name branch, it does not compute the hash. - - Callers must still filter on `collectionSlug`, since a bucket can hold multiple curated - collections' activities. - NODE activityRelations_collection_bucket_routing_2 SQL > % From 7db18a273a92b93aa099951683f64f5b697d29ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C5=A1per=20Grom?= Date: Mon, 17 Aug 2026 10:16:59 +0100 Subject: [PATCH 3/5] fix: remove non-explanatory comment from bucket routing pipe (IN-1231) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Gašper Grom --- .../pipes/activityRelations_collection_bucket_routing.pipe | 1 - 1 file changed, 1 deletion(-) diff --git a/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe b/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe index cf91304032..ded80d61e4 100644 --- a/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe +++ b/services/libs/tinybird/pipes/activityRelations_collection_bucket_routing.pipe @@ -13,6 +13,5 @@ SQL > {% elif bucketId == '7' %} activityRelations_collection_deduplicated_cleaned_bucket_7_ds {% elif bucketId == '8' %} activityRelations_collection_deduplicated_cleaned_bucket_8_ds {% elif bucketId == '9' %} activityRelations_collection_deduplicated_cleaned_bucket_9_ds - -- fallback, should never happen {% else %} activityRelations_collection_deduplicated_cleaned_bucket_0_ds {% end %} as selected_bucket From 5a06cfb24c3dc390d30cc2bdfe1538c8f5f7181e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C5=A1per=20Grom?= Date: Mon, 17 Aug 2026 10:20:21 +0100 Subject: [PATCH 4/5] style: format tinybird pipes with tb fmt (IN-1231) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Gašper Grom --- services/libs/tinybird/pipes/activities_filtered.pipe | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/services/libs/tinybird/pipes/activities_filtered.pipe b/services/libs/tinybird/pipes/activities_filtered.pipe index 5ce113599e..0de624a0d3 100644 --- a/services/libs/tinybird/pipes/activities_filtered.pipe +++ b/services/libs/tinybird/pipes/activities_filtered.pipe @@ -37,9 +37,7 @@ SQL > where {% if defined(collectionSlug) %} segmentId IN (SELECT segmentId FROM segments_filtered_by_collection) - {% if defined(bucketId) %} - AND a.collectionSlug = {{ String(collectionSlug) }} - {% end %} + {% if defined(bucketId) %} AND a.collectionSlug = {{ String(collectionSlug) }} {% end %} {% else %} segmentId = (SELECT segmentId FROM segments_filtered) {% end %} {% if defined(startDate) %} From 0b6e3f0905009df11d26bc0fef913b01516486be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C5=A1per=20Grom?= Date: Mon, 17 Aug 2026 12:47:58 +0100 Subject: [PATCH 5/5] fix: drop slow union fallback in activities_filtered when bucketId missing (IN-1231) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Gašper Grom --- .../libs/tinybird/pipes/activities_filtered.pipe | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/services/libs/tinybird/pipes/activities_filtered.pipe b/services/libs/tinybird/pipes/activities_filtered.pipe index 0de624a0d3..af4e270e2a 100644 --- a/services/libs/tinybird/pipes/activities_filtered.pipe +++ b/services/libs/tinybird/pipes/activities_filtered.pipe @@ -16,8 +16,9 @@ DESCRIPTION > - `activity_types`: Optional array of activity types (e.g., ['authored-commit', 'co-authored-commit']) - `includeCodeContributions`: Optional boolean to include code contribution activities. Defaults to 1. Set to 0 to exclude. Inherited from activityTypes_filtered. - `includeCollaborations`: Optional boolean to include or exclude collaboration activities. Inherited from activityTypes_filtered. - - `bucketId`: Optional pre-resolved collection bucket id (from `collection_buckets.pipe`). When given alongside - `collectionSlug`, routes directly to the single matching bucket table instead of scanning the 10-way union. + - `bucketId`: Pre-resolved collection bucket id (from `collection_buckets.pipe`), required alongside `collectionSlug`. + Routes directly to the single matching bucket table. If `collectionSlug` is given without `bucketId`, the pipe + returns no rows rather than falling back to a 10-way union scan — callers must resolve `bucketId` first. - Response: `id` (activityId), `timestamp`, `type`, `platform`, `memberId`, `organizationId`, `segmentId`. - This pipe is consumed by many of downstream pipes and widgets across the platform for consistent activity filtering. - Performance is optimized through proper sorting keys on `segmentId`, `timestamp`, `type`, `platform`, and `memberId` in the source datasource. @@ -29,15 +30,16 @@ SQL > % SELECT activityId as id, timestamp, type, platform, memberId, organizationId, segmentId FROM - {% if defined(collectionSlug) and defined(bucketId) %} - activityRelations_collection_bucket_routing - {% elif defined(collectionSlug) %} activityRelations_collection_bucket_union + {% if defined(collectionSlug) %} activityRelations_collection_bucket_routing {% else %} activityRelations_bucket_routing {% end %} as a where {% if defined(collectionSlug) %} - segmentId IN (SELECT segmentId FROM segments_filtered_by_collection) - {% if defined(bucketId) %} AND a.collectionSlug = {{ String(collectionSlug) }} {% end %} + {% if defined(bucketId) %} + segmentId IN (SELECT segmentId FROM segments_filtered_by_collection) + AND a.collectionSlug = {{ String(collectionSlug) }} + {% else %} 1 = 0 + {% end %} {% else %} segmentId = (SELECT segmentId FROM segments_filtered) {% end %} {% if defined(startDate) %}