Arquitectura Medallion en Google Cloud: Bronze, Silver y Gold con GCS y BigQuery

CREATE OR REPLACE PROCEDURE `empresa_lakehouse.silver_layer.sp_merge_customer_events`(
    IN p_processing_date DATE
)
BEGIN
  -- 1. Declaración de variables de control
  DECLARE v_records_updated INT64 DEFAULT 0;

  -- 2. Ejecución del MERGE incremental idempotente
  MERGE INTO `empresa_lakehouse.silver_layer.dim_customers` AS target
  USING (
    WITH deduplicated_bronze AS (
      SELECT
        customer_id,
        email,
        country_code,
        account_status,
        updated_at,
        ingestion_timestamp,
        -- Deduplicación determinista: última versión por clave de negocio
        ROW_NUMBER() OVER (
          PARTITION BY customer_id 
          ORDER BY updated_at DESC, ingestion_timestamp DESC
        ) AS row_rank
      FROM `empresa_lakehouse.bronze_layer.ext_gcs_raw_customer_events`
      -- Pruning estricto en la tabla BigLake / Externa
      WHERE _PARTITIONDATE = p_processing_date
    )
    SELECT
      customer_id,
      email,
      country_code,
      account_status,
      updated_at,
      p_processing_date AS partition_date
    FROM deduplicated_bronze
    WHERE row_rank = 1
  ) AS source
  ON target.partition_date = source.partition_date
     AND target.customer_id = source.customer_id

  -- 3. Actualización de registros mutados
  WHEN MATCHED AND source.updated_at > target.updated_at THEN
    UPDATE SET
      target.email = source.email,
      target.country_code = source.country_code,
      target.account_status = source.account_status,
      target.updated_at = source.updated_at,
      target.last_modified_timestamp = CURRENT_TIMESTAMP()

  -- 4. Inserción de nuevas entidades
  WHEN NOT MATCHED THEN
    INSERT (
      customer_id,
      email,
      country_code,
      account_status,
      updated_at,
      partition_date,
      created_timestamp,
      last_modified_timestamp
    )
    VALUES (
      source.customer_id,
      source.email,
      source.country_code,
      source.account_status,
      source.updated_at,
      source.partition_date,
      CURRENT_TIMESTAMP(),
      CURRENT_TIMESTAMP()
    );

END;