TASK 6: Build Normalization And Redshift Staging Load Path

Summary

Implement the KDS CDC normalize_load and load_to_staging path that turns durable raw CDC artifacts into tenant- and entity-scoped S3 load Parquet, then applies those artifacts into source-shaped Redshift staging tables with idempotent COPY-based behavior.

This task freezes the KDS CDC normalization boundary, dedupe and ordering metadata rules, load-artifact contract, staging table catalog, and Redshift staging-application behavior for the four supported KDS entities.

This task does not yet implement curated transformation, curated export, or final checkpoint advancement across the full pipeline.

Current Codebase Findings

Repository analysis shows the codebase already has the shared building blocks for normalize/load and Redshift COPY, but all concrete runtime implementations are Brink-specific and KDS CDC needs its own source-shaped path.

• backend/Kds.Ai.Warehouse/Abstractions/Normalization/INormalizationStep.cs and backend/Kds.Ai.Warehouse/Models/Normalization/NormalizationContext.cs already define a generic normalization seam that consumes a raw artifact context and returns load artifacts.
• backend/Kds.Ai.Warehouse/Models/Normalization/NormalizationResult.cs returns LoadObjectWriteRequests, Entity, OperationName, SchemaVersion, RowCount, and total file size, which fits KDS CDC load-artifact production.
• backend/Kds.Ai.Warehouse/Models/Storage/LoadObjectWriteRequest.cs, LoadObjectKey.cs, and backend/Kds.Ai.Warehouse/Services/Storage/EtlS3KeyBuilder.cs already support S3 load writes with the existing layout load/company_id=.../tenant_id=.../entity=.../batch_id=.../created_at=.../....
• backend/Kds.Ai.Warehouse/Abstractions/IRedshiftStagingLoadExecutor.cs and backend/Kds.Ai.Warehouse/Services/RedshiftStagingLoadExecutor.cs already execute transactional DELETE plus Redshift COPY from a load S3 prefix into a tenant warehouse using RedshiftStagingLoadRequest.
• backend/Kds.Ai.Data/Services/StageUpstreamBatchResolver.cs already provides the repo pattern for resolving current successful upstream stage batches by tenant, business date, and stage name.
• backend/Kds.Ai.Integrations/BrinkPos/BrinkPosNormalizeLoadJobs.cs is the main normalize-load reference: it resolves upstream extract batch, starts normalize_load batch, runs normalization, writes load artifacts through IEtlS3Storage.WriteLoadAsync(...), stores structured diagnostics in DetailsJson, and advances checkpoint only on success.
• backend/Kds.Ai.Integrations/BrinkPos/BrinkPosLoadToStagingJobs.cs is the main load-to-staging reference: it resolves upstream normalize batch, reads structured load diagnostics, looks up table definitions from a catalog, runs IRedshiftStagingLoadExecutor, and records one diagnostic per staging table.
• backend/Kds.Ai.Integrations/BrinkPos/BrinkPosStagingTableCatalog.cs and backend/Kds.Ai.Integrations/Etl/StagingTableDefinition.cs already show the repo pattern for expressing target table name, schema version, delete SQL, copy-template id, and optional metadata-driven delete parameters.
• backend/Kds.Ai.Warehouse/Models/Normalization/NormalizedEtlMetadata.cs already injects standard ETL metadata columns such as _etl_batch_id, _etl_company_id, _etl_tenant_id, _etl_source_type, _etl_raw_s3_path, _etl_parser_version, and _etl_schema_version into normalized rows.
• backend/Kds.Ai.Warehouse/Services/Synthetic/SyntheticKdsSourceToStagingMapper.cs and RedshiftSyntheticKdsStagingWriter.cs are the best current references for KDS source-shaped row mapping and target staging-table columns, but they represent the old synthetic direct-write path rather than S3 Parquet plus COPY.
• Current synthetic KDS direct staging writer only handles staging.kds_orders, staging.kds_items, and staging.kds_item_bumps. It does not implement staging.kds_customers.
• Warehouse bootstrap SQL already contains staging.kds_customers in backend/Kds.Ai.Warehouse/Sql/Bootstrap/Tenant/Staging/Kds/040_create_staging_kds_tables.sql, but current runtime mapper/catalog/load behavior for KDS CDC customers is still absent.
• KDS staging bootstrap is currently selected through Brink POS service enablement in tenant warehouse bootstrap code, so KDS CDC onboarding must not assume customers or other KDS staging tables are available for tenants without Brink POS enabled.

These findings create four important Task 6 constraints:

1. KDS CDC should reuse the shared normalize/load abstractions, but it cannot reuse Brink normalization jobs or contracts directly because those are endpoint-contract shaped rather than CDC-row shaped.
2. KDS CDC normalize-load details must durably record tenant/entity artifact boundaries, dedupe outcomes, and ordering facts so load_to_staging can retry from S3 load without re-reading raw.
3. KDS CDC load-to-staging should use the existing transactional DELETE plus Redshift COPY executor, but it needs a new KDS staging table catalog with idempotent delete semantics keyed to KDS source rows rather than Brink day-based snapshots.
4. Task 6 must explicitly include customers in the staging-load design because the repo does not yet have a KDS customer runtime mapper/catalog/load path, even though staging DDL exists.

Dependencies

• docs/features/ETL/kds-cdc/index.md
• docs/features/ETL/kds-cdc/implementation-breakdown.md
• specs/KdsCdcIngestion/TASK5.md

Source References

• ../../docs/features/ETL/kds-cdc/implementation-breakdown.md#task-checklist
• ../../docs/features/ETL/kds-cdc/index.md#stage-6-normalize-to-s3-load-normalize_load
• ../../docs/features/ETL/kds-cdc/index.md#stage-7-load-to-source-shaped-staging-load_to_staging
• ../../docs/features/ETL/kds-cdc/index.md#checkpointing-and-replay
• ../../docs/features/ETL/kds-cdc/index.md#quarantine-and-failed-packages
• ../../docs/features/ETL/kds-cdc/index.md#supported-corner-cases
• ../../docs/features/ETL/kds-cdc/index.md#selected-behavior-summary
• ../../docs/features/ETL/etl-source-adapter-framework.md

Scope From Breakdown

This task covers only the sixth implementation-breakdown item:

• transform raw CDC envelopes into load-ready Parquet artifacts split by tenant and entity
• preserve delete flags, dedupe keys, and ordering metadata during normalization
• load source-shaped staging tables with idempotent Redshift COPY-based behavior
• fail safely when valid load artifacts or ordered staging application cannot be guaranteed

Files to Create

1. KDS CDC normalization seam

Create a dedicated KDS CDC normalization path that consumes raw CDC artifacts from Task 5 and produces load-ready Parquet artifacts for one tenant/entity slice.

Recommended logical location:

• backend/Kds.Ai.Integrations/KdsCdc/Normalization/

Recommended logical types:

• IKdsCdcNormalizationStep
• KdsCdcNormalizationStep
• IKdsCdcNormalizeLoadJobs
• KdsCdcNormalizeLoadJobs
• KdsCdcNormalizationContext
• KdsCdcNormalizationResult
• KdsCdcNormalizedBatchDetails
• KdsCdcNormalizedArtifactDiagnostic

Minimum responsibilities:

• open the upstream raw artifact using existing raw-reader/storage seams,
• deserialize source-faithful CDC records from Task 5 raw payloads,
• split and validate records by entity while preserving tenant isolation from upstream batch identity,
• map raw envelopes into source-shaped normalized rows for orders, items, item_bumps, and customers,
• preserve delete flags and ordering metadata,
• deduplicate exact duplicates safely using source coordinates plus primary key,
• quarantine record-local mapping or parse failures while letting good records continue,
• produce one or more Parquet load artifacts through LoadObjectWriteRequest.

2. KDS CDC normalized row and contract models

Create explicit row contracts for load-ready Parquet and diagnostics about how they were produced.

Recommended logical location:

• backend/Kds.Ai.Integrations/KdsCdc/Normalization/Models/

Recommended logical types:

• KdsCdcOrderNormalizedRow
• KdsCdcItemNormalizedRow
• KdsCdcItemBumpNormalizedRow
• KdsCdcCustomerNormalizedRow
• KdsCdcDeduplicationDiagnostic
• KdsCdcOrderingRange
• KdsCdcNormalizationQuarantineDiagnostic

Required shape guidance:

• normalized row payloads should match the destination KDS staging-table columns plus ETL metadata,
• preserve source is_deleted, source primary key, and enough ordering fields to validate downstream apply order,
• reuse NormalizedEtlMetadata pattern for shared _etl_* fields rather than inventing a second metadata convention.

3. KDS CDC staging table catalog

Create a dedicated staging-table catalog for KDS CDC source-shaped tables.

Recommended logical location:

• backend/Kds.Ai.Integrations/KdsCdc/

Recommended logical types:

• KdsCdcStagingTableCatalog
• KdsCdcStagingTableDefinition only if source-local extension is truly needed beyond StagingTableDefinition

Minimum catalog coverage:

• staging.kds_orders
• staging.kds_items
• staging.kds_item_bumps
• staging.kds_customers

Each catalog entry should define:

• contract or entity name,
• Redshift table name,
• schema version,
• idempotent delete SQL,
• Redshift copy-template id,
• any delete-parameter metadata keys only if needed.

4. KDS CDC load-to-staging jobs and details contract

Create the stage orchestration seam that reads normalize-load diagnostics and applies the load artifact into Redshift staging.

Recommended logical location:

• backend/Kds.Ai.Integrations/KdsCdc/
• backend/Kds.Ai.Integrations/KdsCdc/Models/

Recommended logical types:

• IKdsCdcLoadToStagingJobs
• KdsCdcLoadToStagingJobs
• KdsCdcLoadToStagingDetails
• KdsCdcStagingTableDiagnostic

Minimum responsibilities:

• resolve current successful upstream normalize_load batch,
• read structured normalized-artifact diagnostics from DetailsJson,
• validate that required artifacts and schema versions are present,
• run IRedshiftStagingLoadExecutor one tenant/entity artifact at a time,
• record per-table load diagnostics,
• fail safely before apply when global ordering cannot be proven.

Files to Modify

5. KDS CDC adapter stage wrappers from Task 1

Implement or refine the KDS CDC stage wrappers introduced in Task 1 so normalize_load and load_to_staging become real framework stages.

Expected areas:

• backend/Kds.Ai.Integrations/KdsCdc/
• backend/Kds.Ai.Integrations/DependencyInjectionExtensions.cs

Required behavior:

• KDS CDC normalize_load stage wrapper should consume Task 5 extract_raw outputs,
• KDS CDC load_to_staging stage wrapper should consume Task 6 normalize_load outputs,
• both wrappers must use shared EtlStageNames.NormalizeLoad and EtlStageNames.LoadToStaging.

6. Existing warehouse normalization and load seams

Review and likely modify existing shared seams only where KDS CDC needs additive capabilities.

Relevant areas:

• backend/Kds.Ai.Warehouse/Abstractions/Normalization/INormalizationStep.cs
• backend/Kds.Ai.Warehouse/Models/Normalization/NormalizationContext.cs
• backend/Kds.Ai.Warehouse/Models/Normalization/NormalizationResult.cs
• backend/Kds.Ai.Warehouse/Models/Storage/LoadObjectWriteRequest.cs
• backend/Kds.Ai.Warehouse/Models/Staging/RedshiftStagingLoadRequest.cs
• backend/Kds.Ai.Warehouse/Services/RedshiftStagingLoadExecutor.cs
• backend/Kds.Ai.Warehouse/Models/Normalization/NormalizedEtlMetadata.cs

Strong guidance:

• prefer additive KDS CDC-specific contracts over broad shared-abstraction changes unless a real cross-source gap exists,
• if ordering metadata, dedupe diagnostics, or artifact-prefix information must travel between stages, prefer structured KDS CDC DetailsJson contracts rather than overloading generic interfaces with KDS-only fields.

7. Warehouse bootstrap and staging schema coverage

Review and likely modify warehouse schema/bootstrap seams so KDS CDC load targets are enabled for all four entities, especially customers, without relying on Brink POS service enablement.

Expected areas:

• tenant warehouse bootstrap or DDL generation under backend/Kds.Ai.Warehouse/Services/
• any source-shaped schema creation paths under backend/Kds.Ai.WarehouseDataModel/ or related Redshift bootstrap code
• backend/Kds.Ai.Warehouse/Services/PostgresSeed/PostgresToRedshiftSeedService.cs only as reference for expected table presence, not as CDC load runtime

Important rules:

• staging.kds_customers DDL exists today, but Task 6 still must assert it is included in the KDS CDC bootstrap path.
• KDS CDC staging bootstrap must be enabled by KDS CDC service/source configuration, not by Brink POS credentials alone.

8. Tests

Add or modify tests under:

• backend/tests/Kds.Ai.Integrations.Tests/KdsCdc/
• backend/tests/Kds.Ai.Data.Tests/
• existing storage or warehouse tests only if shared seams need additive assertions

Required Behavior

Normalize-load boundary

normalize_load reads durable raw S3 artifacts from Task 5 and writes load-ready Parquet to S3 load.

Fixed load key intent from feature doc:

load/company_id={company_id}/tenant_id={tenant_id}/entity={entity}/batch_id={batch_id}/created_at={timestamp}/...

Task 6 must preserve these source rules:

• split output by tenant and entity,
• preserve source is_deleted,
• preserve CDC source coordinates in metadata columns or durable batch diagnostics,
• deduplicate exact duplicate CDC messages by source coordinates and primary key,
• keep global order metadata for downstream validation,
• quarantine record-local parse or mapping errors and continue with good records,
• fail the tenant/entity batch only when a safe load artifact cannot be produced.

S3 raw remains the replay source of truth. Load artifacts are an optimization checkpoint, not the only recoverable state.

Supported entity coverage

Task 6 must normalize and load all four documented KDS entities:

• staging.kds_orders
• staging.kds_items
• staging.kds_item_bumps
• staging.kds_customers

Current repo reference code covers only the first three via synthetic direct-write flow. Task 6 must explicitly close the customers runtime gap in normalization and table catalog, and must verify warehouse bootstrap enables the existing staging customer DDL for KDS CDC tenants.

Source-shaped row mapping

Normalized rows must be source-shaped for the existing KDS staging model, not curated-shaped.

Mapping guidance:

• use SyntheticKdsSourceToStagingMapper.cs as parity reference for field naming and value mapping where shapes overlap,
• do not route KDS CDC through SyntheticKdsSourceToStagingMapper at runtime if doing so would hide CDC-specific metadata, dedupe semantics, or customers support,
• preserve _etl_* metadata columns through normalized rows using NormalizedEtlMetadata or an additive equivalent that extends it cleanly.

At minimum, each normalized row or stage diagnostic must preserve:

• source primary key,
• source operation,
• source is_deleted,
• Debezium coordinate or equivalent source-order marker,
• raw batch id and raw S3 path,
• company id and tenant id,
• source type kds_cdc.

Deduplication and ordering

Exact duplicate CDC messages must be deduplicated by source coordinates plus primary key, not by SQS message id alone.

Task 6 must preserve enough order facts to let load_to_staging validate apply safety before Redshift COPY, including at minimum:

• first and last source coordinate in artifact,
• first and last sequence or ordering value in artifact,
• record count after dedupe,
• duplicate-collapse count,
• whether any ordering ambiguity remains.

If order cannot be proven safely for a tenant/entity slice, Task 6 must fail that slice before staging apply rather than doing best-effort Redshift load.

Record-local quarantine boundary

Record-local failures during normalization should be quarantined and not poison unrelated valid records when safe continuation is possible.

Examples:

• unsupported entity inside otherwise valid raw batch,
• invalid value that prevents one row from being mapped,
• duplicate message that cannot be reconciled safely,
• malformed row payload discovered only during source-row mapping.

Batch-level failure still applies when:

• a valid load artifact cannot be produced safely,
• ordering guarantee for tenant/entity slice is missing or ambiguous,
• required load metadata is incomplete,
• Redshift COPY or pre-apply validation fails.

Load-to-staging boundary

load_to_staging resolves the tenant Redshift database and runs Redshift COPY from S3 load Parquet into source-shaped staging tables.

Task 6 must preserve these load rules:

• use Redshift COPY, not row-by-row inserts,
• load one tenant/entity artifact at a time,
• apply idempotent delete/insert or merge-like behavior around the copied batch,
• preserve is_deleted for later soft-delete handling,
• validate global ordering before applying the batch,
• fail safely if a gap or out-of-order condition is detected,
• retry idempotently from the same S3 load artifact if COPY fails,
• never load records whose tenant mapping failed or was ambiguous.

Idempotent delete/apply semantics

Task 6 must define KDS-specific idempotent apply semantics per entity. Preferred direction:

• delete by source primary keys for the entity slice being applied,
• then COPY the load artifact into the target staging table within the same transaction,
• avoid tenant-day blanket deletes unless they are proven correct for that KDS entity and CDC batch shape.

Task 6 must define how row-key deletion obtains keys from the current load artifact.

Acceptable directions:

• copy into a temporary or staging work table first, then delete matching keys from the target and insert from the work table in one transaction,
• persist a key manifest next to the load artifact and use it to parameterize delete safely,
• extend the existing staging load executor only if the extension remains generic and test-covered.

The design must not pretend a static delete SQL can key-delete arbitrary CDC rows unless it has access to the row keys for the current artifact.

Reason:

• CDC micro-batches are row-level deltas, not full-day snapshots like some Brink flows,
• row-key-based idempotence better matches duplicate replay and partial retry requirements from the feature doc.

Durable normalize/load diagnostics

Task 6 should persist structured diagnostics in DetailsJson for both stages.

normalize_load durable details should include at minimum:

• upstream raw batch id,
• entity,
• schema version,
• artifact prefix URI,
• artifact URIs,
• row count,
• duplicate-collapse count,
• quarantined-record count,
• source coordinate range,
• SQS or global ordering range,
• parser or mapper version.

load_to_staging durable details should include at minimum:

• upstream normalize-load batch id,
• target table name,
• schema version,
• load artifact prefix URI,
• copy-template id,
• deleted-row count when measured,
• copied-row count,
• ordering validation result,
• start and completion timestamps.

This durable metadata must support later replay from load artifact without re-running normalization when COPY alone fails.

Business date rule

For KDS CDC normalize/load and load-to-staging batches, BusinessDate remains processing date inherited from upstream KDS CDC stages:

DateOnly.FromDateTime(DateTime.UtcNow)

Task 6 must not switch to source event date or Brink snapshot date semantics.

Proposed Implementation Shape

Recommended high-level flow:

Task 5 extract_raw batch succeeds for one tenant/entity slice
  -> KDS CDC normalize_load resolves upstream raw batch
  -> read raw artifact from S3 raw
  -> deserialize CDC envelopes
  -> dedupe exact duplicates by source coordinate + primary key
  -> quarantine record-local mapping failures
  -> map good rows into source-shaped normalized rows
  -> write Parquet load artifact(s) to ETL S3 load
  -> persist normalize-load details with artifact URIs and ordering ranges
  -> KDS CDC load_to_staging resolves normalize-load batch
  -> validate artifact schema and ordering metadata
  -> run transactional delete + Redshift COPY for target staging table
  -> persist load-to-staging details for retry and lineage

Recommended normalization job contract shape:

public interface IKdsCdcNormalizeLoadJobs
{
    Task<KdsCdcStageJobResult> ExecuteAsync(
        KdsCdcNormalizeLoadJobInput input,
        CancellationToken ct = default);
}

Recommended load-to-staging job contract shape:

public interface IKdsCdcLoadToStagingJobs
{
    Task<KdsCdcStageJobResult> ExecuteAsync(
        KdsCdcLoadToStagingJobInput input,
        CancellationToken ct = default);
}

Recommended KDS CDC staging catalog entries:

orders      -> staging.kds_orders
items       -> staging.kds_items
item_bumps  -> staging.kds_item_bumps
customers   -> staging.kds_customers

Recommended normalize-load details shape:

KdsCdcNormalizeLoadDetails
  CompanyId
  TenantId
  BusinessDate
  PipelineStage
  UpstreamExtractBatchId
  Entity
  SchemaVersion
  ArtifactPrefixUri
  ArtifactUris
  RowCount
  DuplicateCount
  QuarantinedCount
  SourceCoordinateRange
  OrderingRange
  MapperVersion

Recommended load-to-staging details shape:

KdsCdcLoadToStagingDetails
  CompanyId
  TenantId
  BusinessDate
  PipelineStage
  UpstreamNormalizeLoadBatchId
  TableName
  ArtifactPrefixUri
  CopySqlTemplateId
  DeletedRows
  SourceRowCount
  OrderingValidationStatus
  StartedAt
  CompletedAt

Important design notes:

• prefer one normalize-load batch and one load-to-staging batch per tenant/entity slice, aligned with the upstream raw-batch boundary from Task 5,
• use existing IEtlS3Storage.WriteLoadAsync(...) and current ETL S3 load layout,
• use existing IRedshiftStagingLoadExecutor for transactional apply rather than inventing a second KDS-only Redshift loader,
• keep KDS CDC durable details structured enough that a later retry can reopen the same S3 load prefix and rerun COPY without re-reading SQS,
• keep customers first-class in the design even if current synthetic path lacks it.

Testing

Unit tests

Add tests for:

• raw CDC envelopes map into source-shaped normalized rows for orders, items, item_bumps, and customers,
• normalized rows preserve is_deleted, source key, _etl_* metadata, and source coordinate facts,
• exact duplicates collapse by source coordinates plus primary key,
• duplicate SQS message ids without duplicate source coordinates do not trigger false dedupe,
• record-local mapping failures quarantine bad rows while good rows still produce load artifacts,
• normalize-load details record artifact URIs, row counts, duplicate counts, and ordering ranges,
• load keys use company_id, tenant_id, entity, batch_id, and created_at partitions,
• load-to-staging job rejects missing artifact metadata or schema mismatch before Redshift apply,
• load-to-staging ordering validation blocks apply when ordering gap or ambiguity is present,
• load-to-staging builds correct table selection and copy-template selection for all four entities.

Data and warehouse tests

Add tests for:

• KDS CDC staging table catalog definitions match expected table names and copy-template ids,
• delete SQL is idempotent and keyed to CDC row-apply semantics,
• warehouse bootstrap or schema creation covers staging.kds_customers as well as existing KDS staging tables,
• KDS CDC bootstrap enablement creates KDS staging tables independently of Brink POS credential enablement,
• row-key idempotent apply obtains delete keys from a temp/work table or key manifest and does not use tenant-day blanket delete,
• Redshift copy command generation still produces Parquet COPY SQL with proper S3 path and auth clause.

Integration and composition tests

Add tests proving:

• successful KDS CDC raw batch can be normalized into load artifacts and then applied to Redshift staging through framework stage jobs,
• load_to_staging can retry from the same load artifact when COPY fails after normalization already succeeded,
• unresolved or quarantined records from earlier tasks never reach normalize-load or Redshift apply,
• partial success across unrelated tenant/entity slices remains possible,
• Brink normalize-load and load-to-staging behavior remains unchanged.

Not required in task 6

This task does not require:

• curated merge logic,
• curated-to-S3 export,
• final checkpoint advancement decisions,
• replay CLI or operator tooling.

Acceptance Criteria

• [ ] KDS CDC normalize_load reads Task 5 raw artifacts and writes tenant/entity-scoped Parquet load artifacts under existing ETL S3 load layout
• [ ] Normalization preserves is_deleted, source keys, and durable ordering metadata
• [ ] Exact duplicates are deduplicated by source coordinates plus primary key
• [ ] Record-local normalization failures are quarantined without blocking unrelated good rows when safe continuation is possible
• [ ] KDS CDC load_to_staging uses Redshift COPY, not row-by-row inserts
• [ ] Staging apply is idempotent and retryable from the same load artifact
• [ ] Row-key staging apply design explicitly obtains keys from the current artifact, temp/work table, or key manifest
• [ ] Ordering validation blocks unsafe staging apply before Redshift load
• [ ] All four KDS entities are covered, including customers
• [ ] KDS CDC staging bootstrap does not depend on Brink POS service enablement
• [ ] KDS CDC durable normalize_load and load_to_staging details support later retry and lineage without re-reading SQS
• [ ] Shared EtlStageNames.NormalizeLoad and EtlStageNames.LoadToStaging are used
• [ ] Tests cover normalization mapping, dedupe, ordering validation, staging table coverage, and load retry behavior

Notes

• Follow docs/features/ETL/kds-cdc/index.md#stage-6-normalize-to-s3-load-normalize_load and #stage-7-load-to-source-shaped-staging-load_to_staging as source of truth.
• Use Brink normalize/load jobs as structural reference only; do not inherit Brink snapshot or tenant-day delete semantics unless they are explicitly valid for KDS CDC row deltas.
• Use synthetic KDS mapper and direct staging writer as field-parity references only; KDS CDC runtime must stay on the ETL raw -> load -> Redshift COPY path.
• Keep staging.kds_customers explicit in Task 6 because DDL exists but current runtime mapper/catalog/load behavior and KDS CDC bootstrap enablement are not finished.