ETL & Data Ingestion Architecture

Context

KDSAI needs one reliable pipeline that takes customer data, stores the raw input safely, loads it into Redshift, and makes it queryable for monitoring.

This document replaces the earlier multi-path lakehouse plan with a simpler default architecture:

Customer source -> .NET ingestion API / ETL worker -> S3 raw -> tenant Redshift database (staging -> curated -> serving)

Current state

• No production ETL pipeline exists yet.
• The traffic simulator writes directly to PostgreSQL via EF Core (KDSContext).
• AI agents query warehouse data through raw SQL.
• The current warehouse already contains source-specific schemas, but loading and standardization are still manual or external.
• The current ETL delivery slices bootstrap and use source-mirror Redshift objects such as curated.kds_orders, curated.kds_items, and curated.kds_item_bumps.
• Canonical runtime tables such as curated.orders and curated.order_items are not part of the current branch implementation.

What this document optimizes for

• Get customer data into Redshift quickly.
• Keep one raw copy of everything received.
• Use one default implementation stack based on .NET.
• Make tenant-specific Redshift databases the primary query surface for single-store users.
• Support cross-tenant querying where AI agents query Athena over S3 curated data on behalf of multi-store users.
• Add new source adapters without redesigning the platform.

What this document does not try to build

• A full multi-tenant lakehouse with Crawlers, Lake Formation, and Spectrum-first querying.
• All four ingestion modes with identical implementation depth from the start.
• Query-time cleanup owned by AI agents.
• Heavy streaming infrastructure unless workload or latency requirements justify it.

Default Architecture

Core decision

Use one ingestion pipeline with four standard ingestion modes:

• File upload
• API push
• API pull
• CDC

All four modes should converge into the same downstream flow:

1. Receive data or changes from a customer source.
2. Store the raw payload, response, file, or CDC batch in S3.
3. Normalize it into load-ready files.
4. Select the tenant Redshift database and load its staging schema with COPY.
5. Transform staging data into curated tables inside that tenant database, with optional serving views added only when a published convenience layer is needed later.
6. UNLOAD curated tables from Redshift to S3 curated/ as Parquet, preserving tenant_id partitioning.
7. Let AI agents and monitors query approved tenant-scoped curated tables/views via Redshift for single-store users.
8. Let AI agents query curated/ Parquet in S3 via Athena for multi-store users, scoped by the platform-injected tenant filter.

This keeps the model simple: Redshift is where curated data becomes usable for single-store contexts, Athena is where agents query the same curated data across tenants on behalf of multi-store users, and tenant isolation does not depend on generated SQL filters in either path. Within the single-store Redshift path, approved curated objects are the primary monitor and agent contract. Optional serving views may be added later as a published convenience layer, but they are not required for the initial scope.

Visual references

• docs/C4/views.c4 - etloverview, etl_with_sources
• docs/C4/usecases/etl_ingestion.c4 - etl_file_upload, etl_api_push, etl_api_pull, etl_cdc, etl_end_to_end

Architecture Overview

1. Ingestion API

The ingestion API is a thin .NET service that:

• receives upload requests or API push payloads,
• authenticates the tenant,
• records a batch in operational metadata,
• stores or routes the raw payload to S3,
• enqueues work for the ETL worker.

This should stay intentionally small. It is an entry point, not the transformation engine.

2. ETL Worker

The ETL worker is the main execution engine. It:

• polls scheduled external APIs,
• executes CDC snapshot or change extraction jobs,
• consumes queued ingestion jobs,
• validates and normalizes raw payloads,
• writes load-ready files to S3,
• resolves the target tenant database,
• runs Redshift COPY into that database's staging schema,
• runs SQL transforms from staging into curated tables and optional serving views,
• runs Redshift UNLOAD to export curated tables to S3 curated/ as Parquet for cross-tenant Athena queries,
• updates checkpoints and batch status.

One worker service is enough to start and can be scaled out later if needed.

3. S3 Raw Storage

S3 is the landing zone and replay buffer.

Use it to keep:

• original customer payloads,
• raw API pull responses,
• CDC snapshots or change batches,
• failed payloads that need replay or investigation.

S3 is not the primary query layer for single-store users. It is the durability layer and the cross-tenant query surface for Athena.

4. Tenant Redshift Databases

Each tenant gets a dedicated Redshift database.

The ETL worker resolves the target database from tenant metadata and loads data only into that tenant database. The query service uses the same mapping so the agent never chooses the database itself.

5. Schemas Inside Each Tenant Database

Use the same schema template in every tenant database:

• staging for source-aligned load tables,
• curated for normalized analytics tables,
• serving for optional stable published views over curated data,
• optional etl_meta for tenant-local load audit and operational state.

Within each tenant database:

• staging should stay close to the source shape,
• load metadata such as batch_id, source_type, and ingested_at should be preserved,
• curated should own type cleanup, deduplication, source-specific mapping, business keys, and consistent time fields,
• curated is the primary canonical query surface for agents and monitors,
• serving is optional and should only expose stable published query surfaces when a simplified convenience layer is needed later.

The important shift is that the platform, not the AI agent, owns data correctness.

Data Flow

End-to-end batch lifecycle

1. Data arrives from file upload, API push, scheduled API pull, or CDC extraction.
2. The raw payload is written to S3 under the tenant and source prefix.
3. A load job is created for the ETL worker.
4. The worker validates the payload and produces a load-ready file if conversion is needed.
5. The worker resolves the tenant database and runs COPY into that database's staging schema.
6. The worker runs SQL in that same database to merge or insert data into curated tables and refresh optional serving views when they exist.
7. The worker runs UNLOAD to export the updated curated tables to S3 curated/ as Parquet, partitioned by tenant_id.
8. The worker marks the batch successful only after load, transform, and UNLOAD succeed.
9. Checkpoints advance only after success.

S3 layout

Use a simple, explicit layout:

s3://kdsai-ingestion/
  raw/
    company_id={company_id}/
      tenant_id={tenant_id}/
        source_type={source_type}/
          {yyyy}/{MM}/{dd}/
            batch_id={batch_id}/
              entity={entity}/
                payload_{captured_at_utc}.json
                source_{captured_at_utc}.csv
                source_{captured_at_utc}.xlsx
  load/
    {tenant_id}/
      {entity}/
        {yyyy}/{MM}/{dd}/
          {batch_id}.parquet
  curated/
    company_id={company_id}/
      tenant_id={tenant_id}/
        entity={entity}/
          dt={yyyy-MM-dd}/
            batch_id={batch_id}/
              part-000.parquet
  failed/
    {tenant_id}/
      {source_type}/
        {batch_id}/
          error.json
          original-payload.json

Guidelines:

• raw/ keeps the original customer payload.
• load/ contains files that Redshift can load directly.
• curated/ contains Parquet files UNLOADed from Redshift curated tables. Athena reads this prefix for cross-tenant queries. Data quality is identical to what Redshift serving views expose.
• failed/ supports replay and debugging without losing the original input.

Redshift databases and schemas

Use one Redshift database per tenant, for example tenant_acme, tenant_store42, or another stable tenant naming convention from the control plane.

Use the same schema template in every tenant database:

• staging - source-aligned load tables (we might skip this step and transform the source data directly into curated tables)
• curated - normalized analytics tables.
• serving - stable monitor-facing query surfaces.
• optional etl_meta - tenant-local load metadata, watermarks, and audit history.

The ETL worker and the query service should both resolve the target database from tenant context stored in operational metadata.

Example flow:

-- connect to tenant_acme
COPY staging.orders_incremental
FROM 's3://kdsai-ingestion/load/tenant-123/orders/2026/03/19/'
IAM_ROLE '...'
FORMAT AS PARQUET;

MERGE INTO curated.orders AS target
USING staging.orders_incremental AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET
  updated_at = source.updated_at,
  status = source.status,
  total_amount = source.total_amount
WHEN NOT MATCHED THEN INSERT (...)
VALUES (...);

-- export curated data to S3 for cross-tenant Athena queries
UNLOAD ('SELECT ''tenant-123'' AS tenant_id, * FROM curated.orders')
TO 's3://kdsai-ingestion/curated/tenant-123/orders/2026/03/19/'
IAM_ROLE '...'
FORMAT AS PARQUET
ALLOWOVERWRITE;

The exact SQL will differ by entity, but the pattern matters: connect to the tenant database, land first, standardize inside that tenant database, then UNLOAD curated results to S3 so Athena can query them across tenants. Optional serving views can be added later as convenience projections, but they are not a prerequisite for the first runtime contract.

Supported Ingestion Modes

1. File upload

This is the lowest-friction ingestion mode.

Example flow:

1. Customer requests an upload session from the ingestion API.
2. API records the batch and returns a presigned S3 upload target.
3. Customer uploads CSV or Excel.
4. S3 event or queued job wakes the ETL worker.
5. Worker validates columns, converts to load-ready Parquet, and loads Redshift.

Guardrails:

• quarantine unsupported file types,
• require explicit mapping for each file template,
• capture row counts and validation errors per batch,
• never overwrite the original upload.

2. API push

This mode is for systems that can post payloads or batches directly into KDSAI.

Example flow:

1. Source system posts payloads or batches to the ingestion API.
2. API authenticates the sender, stores the raw payload in S3, and enqueues a load job.
3. ETL worker validates and converts the pushed data into load-ready files.
4. Worker resolves the tenant database, loads staging, and updates curated and serving objects.

Important rule:

• keep requests idempotent and versioned, and move heavy processing out of the request path.

3. API pull

This is the path for third-party systems that expose a usable API but cannot push data to KDSAI.

Example flow:

1. Store encrypted customer API credentials in operational metadata.
2. ETL worker polls the external API on a schedule.
3. Worker stores the raw response in S3.
4. Worker resolves the tenant database, normalizes the response, and loads Redshift.
5. Worker advances the sync checkpoint only after the batch succeeds.

This adapter shares the same pipeline and does not require new platform services.

Important rules:

• checkpoints move only after successful load and transform,
• retries must be idempotent,
• pagination and rate limits are part of the adapter contract.

4. CDC

This is the path for systems where direct database replication or change extraction is available and valuable.

Example flow:

1. Configure the source connection, table allowlist, keys, and replication strategy.
2. Run an initial snapshot into raw storage and the tenant Redshift staging schema.
3. Capture ongoing inserts, updates, and deletes as CDC batches.
4. Load staging tables that stay close to the source table shape.
5. Merge or rebuild curated tables and refresh serving views from the replicated source data.

Important rules:

• keep CDC confined to source-aligned staging or mirror tables,
• handle deletes explicitly,
• version schema changes and table allowlists,
• treat tools like AWS DMS as implementation choices, not as the whole platform architecture.

Data Ownership and Transformation Rules

The pipeline owns correctness

The ETL pipeline is responsible for making data usable. That means:

• source cleanup happens in the worker or in Redshift SQL,
• tenant Redshift databases carry the stable semantics,
• AI agents and monitors query the result of platform-owned transformations through approved curated objects.

The AI agent does not fix the warehouse at query time

AI agents should not be the main normalization strategy, and they should not have direct access to cross-tenant or internal ETL objects.

Bad default:

• raw source data in S3,
• business cleanup deferred to generated SQL at query time,
• every monitor rediscovering schema and quality problems.

Good default:

• raw payload preserved in S3,
• source-specific cleanup done once,
• approved curated objects reused by monitors within the tenant database.

Tenant Isolation

Use database per tenant as the default isolation model for single-store operations.

• every raw object path includes tenant_id,
• every batch and checkpoint record is tenant-scoped,
• every tenant has a dedicated Redshift database with the same schema template,
• the ETL worker resolves the target database from tenant metadata and never writes a batch into the wrong tenant database,
• for single-store users, the query service resolves the database from the authenticated tenant context and the agent never chooses the database,
• for multi-store users, the query service resolves the authorized tenant list and routes the agent to Athena with a platform-injected WHERE tenant_id IN (...) filter,
• agents get read access only to approved curated tables/views in Redshift, plus optional serving views if present later, or curated Athena tables, always scoped to the user's authorized tenants.

Inside a tenant-specific Redshift database, tenant_id columns are optional in curated tables for single-store queries. However, the UNLOAD step must include tenant_id in every Parquet file written to S3 curated/, because Athena cross-tenant queries depend on it for partition pruning and result attribution.

Do not make Lake Formation a prerequisite for this architecture. If cross-engine governance or stronger AWS boundaries become necessary later, add them without changing the core ingestion pattern.

Cross-Tenant Query Layer

Problem

Some users manage multiple stores: chain managers, district managers, regional VPs, or franchise executives. When these users interact with the system, the AI agent acting on their behalf needs to query across their stores with the same data quality as single-store Redshift queries. The database-per-tenant Redshift model is not designed for cross-database queries, so a separate query path is needed for the agent.

Athena as the cross-tenant engine

AWS Athena queries S3 curated/ Parquet data that was UNLOADed from Redshift curated tables after each ETL batch. This guarantees data parity: cross-store queries see the same transformed, cleaned data that single-store Redshift runtime queries use.

Key properties:

• No data duplication. Athena reads from the same S3 Parquet files the ETL worker UNLOADs from Redshift curated tables. There is no separate materialization or copy job.
• Partition pruning. Athena scans only the tenant_id partitions the user is authorized to see. A query across 5 of 200 stores scans only those 5 prefixes.
• Serverless, pay-per-scan. Cross-store queries are expected to be less frequent than single-store monitoring. Athena charges per TB scanned with no idle cost.
• Keeps Redshift databases clean. Single-store Redshift databases stay isolated. No cross-tenant views, no union queries, no data leakage paths inside Redshift.

Glue Data Catalog

Define Glue Data Catalog tables over S3 curated/ data with tenant_id as a partition key. Use partition projection to avoid crawlers and make new partitions visible immediately:

CREATE EXTERNAL TABLE kdsai_crossstore.orders (
  order_id STRING,
  store_id INT,
  status STRING,
  total_amount DECIMAL(10,2),
  created_at TIMESTAMP,
  updated_at TIMESTAMP
)
PARTITIONED BY (tenant_id STRING, dt STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS PARQUET
LOCATION 's3://kdsai-ingestion/curated/'
TBLPROPERTIES (
  'projection.enabled' = 'true',
  'projection.tenant_id.type' = 'injected',
  'projection.dt.type' = 'date',
  'projection.dt.range' = '2024/01/01,NOW',
  'projection.dt.format' = 'yyyy/MM/dd',
  'storage.location.template' =
    's3://kdsai-ingestion/curated/${tenant_id}/orders/${dt}/'
);

Partition projection eliminates the need for MSCK REPAIR TABLE or Glue Crawlers. New tenant data appears in Athena as soon as the UNLOAD writes the Parquet files to S3.

Dual-path query resolution

The query service uses the authenticated user's tenant access to choose the query engine for the agent:

The platform injects the tenant filter in both paths. The agent generates the analytical SQL (aggregations, comparisons, anomaly detection), and the platform wraps it with the authorized tenant scope. The only difference is that in the multi-store case, the agent has broader read access across multiple tenants via Athena instead of being confined to a single Redshift database. The agent never constructs the tenant filter itself.

Data consistency

The UNLOAD step runs after Redshift curated transforms complete successfully, so Athena cross-store queries see the same data quality as single-store Redshift curated queries. If an UNLOAD fails, the batch is not marked successful, and the curated S3 data for that batch stays at the previous state until a successful retry.

UNLOAD freshness depends on ETL batch frequency. For most restaurant monitoring use cases (hourly or daily batches), this lag is acceptable. If near-real-time cross-store queries become a requirement, consider supplementing with Redshift data sharing or Redshift Spectrum as a future optimization.

Access Control and Store Groups

User-to-tenant mapping

Every user has a resolved list of tenant IDs they can access. This is stored in the user_tenant_access table in operational metadata:

The query service always reads from user_tenant_access to resolve which tenants a user can query. This table is the single source of truth for access control at query time.

• A single-store manager has one row.
• A district manager has rows for each store in their district.
• A chain executive has rows for every store in the chain.
• Adding or removing store access is a single row insert or delete.

Store groups as a convenience layer

For bulk assignment, store groups provide a named abstraction over sets of tenants:

When a user is added to a store group, the platform expands the group membership into individual user_tenant_access rows. When a tenant is added to or removed from a group, the platform updates the affected users' user_tenant_access rows accordingly.

Store groups are a management convenience. The query service does not resolve groups at query time -- it only reads the flat user_tenant_access table.

Resolution flow

1. Authenticate the user.
2. Look up all tenant_ids in user_tenant_access for that user.
3. If exactly one tenant: route to that tenant's Redshift database (single-store path).
4. If multiple tenants: route to Athena with WHERE tenant_id IN (...) injected from the resolved list.
5. The agent generates analytical SQL; the platform owns the tenant scope in both paths.

Operational Requirements

These are baseline operating requirements.

Batch metadata

Track every ingestion batch with:

• batch_id,
• tenant_id,
• tenant_database,
• ingestion_mode,
• source_type,
• received_at,
• load_started_at,
• load_completed_at,
• status,
• row counts and error details,
• raw S3 path and load S3 path.

Checkpointing

For scheduled pulls and CDC sources, store checkpoints in operational metadata and advance them only after the tenant Redshift load finishes successfully.

Retry and DLQ

Use a queue and dead-letter path for failed jobs. A bad payload should fail one batch, not the whole pipeline.

Replay and backfill

The worker must be able to replay from S3 raw objects or rerun a date range without manual data surgery.

Observability

At minimum, capture:

• batches received,
• batches failed,
• load duration,
• Redshift row counts by tenant database,
• checkpoint lag,
• per-tenant freshness,
• UNLOAD success/failure and duration per tenant,
• Athena queries per user and per store group,
• Athena bytes scanned and estimated cost per query,
• curated S3 freshness lag (time since last successful UNLOAD per tenant and entity).

Credentials and security

• Encrypt customer API credentials at rest.
• Avoid storing secrets in raw payloads when not needed.
• Prefer presigned uploads over broad S3 write access.
• Keep the tenant-to-database mapping in service-owned metadata, not in agent prompts or user-provided SQL.

Implementation Areas

Foundation

1. Create S3 raw, load, and failed prefixes.
2. Define the standard tenant Redshift database template with staging, curated, serving, and optional etl_meta schemas.
3. Create batch, checkpoint, and tenant-to-database mapping tables in the operational database.
4. Build the ETL worker skeleton with queued job execution and tenant database routing.

File upload path

1. Add upload session endpoint.
2. Add file validation and mapping.
3. Convert supported files to load-ready format.
4. Load tenant staging and populate curated plus serving objects.

Programmable adapters

Support one or more of these based on the source system:

• API push for systems that can post data directly.
• API pull for external integration APIs.
• CDC for sources where a warehouse copy of database tables is the right fit.

Curated runtime layer

1. Build the first stable curated tables.
2. Point AI agents, monitors, and analytics to approved curated objects first.
3. Add serving views only when a published convenience layer is justified by a concrete use case.

Operational hardening

1. Add replay tooling.
2. Add DLQ handling and alerting.
3. Add per-tenant freshness dashboards.
4. Add backfill support.

Cross-tenant query layer

1. Create Glue Data Catalog database kdsai_crossstore.
2. Define Glue tables with partition projection over S3 curated/ for each entity (orders, order_items, payments, etc.).
3. Create an Athena workgroup with cost controls (per-query byte limit, workgroup-level budget).
4. Add the UNLOAD step to the ETL worker after curated transforms succeed.
5. Implement dual-path query resolution in the query service (single tenant -> Redshift, multiple tenants -> Athena).
6. Ensure the platform injects the tenant filter from user_tenant_access for every Athena query.

Access control

1. Create user_tenant_access, store_groups, store_group_members, and store_group_users tables in operational metadata.
2. Build the group expansion logic: adding a user to a group or a tenant to a group updates user_tenant_access.
3. Integrate the query service with user_tenant_access for tenant resolution.
4. Add admin endpoints for managing store groups and user-tenant assignments.

Optional Components

Add these only when a concrete requirement justifies them:

• Kinesis or Firehose for ingestion buffering.
• Glue Crawlers for schema discovery (partition projection is the default; crawlers are only needed if dynamic schema detection becomes a requirement).
• Lake Formation LF-Tags or separate Redshift workgroups for stricter isolation.
• Redshift Spectrum as the main query path.
• Redshift data sharing for near-real-time cross-tenant queries (alternative to UNLOAD + Athena if latency requirements tighten).
• Side-by-side AWS-native and .NET-native architectures.

If CDC is needed, choose the narrowest implementation that works for the source. AWS DMS can be one option, but it should not become the default foundation for every ingestion mode.

Each of these may become useful later, but none of them is required to solve the initial problem of getting customer data into Redshift and querying it.

Relationship to Other Documents

• docs/REDSHIFT.md describes the warehouse itself. This ETL document defines how data gets into Redshift.
• docs/C4/ contains the diagrams for the simplified ingestion pipeline.
• docs/README.md should treat this pipeline as the default ingestion path for customer data.

Decision Summary