10B

-- 10b: Accept latest valid rows
WITH ranked AS (
 SELECT r.*,ROW_NUMBER() OVER (PARTITION BY source_order_id ORDER BY loaded_at DESC,batch_row DESC) AS position
 FROM aabha_sql.raw_orders AS r
)
SELECT r.batch_row,r.source_order_id,lower(trim(r.status_text)) AS status,r.amount
FROM ranked AS r JOIN aabha_sql.customers AS c ON c.customer_id=r.customer_id
WHERE r.position=1 AND r.source_order_id IS NOT NULL AND r.amount>=0
 AND lower(trim(r.status_text)) IN ('completed','pending','cancelled')
 AND r.order_date_text IS NOT NULL AND pg_input_is_valid(r.order_date_text,'date')
ORDER BY r.source_order_id;