Oracle to PostgreSQL Time-Window Data Reload

The problem

AWS DMS has no facility for reloading a specific window of history while CDC continues. The ReloadTables API performs a complete reload of the tables it is given; it has no time predicate. CDC itself works forward from the transaction log, so injecting older rows into that stream without creating duplicates or inconsistencies is not something the service arbitrates.

Reloading a date range therefore means building it, and there are four shapes it can take.

Staging tables and a merge. A second, one-off DMS task loads the window into staging tables in the target database, and application-owned SQL merges staging into production. The ongoing CDC task is untouched throughout. This is the approach documented below.

A second CDC task from an earlier start point. Create a CDC-only task whose start position is the historical point of interest, direct it at a separate schema, and merge the result. This replays changes rather than reloading rows, so it only reaches as far back as the source’s transaction logs do — and a task’s CDC start point cannot be changed once the task exists, so it must be created at the right position.

Restore a snapshot. Restore a backup from the point in time into a separate environment, extract only the rows needed, and merge. Slow and expensive, but it is the only option when the source’s logs no longer cover the window.

Reconcile row by row. Run DMS data validation to identify which rows actually differ between source and target, and reload only those. Narrowest blast radius, and appropriate when the divergence is known to be small.

The rest of this page implements the first of these. The table and column names are from a particular migration and are illustrative; the dates in the examples are a January 2025 window.

What the approach has to satisfy

• The ongoing CDC replication is not interrupted or reconfigured.
• The merge into production is a single transaction — either the whole window lands or none of it does.
• The refresh is confined to the chosen time window.
• Every merge leaves an audit record of what it changed.

Prerequisites

• AWS DMS with a source Oracle endpoint, a target PostgreSQL endpoint, and a replication instance with capacity to run a second task alongside the existing CDC task.
• Credentials for both databases with sufficient privilege, held in a secret store rather than in the scripts.
• AWS CLI installed and configured on the workstation running the orchestration.
• A source column that reliably records when a row last changed. The whole approach rests on UPDATED_AT being maintained; rows changed without it being updated will not be in the window.

Step 1: prepare the target PostgreSQL environment

Staging schema and tables

Staging tables mirror the production tables, plus one column to track merge state.

CREATE SCHEMA IF NOT EXISTS staging;

CREATE TABLE staging.customers (
    customer_id INTEGER PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100),
    status VARCHAR(20),
    created_at TIMESTAMP,
    updated_at TIMESTAMP,
    _dms_processing_status VARCHAR(20) DEFAULT 'NEW'
);

CREATE TABLE staging.orders (
    order_id INTEGER PRIMARY KEY,
    customer_id INTEGER,
    amount NUMERIC(10,2),          -- Oracle NUMBER maps to NUMERIC
    order_date TIMESTAMP,
    updated_at TIMESTAMP,
    _dms_processing_status VARCHAR(20) DEFAULT 'NEW'
);

CREATE INDEX idx_customers_updated_at ON staging.customers(updated_at);
CREATE INDEX idx_orders_updated_at ON staging.orders(updated_at);

CREATE SCHEMA IF NOT EXISTS admin;

CREATE TABLE admin.merge_logs (
    id SERIAL PRIMARY KEY,
    table_name VARCHAR(100),
    merge_time TIMESTAMP,
    records_updated INT,
    records_inserted INT
);

CREATE TABLE admin.merge_summary (
    id SERIAL PRIMARY KEY,
    merge_start TIMESTAMP,
    merge_end TIMESTAMP,
    status VARCHAR(20),
    notes TEXT
);

Merge procedure

The procedure updates rows that already exist and inserts those that do not, per table, then records what it did.

CREATE OR REPLACE PROCEDURE merge_time_window_data()
LANGUAGE plpgsql
AS $$
DECLARE
    merge_start_time TIMESTAMP;
    records_updated INT;
    records_inserted INT;
BEGIN
    merge_start_time := CURRENT_TIMESTAMP;

    WITH updated_rows AS (
        UPDATE production.customers p
        SET name       = s.name,
            email      = s.email,
            status     = s.status,
            updated_at = s.updated_at
        FROM staging.customers s
        WHERE p.customer_id = s.customer_id
          AND s.updated_at > p.updated_at
        RETURNING p.customer_id
    )
    SELECT COUNT(*) INTO records_updated FROM updated_rows;

    WITH inserted_rows AS (
        INSERT INTO production.customers
            (customer_id, name, email, status, created_at, updated_at)
        SELECT s.customer_id, s.name, s.email, s.status, s.created_at, s.updated_at
        FROM staging.customers s
        LEFT JOIN production.customers p ON s.customer_id = p.customer_id
        WHERE p.customer_id IS NULL
        RETURNING 1
    )
    SELECT COUNT(*) INTO records_inserted FROM inserted_rows;

    INSERT INTO admin.merge_logs (table_name, merge_time, records_updated, records_inserted)
    VALUES ('customers', CURRENT_TIMESTAMP, records_updated, records_inserted);

    WITH updated_orders AS (
        UPDATE production.orders p
        SET customer_id = s.customer_id,
            amount      = s.amount,
            order_date  = s.order_date,
            updated_at  = s.updated_at
        FROM staging.orders s
        WHERE p.order_id = s.order_id
          AND s.updated_at > p.updated_at
        RETURNING p.order_id
    )
    SELECT COUNT(*) INTO records_updated FROM updated_orders;

    WITH inserted_orders AS (
        INSERT INTO production.orders
            (order_id, customer_id, amount, order_date, updated_at)
        SELECT s.order_id, s.customer_id, s.amount, s.order_date, s.updated_at
        FROM staging.orders s
        LEFT JOIN production.orders p ON s.order_id = p.order_id
        WHERE p.order_id IS NULL
        RETURNING 1
    )
    SELECT COUNT(*) INTO records_inserted FROM inserted_orders;

    INSERT INTO admin.merge_logs (table_name, merge_time, records_updated, records_inserted)
    VALUES ('orders', CURRENT_TIMESTAMP, records_updated, records_inserted);

    UPDATE staging.customers SET _dms_processing_status = 'PROCESSED';
    UPDATE staging.orders     SET _dms_processing_status = 'PROCESSED';

    INSERT INTO admin.merge_summary (merge_start, merge_end, status, notes)
    VALUES (merge_start_time, CURRENT_TIMESTAMP, 'SUCCESS',
            'Merged Oracle data from the requested time window');
END;
$$;

Two things about this procedure are deliberate and are worth understanding before it is edited.

There is no COMMIT and no ROLLBACK. PL/pgSQL permits transaction control inside a procedure, but not inside a block that carries an exception handler: such a block is implemented as a subtransaction, so PostgreSQL raises invalid transaction termination (SQLSTATE 2D000) on the first COMMIT. A version of this procedure that wraps its work in BEGIN … COMMIT; … EXCEPTION WHEN OTHERS THEN ROLLBACK; … END cannot run at all. Leaving transaction control to the caller gives the required all-or-nothing behaviour for free: the CALL commits on success and rolls back entirely on failure.

There is no EXCEPTION block either, because it would not achieve what it looks like it achieves. A failure row inserted into admin.merge_summary by a handler that then re-raises is rolled back along with everything else. Recording a failure durably means writing it from outside the failed transaction — which is what the orchestration script below does when psql returns non-zero.

Monitoring view

CREATE OR REPLACE VIEW admin.oracle_pg_migration_status AS
SELECT
    'customers' AS table_name,
    COUNT(*) AS total_records,
    COUNT(*) FILTER (WHERE _dms_processing_status = 'NEW')       AS pending_records,
    COUNT(*) FILTER (WHERE _dms_processing_status = 'PROCESSED') AS processed_records,
    COUNT(*) FILTER (WHERE _dms_processing_status = 'ERROR')     AS error_records
FROM staging.customers
UNION ALL
SELECT
    'orders',
    COUNT(*),
    COUNT(*) FILTER (WHERE _dms_processing_status = 'NEW'),
    COUNT(*) FILTER (WHERE _dms_processing_status = 'PROCESSED'),
    COUNT(*) FILTER (WHERE _dms_processing_status = 'ERROR')
FROM staging.orders;

Step 2: configure the DMS time-window task

Task settings

Save as oracle-pg-task-settings.json. The target schema is staging, and TargetTablePrepMode is DO_NOTHING because the staging tables were created deliberately above.

{
  "TargetMetadata": {
    "TargetSchema": "staging",
    "SupportLobs": true,
    "FullLobMode": false,
    "LobChunkSize": 64,
    "LimitedSizeLobMode": true,
    "LobMaxSize": 32
  },
  "FullLoadSettings": {
    "TargetTablePrepMode": "DO_NOTHING",
    "CreatePkAfterFullLoad": false,
    "StopTaskCachedChangesApplied": false,
    "StopTaskCachedChangesNotApplied": false,
    "MaxFullLoadSubTasks": 8,
    "TransactionConsistencyTimeout": 600,
    "CommitRate": 10000
  },
  "Logging": {
    "EnableLogging": true,
    "LogComponents": [
      { "Id": "SOURCE_UNLOAD",  "Severity": "LOGGER_SEVERITY_DEFAULT" },
      { "Id": "TARGET_LOAD",    "Severity": "LOGGER_SEVERITY_DEFAULT" },
      { "Id": "TRANSFORMATION", "Severity": "LOGGER_SEVERITY_DEFAULT" }
    ]
  },
  "OracleSettings": {
    "ReadTableSpaceName": false,
    "EnableHomogenousTablespace": false,
    "StandbyDelayTime": 0,
    "ArchivedLogsOnly": false,
    "ArchivedLogDestId": 0,
    "UseLogminerReader": true,
    "SecurityDbEncryption": false,
    "DirectPathNoLog": false,
    "AllowSelectNestedTables": true,
    "ConvertTimestampWithZoneToUTC": true,
    "NumberDataTypeScale": 38,
    "CharLengthSemantics": "CHAR",
    "AddSupplementalLogging": true,
    "ReadAheadBlocks": 1000
  }
}

UseLogminerReader: true selects LogMiner, which is AWS’s general recommendation for Oracle sources. Binary Reader is the alternative, and is preferred where the redo volume is high, where several tasks read the same source, or on Oracle RAC — see GoldenGate to PostgreSQL with DMS for the trade-off and the endpoint attributes it needs. This is a full-load task, so the choice matters less here than it does for the CDC task running alongside it.

Table mappings

Save as oracle-pg-time-window-mapping.json, adjusting the dates. The selection rules carry the time filter; the transformation rules pin the Oracle types that do not map cleanly.

{
  "rules": [
    {
      "rule-type": "selection",
      "rule-id": "1",
      "rule-name": "customers-time-window",
      "object-locator": { "schema-name": "SOURCE_SCHEMA", "table-name": "CUSTOMERS" },
      "rule-action": "include",
      "filters": [
        {
          "filter-type": "source",
          "column-name": "UPDATED_AT",
          "filter-conditions": [
            { "filter-operator": "gte", "value": "2025-01-01 00:00:00" },
            { "filter-operator": "lte", "value": "2025-01-31 23:59:59" }
          ]
        }
      ]
    },
    {
      "rule-type": "selection",
      "rule-id": "2",
      "rule-name": "orders-time-window",
      "object-locator": { "schema-name": "SOURCE_SCHEMA", "table-name": "ORDERS" },
      "rule-action": "include",
      "filters": [
        {
          "filter-type": "source",
          "column-name": "UPDATED_AT",
          "filter-conditions": [
            { "filter-operator": "gte", "value": "2025-01-01 00:00:00" },
            { "filter-operator": "lte", "value": "2025-01-31 23:59:59" }
          ]
        }
      ]
    },
    {
      "rule-type": "transformation",
      "rule-id": "3",
      "rule-name": "convert-oracle-timestamp",
      "rule-action": "change-data-type",
      "rule-target": "column",
      "object-locator": {
        "schema-name": "SOURCE_SCHEMA",
        "table-name": "%",
        "column-name": "%_AT"
      },
      "data-type": { "type": "datetime", "precision": 0, "scale": 0 }
    },
    {
      "rule-type": "transformation",
      "rule-id": "4",
      "rule-name": "convert-oracle-number",
      "rule-action": "change-data-type",
      "rule-target": "column",
      "object-locator": {
        "schema-name": "SOURCE_SCHEMA",
        "table-name": "ORDERS",
        "column-name": "AMOUNT"
      },
      "data-type": { "type": "numeric", "precision": 10, "scale": 2 }
    }
  ]
}

The action for a type change is change-data-type, and it is only valid with a rule-target of column. convert-column-type is not a DMS rule action; a mapping document containing it is rejected when the task is created, so nothing downstream runs. Table and column names are case-sensitive in transformation rules and must be given in upper case for an Oracle source.

Step 3: run the orchestration

Save as oracle-pg-time-window-reload.sh. The database password is fetched from Secrets Manager at run time rather than being written into the script or passed on a command line, where it would be visible to any local ps and would land in shell history.

#!/bin/bash
set -euo pipefail

SOURCE_ENDPOINT_ARN="arn:aws:dms:region:account:endpoint:oracle-source-endpoint"
TARGET_ENDPOINT_ARN="arn:aws:dms:region:account:endpoint:postgres-target-endpoint"
REPLICATION_INSTANCE_ARN="arn:aws:dms:region:account:rep:replication-instance-id"
TABLE_MAPPINGS_FILE="oracle-pg-time-window-mapping.json"
TASK_SETTINGS_FILE="oracle-pg-task-settings.json"
START_DATE="2025-01-01T00:00:00"
END_DATE="2025-01-31T23:59:59"
PG_HOST="postgres-host"
PG_PORT="5432"
PG_NAME="postgres-db"
PG_USER="merge_user"
PG_SECRET_ID="prod/postgres/merge_user"

# Assign first, then export: `export VAR=$(...)` hides a failing command substitution
# from `set -e`, because export itself succeeds.
PGPASSWORD=$(aws secretsmanager get-secret-value \
  --secret-id "$PG_SECRET_ID" --query SecretString --output text \
  | python3 -c 'import json,sys; print(json.load(sys.stdin)["password"])')
export PGPASSWORD

echo "Reloading $START_DATE to $END_DATE from Oracle into PostgreSQL staging"

TASK_ARN=$(aws dms create-replication-task \
  --replication-task-identifier "oracle-pg-historical-reload-$(date +%Y%m%d%H%M%S)" \
  --source-endpoint-arn "$SOURCE_ENDPOINT_ARN" \
  --target-endpoint-arn "$TARGET_ENDPOINT_ARN" \
  --replication-instance-arn "$REPLICATION_INSTANCE_ARN" \
  --migration-type "full-load" \
  --table-mappings "file://$TABLE_MAPPINGS_FILE" \
  --replication-task-settings "file://$TASK_SETTINGS_FILE" \
  --tags Key=Purpose,Value=OraclePgHistoricalReload \
         Key=TimeWindow,Value="${START_DATE}-${END_DATE}" \
  --query 'ReplicationTask.ReplicationTaskArn' \
  --output text)

echo "Created replication task: $TASK_ARN"

aws dms start-replication-task \
  --replication-task-arn "$TASK_ARN" \
  --start-replication-task-type start-replication

while true; do
  STATUS=$(aws dms describe-replication-tasks \
    --filters Name=replication-task-arn,Values="$TASK_ARN" \
    --query 'ReplicationTasks[0].Status' --output text)

  PERCENT=$(aws dms describe-replication-tasks \
    --filters Name=replication-task-arn,Values="$TASK_ARN" \
    --query 'ReplicationTasks[0].ReplicationTaskStats.FullLoadProgressPercent' \
    --output text || echo "N/A")

  echo "$(date -u +%H:%M:%S) status=$STATUS progress=${PERCENT}%"

  if [ "$STATUS" = "stopped" ] || [ "$STATUS" = "failed" ]; then
    break
  fi
  sleep 120
done

if [ "$STATUS" != "stopped" ]; then
  echo "Historical load did not complete; leaving staging untouched"
  exit 1
fi

echo "Load complete. Merging staging into production."
if psql -h "$PG_HOST" -p "$PG_PORT" -U "$PG_USER" -d "$PG_NAME" \
        -v ON_ERROR_STOP=1 -c "CALL merge_time_window_data();"; then
  echo "Merge completed"
else
  # The merge rolled back in full, including any log rows it wrote.
  # Record the failure from outside that transaction.
  psql -h "$PG_HOST" -p "$PG_PORT" -U "$PG_USER" -d "$PG_NAME" -c \
    "INSERT INTO admin.merge_summary (merge_start, merge_end, status, notes)
     VALUES (CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, 'FAILED',
             'merge_time_window_data() failed; see psql output and PostgreSQL logs');"
  exit 1
fi

Truncating the staging tables afterwards is a separate, deliberate step rather than part of the run — keeping them until the merge has been verified is what makes a second attempt possible.

psql -h "$PG_HOST" -p "$PG_PORT" -U "$PG_USER" -d "$PG_NAME" \
  -c "TRUNCATE TABLE staging.customers; TRUNCATE TABLE staging.orders;"

Step 4: verify

Check the load and the merge:

SELECT * FROM admin.oracle_pg_migration_status;
SELECT * FROM admin.merge_logs    ORDER BY merge_time DESC;
SELECT * FROM admin.merge_summary ORDER BY merge_start DESC;

Then check that the staging tables and the production tables agree on types, which is where the Oracle to PostgreSQL mapping failures show up:

CREATE OR REPLACE VIEW admin.oracle_pg_type_verification AS
WITH target_columns AS (
    SELECT c.table_name, c.column_name, c.data_type,
           c.character_maximum_length, c.numeric_precision, c.numeric_scale
    FROM information_schema.tables t
    JOIN information_schema.columns c
      ON t.table_schema = c.table_schema AND t.table_name = c.table_name
    WHERE t.table_schema = 'production' AND t.table_type = 'BASE TABLE'
),
staging_columns AS (
    SELECT c.table_name, c.column_name, c.data_type,
           c.character_maximum_length, c.numeric_precision, c.numeric_scale
    FROM information_schema.tables t
    JOIN information_schema.columns c
      ON t.table_schema = c.table_schema AND t.table_name = c.table_name
    WHERE t.table_schema = 'staging' AND t.table_type = 'BASE TABLE'
)
SELECT
    t.table_name,
    t.column_name,
    t.data_type AS target_data_type,
    s.data_type AS staging_data_type,
    CASE
        WHEN t.data_type <> s.data_type THEN 'DATA TYPE MISMATCH'
        WHEN t.character_maximum_length IS DISTINCT FROM s.character_maximum_length
             AND t.character_maximum_length IS NOT NULL THEN 'LENGTH MISMATCH'
        WHEN t.numeric_precision IS DISTINCT FROM s.numeric_precision
             AND t.numeric_precision IS NOT NULL THEN 'PRECISION MISMATCH'
        WHEN t.numeric_scale IS DISTINCT FROM s.numeric_scale
             AND t.numeric_scale IS NOT NULL THEN 'SCALE MISMATCH'
        ELSE 'OK'
    END AS status
FROM target_columns t
JOIN staging_columns s
  ON t.table_name = s.table_name AND t.column_name = s.column_name
WHERE t.data_type IS DISTINCT FROM s.data_type
   OR (t.character_maximum_length IS DISTINCT FROM s.character_maximum_length
       AND t.character_maximum_length IS NOT NULL)
   OR (t.numeric_precision IS DISTINCT FROM s.numeric_precision
       AND t.numeric_precision IS NOT NULL)
   OR (t.numeric_scale IS DISTINCT FROM s.numeric_scale
       AND t.numeric_scale IS NOT NULL);

SELECT * FROM admin.oracle_pg_type_verification;

Troubleshooting

The task fails to create. The table-mapping document is validated as a whole. An invalid rule action or a mistyped rule target rejects the entire file.

The task fails during the load. Check the CloudWatch log group for the task. The usual causes are missing supplemental logging on a source table, insufficient disk on the target, and an Oracle type with no clean PostgreSQL equivalent.

Type conversion problems. Oracle NUMBER without explicit precision and scale needs a change-data-type rule to land as a defined NUMERIC. Oracle DATE carries a time component and is not a PostgreSQL date. Character-set differences between the two databases surface as string length errors rather than as encoding errors.

The merge fails. Look at admin.merge_summary and the psql output together — the summary row written by the orchestration script survives, the one the procedure would have written does not. The recurring causes are a primary key that exists in staging but not in production, and constraint violations on rows that were valid in Oracle.

Considerations

• MaxFullLoadSubTasks can be raised for large tables; the ceiling is the source’s tolerance for concurrent reads.
• Raising work_mem and maintenance_work_mem on the target for the duration of the load is usually worth more than any DMS-side tuning.
• Check that the time-window task and the ongoing CDC task are not writing the same rows at the same time. Staging tables keep them apart until the merge, which is the point of the design.
• Back up the production tables before the merge. The merge is all-or-nothing within its transaction, but a successful merge of the wrong window is not something a transaction protects against.
• Record which window was reloaded, and why, somewhere that outlives the ticket.