diff --git a/scripts/migrate_tables.py b/scripts/migrate_tables.py index ee135b8..5f912b3 100644 --- a/scripts/migrate_tables.py +++ b/scripts/migrate_tables.py @@ -33,7 +33,9 @@ Per-table modes (encode the plan's dual-delivery-safe semantics): Take an on-demand backup of the dest table and wait for it to be AVAILABLE before running (the script refuses without a fresh --backup-arn - of THIS table). + of THIS table, on DRY-RUN as well as --execute: + the rehearsal validates the same preconditions + the real run will). Usage: # copy (dry-run first, then --execute) -- run tables INDIVIDUALLY, in order @@ -94,6 +96,21 @@ def session_table(profile, expected_account, table_name): return session.resource("dynamodb").Table(table_name) +def check_key_schema(table, name, keys): + """Abort with a clear message if the configured keys drift from the live + table's key schema. Every DynamoDB item necessarily carries its table's + key attributes, so once this passes the `item[k]` key projections below + cannot KeyError; a mismatch here is the only way they could. Set (not + ordered) comparison is deliberate: keys are only ever used to build dict + Key= projections, which are order-insensitive.""" + live = {k["AttributeName"] for k in table.key_schema} + if live != set(keys): + sys.exit( + f"ERROR: {name} live key schema {sorted(live)} != configured " + f"{sorted(keys)}; refusing to build keys from a stale TABLES entry." + ) + + def scan_items(table, filter_kwargs=None): kwargs = dict(filter_kwargs or {}) while True: @@ -197,8 +214,9 @@ def _validate_truncate_backup(name, dst, args): def _truncate_destination(name, dst, args, keys): - if args.execute: - _validate_truncate_backup(name, dst, args) + # Validate in dry-run too: a stale/missing --backup-arn should surface on + # the rehearsal run, not first appear when the operator adds --execute. + _validate_truncate_backup(name, dst, args) dest_keys = [{k: item[k] for k in keys} for item in scan_items(dst)] print(f"[{name}] truncate: {len(dest_keys)} destination rows to delete") if args.execute: @@ -213,6 +231,8 @@ def _truncate_destination(name, dst, args, keys): def copy_table(name, cfg, args): src = session_table(args.source_profile, SOURCE_ACCOUNT, name) dst = session_table(args.dest_profile, DEST_ACCOUNT, name) + check_key_schema(src, name, cfg["keys"]) + check_key_schema(dst, name, cfg["keys"]) filter_kwargs = None if cfg["mode"] == "cutoff": @@ -259,6 +279,8 @@ def copy_table(name, cfg, args): def verify_table(name, cfg, args): src = session_table(args.source_profile, SOURCE_ACCOUNT, name) dst = session_table(args.dest_profile, DEST_ACCOUNT, name) + check_key_schema(src, name, cfg["keys"]) + check_key_schema(dst, name, cfg["keys"]) src_filter = None if cfg["mode"] == "cutoff": @@ -279,9 +301,12 @@ def verify_table(name, cfg, args): print(f"[{name}] source={src_count} dest={dst_count} -> {status}") ok = relation - # Spot-check every mode. For cutoff tables the source scan is cutoff-filtered - # (so no dual-window rows are sampled); for truncate_load keys are copied - # verbatim, so key-existence is a sound check there too. + # Spot-check every mode. Key-existence lookups are sound for ALL modes + # because the copy batch-Puts source items VERBATIM (source comment_id + # included) -- per-account comment_id divergence only affects rows each + # pipeline ingests itself, and those dual-window rows are excluded here by + # the same cutoff filter the copy used. For truncate_load keys are likewise + # copied verbatim. if args.sample: sampled = mismatched = 0 for item in scan_items(src, src_filter): @@ -351,7 +376,8 @@ def main(): p.add_argument("--execute", action="store_true") p.add_argument( "--backup-arn", - help="required with --execute for truncate-and-load tables", + help="required for truncate-and-load tables (validated on " + "dry-run as well as --execute)", ) else: p.add_argument(