Ingest (batch): land every file once.

BEGIN;

-- 1. Type the rows of files not loaded before, drop exact duplicates, add load metadata.
--    A value that cannot be cast keeps its row out of bronze (it goes to a rejects table, not shown).
INSERT INTO bronze.orders (order_id, amount, ordered_at, _source_file, _loaded_at)
SELECT order_id, amount, ordered_at, source_file, CURRENT_TIMESTAMP
FROM (
  SELECT
    source_file,
    TRY_CAST(raw_order_id AS BIGINT)       AS order_id,
    TRY_CAST(raw_amount AS DECIMAL(18, 2)) AS amount,
    TRY_CAST(raw_ordered_at AS TIMESTAMP)  AS ordered_at,
    ROW_NUMBER() OVER (
      PARTITION BY raw_order_id, raw_amount, raw_ordered_at
      ORDER BY source_file
    ) AS copy_number
  FROM staged.orders_files
  WHERE source_file NOT IN (SELECT source_file FROM bronze.load_log)
) AS typed
WHERE copy_number = 1
  AND order_id IS NOT NULL AND amount IS NOT NULL AND ordered_at IS NOT NULL;

-- 2. Record the files in the same transaction, so the log and the rows never disagree.
INSERT INTO bronze.load_log (source_file, loaded_at)
SELECT DISTINCT source_file, CURRENT_TIMESTAMP
FROM staged.orders_files
WHERE source_file NOT IN (SELECT source_file FROM bronze.load_log);

COMMIT;