Data Pipeline Scalability and Performance Questions

CREATE TABLE analytics.etl_state (
  source_id text PRIMARY KEY,
  last_ingested_at timestamptz
);

WITH hwm AS (
  SELECT COALESCE(last_ingested_at, '1970-01-01'::timestamptz) AS last_ingested_at
  FROM analytics.etl_state WHERE source_id = 'source_events'
),
candidates AS (
  SELECT s.*
  FROM source_events s, hwm
  WHERE s.updated_at > hwm.last_ingested_at
)
INSERT INTO analytics.fact_events (event_id, user_id, event_time, payload, last_ingested_at)
SELECT source_id, user_id, event_time, payload, now()
FROM candidates
ON CONFLICT (event_id) DO UPDATE
  SET user_id = EXCLUDED.user_id,
      event_time = EXCLUDED.event_time,
      payload = EXCLUDED.payload,
      last_ingested_at = EXCLUDED.last_ingested_at
  WHERE analytics.fact_events.last_ingested_at IS NULL
     OR analytics.fact_events.last_ingested_at < EXCLUDED.last_ingested_at;

MERGE INTO analytics.fact_events t
USING (
  SELECT s.*, CURRENT_TIMESTAMP() AS last_ingested_at
  FROM source_events s
  JOIN analytics.etl_state e ON e.source_id = 'source_events'
  WHERE s.updated_at > e.last_ingested_at
) s
ON t.event_id = s.source_id
WHEN MATCHED AND t.last_ingested_at < s.last_ingested_at THEN
  UPDATE SET user_id = s.user_id, event_time = s.event_time, payload = s.payload, last_ingested_at = s.last_ingested_at
WHEN NOT MATCHED THEN
  INSERT (event_id, user_id, event_time, payload, last_ingested_at)
  VALUES (s.source_id, s.user_id, s.event_time, s.payload, s.last_ingested_at);

UPDATE analytics.etl_state
SET last_ingested_at = (SELECT MAX(updated_at) FROM source_events WHERE updated_at > last_ingested_at)
WHERE source_id = 'source_events';