pull_request_analysis_baseline_merge_MV.pipe

DESCRIPTION >
    Compacts activities from same PR into one, keeping necessary information in a single row.
    Uses existing pull_requests_analyzed data as baseline and merges new events on top.

NODE snapshot_resolver
DESCRIPTION >
    Resolve latest snapshot ID once for reuse across all nodes.

SQL >
    SELECT max(snapshotId) as latestSnapshotId FROM activityRelations_enrich_snapshot_MV_ds

NODE new_pull_request_related_activity
SQL >
    SELECT
        activityId as id,
        sourceId,
        sourceParentId,
        channel,
        timestamp as ts,
        segmentId,
        gitChangedLinesBucket,
        memberId,
        organizationId,
        platform,
        updatedAt,
        type,
        pullRequestReviewState
    FROM activityRelations_enrich_snapshot_MV_ds
    WHERE
        snapshotId = (SELECT latestSnapshotId FROM snapshot_resolver)
        AND type IN (
            'pull_request-opened',
            'merge_request-opened',
            'changeset-created',
            'pull_request-assigned',
            'merge_request-assigned',
            'pull_request-review-requested',
            'merge_request-review-requested',
            'pull_request-reviewed',
            'merge_request-review-changes-requested',
            'patchset-created',
            'merge_request-review-approved',
            'patchset_approval-created',
            'pull_request-closed',
            'merge_request-closed',
            'changeset-closed',
            'changeset-abandoned',
            'pull_request-merged',
            'merge_request-merged',
            'changeset-merged'
        )

NODE new_events_aggregated
DESCRIPTION >
    Aggregates new events from the latest snapshot by PR source ID.

SQL >
    SELECT
        if(
            type IN (
                'pull_request-opened',
                'merge_request-opened',
                'changeset-created',
                'changeset-abandoned',
                'changeset-merged'
            ),
            sourceId,
            sourceParentId
        ) AS prSourceId,
        argMinIf(
            id, ts, type IN ('pull_request-opened', 'merge_request-opened', 'changeset-created')
        ) AS id,
        argMinIf(
            channel, ts, type IN ('pull_request-opened', 'merge_request-opened', 'changeset-created')
        ) AS channel,
        argMinIf(
            segmentId, ts, type IN ('pull_request-opened', 'merge_request-opened', 'changeset-created')
        ) AS segmentId,
        argMinIf(
            gitChangedLinesBucket,
            ts,
            type IN ('pull_request-opened', 'merge_request-opened', 'changeset-created')
        ) AS gitChangedLinesBucket,
        argMinIf(
            memberId, ts, type IN ('pull_request-opened', 'merge_request-opened', 'changeset-created')
        ) AS memberId,
        argMinIf(
            organizationId,
            ts,
            type IN ('pull_request-opened', 'merge_request-opened', 'changeset-created')
        ) AS organizationId,
        argMinIf(
            platform, ts, type IN ('pull_request-opened', 'merge_request-opened', 'changeset-created')
        ) AS platform,
        minIf(
            ts, type IN ('pull_request-opened', 'merge_request-opened', 'changeset-created')
        ) AS openedAt,
        argMinIf(
            updatedAt, ts, type IN ('pull_request-opened', 'merge_request-opened', 'changeset-created')
        ) AS openedUpdatedAt,
        toInt64(
            countIf(type = 'patchset-created' AND sourceParentId = prSourceId)
        ) AS numberOfPatchsets,
        argMin(
            updatedAt, if(type IN ('pull_request-assigned', 'merge_request-assigned'), ts, NULL)
        ) AS assignedUpdatedAt,
        min(if(type IN ('pull_request-assigned', 'merge_request-assigned'), ts, NULL)) AS assignedAt,
        argMin(
            updatedAt,
            if(type IN ('pull_request-review-requested', 'merge_request-review-requested'), ts, NULL)
        ) AS reviewRequestedUpdatedAt,
        min(
            if(type IN ('pull_request-review-requested', 'merge_request-review-requested'), ts, NULL)
        ) AS reviewRequestedAt,
        argMin(
            updatedAt,
            if(
                type IN (
                    'pull_request-reviewed',
                    'merge_request-review-changes-requested',
                    'merge_request-review-approved'
                )
                OR (
                    type = 'patchset_approval-created'
                    AND splitByChar('-', sourceParentId)[1] = prSourceId
                ),
                ts,
                NULL
            )
        ) AS reviewedUpdatedAt,
        min(
            if(
                type IN (
                    'pull_request-reviewed',
                    'merge_request-review-changes-requested',
                    'merge_request-review-approved'
                )
                OR (
                    type = 'patchset_approval-created'
                    AND splitByChar('-', sourceParentId)[1] = prSourceId
                ),
                ts,
                NULL
            )
        ) AS reviewedAt,
        argMin(
            updatedAt,
            if(
                (type = 'pull_request-reviewed' AND pullRequestReviewState = 'APPROVED')
                OR type = 'merge_request-review-approved'
                OR (
                    type = 'patchset_approval-created'
                    AND splitByChar('-', sourceParentId)[1] = prSourceId
                ),
                ts,
                NULL
            )
        ) AS approvedUpdatedAt,
        min(
            if(
                (type = 'pull_request-reviewed' AND pullRequestReviewState = 'APPROVED')
                OR type = 'merge_request-review-approved'
                OR (
                    type = 'patchset_approval-created'
                    AND splitByChar('-', sourceParentId)[1] = prSourceId
                ),
                ts,
                NULL
            )
        ) AS approvedAt,
        argMin(
            updatedAt,
            if(
                type IN (
                    'pull_request-closed',
                    'merge_request-closed',
                    'changeset-closed',
                    'changeset-abandoned'
                ),
                ts,
                NULL
            )
        ) AS closedUpdatedAt,
        min(
            if(
                type IN (
                    'pull_request-closed',
                    'merge_request-closed',
                    'changeset-closed',
                    'changeset-abandoned'
                ),
                ts,
                NULL
            )
        ) AS closedAt,
        argMin(
            updatedAt,
            if(type IN ('pull_request-merged', 'merge_request-merged', 'changeset-merged'), ts, NULL)
        ) AS mergedUpdatedAt,
        min(
            if(type IN ('pull_request-merged', 'merge_request-merged', 'changeset-merged'), ts, NULL)
        ) AS mergedAt,
        argMin(
            updatedAt,
            if(
                type IN (
                    'pull_request-closed',
                    'pull_request-merged',
                    'merge_request-closed',
                    'merge_request-merged',
                    'changeset-merged',
                    'changeset-closed',
                    'changeset-abandoned'
                ),
                ts,
                NULL
            )
        ) AS resolvedUpdatedAt,
        min(
            if(
                type IN (
                    'pull_request-closed',
                    'pull_request-merged',
                    'merge_request-closed',
                    'merge_request-merged',
                    'changeset-merged',
                    'changeset-closed',
                    'changeset-abandoned'
                ),
                ts,
                NULL
            )
        ) AS resolvedAt
    FROM new_pull_request_related_activity
    GROUP BY prSourceId
    HAVING prSourceId != ''

NODE baseline_filtered
DESCRIPTION >
    Pre-filter baseline using semi-join pattern.
    This significantly reduces the data scanned in the final JOIN.

SQL >
    SELECT *
    FROM pull_requests_analyzed
    WHERE sourceId IN (SELECT prSourceId FROM new_events_aggregated)

NODE pull_request_analysis_results_merged
DESCRIPTION >
    Merge new events with filtered baseline using ANY LEFT JOIN.
    New data overwrites existing data where present, preserves existing for unchanged fields.

SQL >
    SELECT
        if(existing.id != '', existing.id, new.id) as id,
        if(existing.sourceId != '', existing.sourceId, new.prSourceId) as sourceId,
        if(
            existing.openedAt != toDateTime(0), existing.openedAt, new.openedAt
        ) as openedAt,
        if(existing.segmentId != '', existing.segmentId, new.segmentId) as segmentId,
        if(existing.channel != '', existing.channel, new.channel) as channel,
        if(existing.memberId != '', existing.memberId, new.memberId) as memberId,
        if(
            existing.organizationId != '', existing.organizationId, new.organizationId
        ) as organizationId,
        if(
            existing.gitChangedLinesBucket != '',
            existing.gitChangedLinesBucket,
            new.gitChangedLinesBucket
        ) as gitChangedLinesBucket,
        COALESCE(
            if(
                new.assignedAt IS NOT NULL AND new.assignedAt != toDateTime(0),
                least(existing.assignedAt, new.assignedAt),
                existing.assignedAt
            ),
            new.assignedAt
        ) AS assignedAt,
        COALESCE(
            if(
                new.reviewRequestedAt IS NOT NULL AND new.reviewRequestedAt != toDateTime(0),
                least(existing.reviewRequestedAt, new.reviewRequestedAt),
                existing.reviewRequestedAt
            ),
            new.reviewRequestedAt
        ) AS reviewRequestedAt,
        COALESCE(
            if(
                new.reviewedAt IS NOT NULL AND new.reviewedAt != toDateTime(0),
                least(existing.reviewedAt, new.reviewedAt),
                existing.reviewedAt
            ),
            new.reviewedAt
        ) AS reviewedAt,
        COALESCE(
            if(
                new.approvedAt IS NOT NULL AND new.approvedAt != toDateTime(0),
                least(existing.approvedAt, new.approvedAt),
                existing.approvedAt
            ),
            new.approvedAt
        ) AS approvedAt,
        COALESCE(
            if(
                new.closedAt IS NOT NULL AND new.closedAt != toDateTime(0),
                least(existing.closedAt, new.closedAt),
                existing.closedAt
            ),
            new.closedAt
        ) AS closedAt,
        COALESCE(
            if(
                new.mergedAt IS NOT NULL AND new.mergedAt != toDateTime(0),
                least(existing.mergedAt, new.mergedAt),
                existing.mergedAt
            ),
            new.mergedAt
        ) AS mergedAt,
        COALESCE(
            if(
                new.resolvedAt IS NOT NULL AND new.resolvedAt != toDateTime(0),
                least(existing.resolvedAt, new.resolvedAt),
                existing.resolvedAt
            ),
            new.resolvedAt
        ) AS resolvedAt,
        IF(
            assignedAt IS NULL, NULL, toUnixTimestamp(assignedAt) - toUnixTimestamp(openedAt)
        ) AS assignedInSeconds,
        IF(
            reviewRequestedAt IS NULL,
            NULL,
            toUnixTimestamp(reviewRequestedAt) - toUnixTimestamp(openedAt)
        ) AS reviewRequestedInSeconds,
        IF(
            reviewedAt IS NULL, NULL, toUnixTimestamp(reviewedAt) - toUnixTimestamp(openedAt)
        ) AS reviewedInSeconds,
        IF(
            closedAt IS NULL, NULL, toUnixTimestamp(closedAt) - toUnixTimestamp(openedAt)
        ) AS closedInSeconds,
        IF(
            mergedAt IS NULL, NULL, toUnixTimestamp(mergedAt) - toUnixTimestamp(openedAt)
        ) AS mergedInSeconds,
        IF(
            resolvedAt IS NULL, NULL, toUnixTimestamp(resolvedAt) - toUnixTimestamp(openedAt)
        ) AS resolvedInSeconds,
        if(existing.platform != '', existing.platform, new.platform) as platform,
        COALESCE(
            if(new.numberOfPatchsets > 0, new.numberOfPatchsets, NULL), existing.numberOfPatchsets
        ) as numberOfPatchsets,
        toStartOfInterval(
            greatest(
                COALESCE(new.openedUpdatedAt, existing.snapshotId, toDateTime(0)),
                COALESCE(new.assignedUpdatedAt, toDateTime(0)),
                COALESCE(new.reviewRequestedUpdatedAt, toDateTime(0)),
                COALESCE(new.reviewedUpdatedAt, toDateTime(0)),
                COALESCE(new.approvedUpdatedAt, toDateTime(0)),
                COALESCE(new.closedUpdatedAt, toDateTime(0)),
                COALESCE(new.mergedUpdatedAt, toDateTime(0)),
                COALESCE(new.resolvedUpdatedAt, toDateTime(0))
            ),
            INTERVAL 1 hour
        )
        + INTERVAL 1 hour as snapshotId
    FROM new_events_aggregated new ANY
    LEFT JOIN baseline_filtered existing ON new.prSourceId = existing.sourceId
    WHERE if(existing.id != '', existing.id, new.id) != ''

TYPE MATERIALIZED
DATASOURCE pull_request_analyzed_MV_ds