Spensara Spensara Knowledge Base & Documentation Repository
System Overview Brink & Borg Master Index (722 Docs) Spensara Platform Project Tracker

Repository Docs & Knowledge Base

Consolidated Spensara documentation repository (722 Markdown & architecture documents in ../repository)
Built Word Documents (.docx): Spensara Overview (.docx) Capability Assessment (.docx) Brink POS Ingestion Analysis (.docx) The Borg Ingestion Architecture (.docx)

Categories

Branches (328)

Branch-specific documentation across extract-raw-xml, feature-fleet-import, kds_cdc_improvments, life-cycle, and tenat-admin.
branches/feature-fleet-import/implementation-specs/specs/ImportFromS3ToRedshift/TASK3.md 8,611 bytes · Updated Oct 23, 2026 10:44 AM EDT
Raw File

TASK 3: Add Redshift JSON Load, Validation, And Atomic Curated Replacement

Summary

Implement the Fleet-specific Redshift executor that loads date-grouped JSON manifests into temporary SUPER data, transforms and validates all candidates, then atomically replaces the exact curated source slice on one tenant connection.

Dependencies

  • TASK 0
  • TASK 0 frozen snapshot, replacement-intent, and result contracts. Integrate with Task 2 after both implementations exist.

Current Codebase Findings

  • ITenantWarehouseConnectionFactory opens a connection to the resolved tenant database.
  • IRedshiftS3AuthClauseFactory centralizes Redshift S3 authorization.
  • IRedshiftCopyCommandFactory emits Parquet COPY and must remain unchanged.
  • CuratedTargetExecutor demonstrates transaction and command-timeout handling, but its target/date parameter model does not fit a whole-folder Fleet import.
  • target DDL is defined in 0612_create_curated_pos_orders.sql and 0611_create_curated_pos_order_items.sql.

Files To Create

Recommended locations:

  • backend/Kds.Ai.Integrations/FleetS3/IFleetS3RedshiftImportExecutor.cs
  • backend/Kds.Ai.Integrations/FleetS3/FleetS3RedshiftImportExecutor.cs
  • backend/Kds.Ai.Integrations/FleetS3/FleetS3ImportSql.cs

Task 0 owns the shared request/result and diagnostic record files; Task 3 implements those frozen contracts.

Keep SQL explicit and source-specific. Inline raw string constants are acceptable; embedded SQL is also acceptable if the project is configured consistently and tests can load it.

One-Connection Execution

All temporary data and permanent replacement work must use one open tenant connection:

  1. create reusable session-scoped raw SUPER table;
  2. create aggregate temporary order/item candidate tables with target-compatible types plus validation lineage;
  3. for each date manifest in deterministic ascending order:
  4. truncate the raw temporary table;
  5. execute dedicated COPY ... MANIFEST FORMAT JSON 'noshred';
  6. assert loaded raw count equals manifest object count;
  7. transform rows into aggregate candidates while injecting the manifest path date as path_business_date;
  8. validate the complete aggregate candidates;
  9. replace permanent rows in one transaction;
  10. return counts and old/new affected dates.

The executor must not call the Parquet IRedshiftCopyCommandFactory.

Dedicated COPY Shape

The command uses fixed temporary identifiers, a storage-produced manifest URI, centralized S3 auth, and no caller-provided SQL:

sql
COPY fleet_s3_raw
FROM 's3://.../dt=2026-07-08.manifest.json'
<central auth clause>
MANIFEST
FORMAT JSON 'noshred';

Apply the configured long-running command timeout to COPY, validation, and replacement commands.

Candidate Mapping

Implement every mapping in proposed-solution.md, including:

  • canonical tenant and source metadata on both targets;
  • USD currency;
  • validated JSON/path/local timestamp date as business_date;
  • normalized status/channel values;
  • one-based item array ordinal;
  • compact modifier JSON and source-order modifier summary;
  • deterministic positive 32-bit order id using a documented Redshift hash expression;
  • deterministic {transaction_id}:{ordinal} source line identity.

Recommended order ID expression:

sql
ABS(MOD(FNV_HASH(transaction_id), 2147483646)) + 1

The collision check is mandatory; the hash is not treated as globally collision-free.

Channel normalization is exact:

  • apply the documented normalization function to order.order_type for service_channel;
  • apply it to nonblank order.destination for fulfillment_channel, falling back to order.order_type only when destination is null/blank;
  • never merge two conflicting values or choose precedence heuristically.

Complete Validation

Validation occurs before permanent deletes and returns aggregate counts plus at most the configured sample limit. It must detect:

  • raw count mismatch with snapshot manifest index;
  • root JSON type and schema version;
  • all required root, location, order, items, kds, partition, item, and modifier fields/types from schema v1.0;
  • empty item arrays;
  • transaction ID pattern, uniqueness, nonempty value, and VARCHAR(50) fit;
  • valid UTC/local timestamps;
  • path date equals JSON partition date and local timestamp calendar date;
  • location.terminal_id exactly equals the case-sensitive canonical source folder;
  • integer range/positivity rules;
  • decimal range/scale compatibility with DECIMAL(10,2) without silent rounding beyond the source contract;
  • all target string lengths, including generated source IDs and modifier fields;
  • order hash collisions and generated line-ID collisions;
  • candidate order/item count greater than zero;
  • every item has exactly one candidate parent;
  • unsupported/malformed values before casts that could turn them into null.

Do not skip, truncate, or coerce invalid rows. Full payloads and personal fields must not appear in diagnostics.

Atomic Replacement

After validation:

  1. begin a transaction;
  2. lock curated.pos_orders first and curated.pos_order_items second, matching the existing POS target order, before any delete;
  3. query old affected dates for the exact tenant/source discriminator;
  4. recheck matching-row existence and overrideExisting inside the transaction;
  5. construct a replacement intent containing batch ID, old/new/affected dates, candidate counts, source identity, and a deterministic intent checksum;
  6. invoke the Task 0 checkpoint contract and require PostgreSQL to durably persist that intent before any permanent delete;
  7. delete items, then orders, using both exact tenant_name and _etl_source_type;
  8. insert orders, then items, from candidates using INSERT ... SELECT;
  9. verify inserted counts, duplicate business keys, source lineage, and item-parent integrity;
  10. commit;
  11. return the persisted replacement intent plus commit result for publication.

If replacement-intent persistence fails, roll back Redshift without deleting permanent data. Persisting an intent before a Redshift rollback is safe: reconciliation sees no complete current slice for this batch and may restart pre-commit execution. This ordering ensures that a crash immediately after Redshift commit cannot lose old-only dates that Athena must remove.

Never delete by tenant, source system, batch, business date, or order ID alone.

Crash Reconciliation

A process can die after Redshift commits but before PostgreSQL records the commit flag. The replacement intent already exists at that point. Before any retry repeats replacement, load that intent and probe the current Fleet source slice:

  • if every row in both target tables is stamped with the current batch ID and counts/integrity match the durable snapshot expectations, return a reconciled committed result without deleting/reinserting;
  • if no current rows use this batch ID, normal pre-commit execution may continue;
  • if lineage is mixed or counts are inconsistent, fail with a non-destructive reconciliation error requiring investigation.

This check makes an interrupted same-batch delivery safe when Task 1's execution lease is reacquired. A batch already marked as a handled pre-commit failure is terminal and is not automatically resumed.

Testing

Add unit/recording-connection tests for:

  • dedicated JSON/noshred/manifest COPY and centralized auth;
  • no use or regression of Parquet COPY factory;
  • one COPY per populated path date, not per file;
  • fixed identifier and parameter/literal escaping;
  • transaction/lock/delete/insert ordering;
  • orders-then-items lock order and items-then-orders delete order;
  • exact tenant/source delete predicates;
  • rollback on every injected failure point;
  • overrideExisting true/false behavior;
  • old/new affected date union;
  • crash before intent persistence, after intent persistence/before commit, and immediately after commit;
  • commit reconciliation states.

Add opt-in live Redshift tests for SUPER traversal, array ordinals, all mappings, validation queries, hash collision detection fixture, rollback, and sample one-order/four-item result.

Acceptance Criteria

  • no permanent curated mutation happens before complete validation;
  • replacement of both tables is one Redshift transaction;
  • previous rows survive all pre-commit failures;
  • only the exact tenant/Fleet source slice changes;
  • repeated override runs produce the same business rows with new lineage;
  • no curated DDL or staging table is added.