feat(git): harden Bitbucket Cloud connector reliability

src/ingestion/connectors/git/bitbucket-cloud/.dockerignore

@@ -0,0 +1,7 @@
+.venv
+.pytest_cache
+tests
+__pycache__
+*.pyc
+*.egg-info
+dbt

src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__bronze_promoted.sql

@@ -15,10 +15,23 @@
 
 {% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.branches',              order_by='unique_key') %}
 {% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.commits',               order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.commit_branch_reachability', order_by='unique_key') %}
 {% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.file_changes',          order_by='unique_key') %}
 {% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.pull_requests',         order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.pull_request_diffstat', order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.pull_request_activity', order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.pull_request_tasks',    order_by='unique_key') %}
 {% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.pull_request_comments', order_by='unique_key') %}
 {% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.pull_request_commits',  order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.pipelines',             order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.pipeline_steps',         order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.pipeline_step_test_reports', order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.deployments',            order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.environments',           order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.tags',                   order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.issues',                 order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.issue_comments',         order_by='unique_key') %}
+{% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.issue_changes',          order_by='unique_key') %}
 {% do promote_bronze_to_rmt(table='bronze_bitbucket_cloud.repositories',          order_by='unique_key') %}
 
 SELECT 1 AS promoted

src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__commit_branch_reachability.sql

@@ -0,0 +1,31 @@
+{% set _bronze_promoted = ref('bitbucket_cloud__bronze_promoted') %}
+{{ config(
+    materialized='incremental',
+    unique_key='unique_key',
+    order_by=['unique_key'],
+    settings={'allow_nullable_key': 1},
+    schema='staging',
+    tags=['bitbucket-cloud']
+) }}
+
+SELECT
+    tenant_id,
+    source_id,
+    entity_key AS unique_key,
+    COALESCE(workspace, '') AS project_key,
+    COALESCE(repo_slug, '') AS repo_slug,
+    repository_uuid,
+    COALESCE(branch_name, '') AS branch_name,
+    branch_head_sha,
+    default_branch_name,
+    commit_sha,
+    parseDateTimeBestEffortOrNull(committed_at) AS committed_at,
+    COALESCE(reachability_action, '') AS reachability_action,
+    'insight_bitbucket_cloud' AS data_source,
+    toUnixTimestamp64Milli(now64()) AS _version,
+    _airbyte_extracted_at
+FROM {{ source('bronze_bitbucket_cloud', 'commit_branch_reachability') }} FINAL
+WHERE record_type = 'item'
+{% if is_incremental() %}
+AND _airbyte_extracted_at > (SELECT max(_airbyte_extracted_at) FROM {{ this }})
+{% endif %}

src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__commits.sql

@@ -8,46 +8,95 @@
     tags=['bitbucket-cloud', 'silver:class_git_commits']
 ) }}
 
+WITH file_change_generations AS (
+    SELECT
+        tenant_id,
+        source_id,
+        repository_uuid,
+        sha,
+        generation_id,
+        countIf(record_type = 'item') AS observed_count,
+        maxIf(snapshot_item_count, record_type = 'snapshot_complete') AS expected_count,
+        maxIf(_airbyte_extracted_at, record_type = 'snapshot_complete') AS completed_at,
+        countIf(record_type = 'snapshot_complete' AND snapshot_available) AS completion_count
+    FROM {{ source('bronze_bitbucket_cloud', 'file_changes') }} FINAL
+    GROUP BY tenant_id, source_id, repository_uuid, sha, generation_id
+    HAVING completion_count > 0 AND observed_count = expected_count
+),
+latest_file_change_generation AS (
+    SELECT
+        tenant_id,
+        source_id,
+        repository_uuid,
+        sha,
+        argMax(generation_id, completed_at) AS generation_id,
+        max(completed_at) AS completed_at
+    FROM file_change_generations
+    GROUP BY tenant_id, source_id, repository_uuid, sha
+),
+file_changes AS (
+    SELECT
+        change.tenant_id,
+        change.source_id,
+        change.repository_uuid,
+        change.sha,
+        count() AS files_changed,
+        if(countIf(change.additions IS NULL OR change.deletions IS NULL) > 0, NULL, sum(change.additions)) AS lines_added,
+        if(countIf(change.additions IS NULL OR change.deletions IS NULL) > 0, NULL, sum(change.deletions)) AS lines_removed,
+        max(latest.completed_at) AS completed_at
+    FROM {{ source('bronze_bitbucket_cloud', 'file_changes') }} AS change FINAL
+    INNER JOIN latest_file_change_generation AS latest
+        USING (tenant_id, source_id, repository_uuid, sha, generation_id)
+    WHERE change.record_type = 'item'
+    GROUP BY change.tenant_id, change.source_id, change.repository_uuid, change.sha
+),
+empty_file_changes AS (
+    SELECT
+        latest.tenant_id,
+        latest.source_id,
+        latest.repository_uuid,
+        latest.sha,
+        0 AS files_changed,
+        CAST(0, 'Nullable(Int64)') AS lines_added,
+        CAST(0, 'Nullable(Int64)') AS lines_removed,
+        latest.completed_at
+    FROM latest_file_change_generation AS latest
+    LEFT ANTI JOIN file_changes AS change
+        USING (tenant_id, source_id, repository_uuid, sha)
+),
+complete_file_changes AS (
+    SELECT * FROM file_changes
+    UNION ALL
+    SELECT * FROM empty_file_changes
+)
 SELECT
     c.tenant_id,
     c.source_id,
-    c.unique_key,
+    c.entity_key AS unique_key,
     COALESCE(c.workspace, '') AS project_key,
     COALESCE(c.repo_slug, '') AS repo_slug,
     COALESCE(c.hash, '') AS commit_hash,
     COALESCE(c.branch_name, '') AS branch,
     COALESCE(c.author_name, '') AS author_name,
     COALESCE(c.author_email, '') AS author_email,
-    '' AS committer_name,
-    '' AS committer_email,
+    COALESCE(c.committer_name, '') AS committer_name,
+    COALESCE(c.committer_email, '') AS committer_email,
     COALESCE(c.message, '') AS message,
     parseDateTimeBestEffortOrNull(c.date) AS date,
-    COALESCE(fc.files_changed, 0) AS files_changed,
-    COALESCE(fc.lines_added, 0) AS lines_added,
-    COALESCE(fc.lines_removed, 0) AS lines_removed,
+    CAST(fc.files_changed, 'Nullable(Int64)') AS files_changed,
+    CAST(fc.lines_added, 'Nullable(Int64)') AS lines_added,
+    CAST(fc.lines_removed, 'Nullable(Int64)') AS lines_removed,
     if(JSONLength(COALESCE(toString(c.parent_hashes), '[]')) > 1, 1, 0) AS is_merge_commit,
     'insight_bitbucket_cloud' AS data_source,
     toUnixTimestamp64Milli(now64()) AS _version,
-    c._airbyte_extracted_at
+    greatest(c._airbyte_extracted_at, COALESCE(fc.completed_at, c._airbyte_extracted_at)) AS _airbyte_extracted_at
 FROM {{ source('bronze_bitbucket_cloud', 'commits') }} AS c FINAL
-LEFT JOIN (
-    SELECT
-        tenant_id,
-        workspace,
-        repo_slug,
-        sha,
-        count() AS files_changed,
-        SUM(COALESCE(additions, 0)) AS lines_added,
-        SUM(COALESCE(deletions, 0)) AS lines_removed
-    -- FINAL: dedup file_changes before count()/SUM, else bronze dupes inflate
-    -- files_changed / lines_added / lines_removed (baked into one row, RMT
-    -- downstream cannot undo it). See ADR-0001.
-    FROM {{ source('bronze_bitbucket_cloud', 'file_changes') }} FINAL
-    GROUP BY tenant_id, workspace, repo_slug, sha
-) AS fc ON fc.sha = c.hash
+LEFT JOIN complete_file_changes AS fc ON fc.sha = c.hash
     AND fc.tenant_id = c.tenant_id
-    AND fc.workspace = c.workspace
-    AND fc.repo_slug = c.repo_slug
+    AND fc.source_id = c.source_id
+    AND fc.repository_uuid = c.repository_uuid
+WHERE c.record_type = 'item'
 {% if is_incremental() %}
-WHERE c._airbyte_extracted_at > (SELECT max(_airbyte_extracted_at) FROM {{ this }})
+AND greatest(c._airbyte_extracted_at, COALESCE(fc.completed_at, c._airbyte_extracted_at))
+    > (SELECT max(_airbyte_extracted_at) FROM {{ this }})
 {% endif %}

src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__file_changes.sql

@@ -8,14 +8,40 @@
     tags=['bitbucket-cloud', 'silver:class_git_file_changes']
 ) }}
 
+WITH generations AS (
+    SELECT
+        tenant_id,
+        source_id,
+        repository_uuid,
+        sha,
+        generation_id,
+        countIf(record_type = 'item') AS observed_count,
+        maxIf(snapshot_item_count, record_type = 'snapshot_complete') AS expected_count,
+        maxIf(_airbyte_extracted_at, record_type = 'snapshot_complete') AS completed_at,
+        countIf(record_type = 'snapshot_complete' AND snapshot_available) AS completion_count
+    FROM {{ source('bronze_bitbucket_cloud', 'file_changes') }} FINAL
+    GROUP BY tenant_id, source_id, repository_uuid, sha, generation_id
+    HAVING completion_count > 0 AND observed_count = expected_count
+),
+latest AS (
+    SELECT
+        tenant_id,
+        source_id,
+        repository_uuid,
+        sha,
+        argMax(generation_id, completed_at) AS generation_id,
+        max(completed_at) AS completed_at
+    FROM generations
+    GROUP BY tenant_id, source_id, repository_uuid, sha
+)
 SELECT
-    tenant_id,
-    source_id,
-    unique_key,
-    COALESCE(workspace, '') AS project_key,
-    COALESCE(repo_slug, '') AS repo_slug,
-    COALESCE(sha, '') AS commit_hash,
-    COALESCE(filename, '') AS file_path,
+    change.tenant_id,
+    change.source_id,
+    change.entity_key AS unique_key,
+    COALESCE(change.workspace, '') AS project_key,
+    COALESCE(change.repo_slug, '') AS repo_slug,
+    COALESCE(change.sha, '') AS commit_hash,
+    COALESCE(change.filename, '') AS file_path,
     -- File extension: last segment after the final '.', empty when none.
     -- Earlier shape (issue #494) used `position('.', filename) > 0` as the
     -- guard — but ClickHouse `position` is function-style
@@ -27,18 +53,20 @@ SELECT
     -- `Makefile` (where the position-based guard would also fire 0 by
     -- accident, but the array-length guard is the explicit predicate).
     if(
-        length(splitByChar('.', COALESCE(filename, ''))) > 1,
-        arrayElement(splitByChar('.', COALESCE(filename, '')), -1),
+        length(splitByChar('.', COALESCE(change.filename, ''))) > 1,
+        arrayElement(splitByChar('.', COALESCE(change.filename, '')), -1),
         ''
     ) AS file_extension,
-    COALESCE(status, '') AS change_type,
-    COALESCE(additions, 0) AS lines_added,
-    COALESCE(deletions, 0) AS lines_removed,
-    COALESCE(source_type, '') AS source_type,
+    COALESCE(change.status, '') AS change_type,
+    change.additions AS lines_added,
+    change.deletions AS lines_removed,
+    COALESCE(change.source_type, '') AS source_type,
     'insight_bitbucket_cloud' AS data_source,
     toUnixTimestamp64Milli(now64()) AS _version,
-    _airbyte_extracted_at
-FROM {{ source('bronze_bitbucket_cloud', 'file_changes') }}
+    latest.completed_at AS _airbyte_extracted_at
+FROM {{ source('bronze_bitbucket_cloud', 'file_changes') }} AS change FINAL
+INNER JOIN latest USING (tenant_id, source_id, repository_uuid, sha, generation_id)
+WHERE change.record_type = 'item'
 {% if is_incremental() %}
-WHERE _airbyte_extracted_at > (SELECT max(_airbyte_extracted_at) FROM {{ this }})
+AND latest.completed_at > (SELECT max(_airbyte_extracted_at) FROM {{ this }})
 {% endif %}