CDC レコードについて

Aurora DSQL CDC は、各変更を JSON レコードとして配信します。レコードは、オペレーションタイプ、前後の行イメージ、ソースメタデータを含むエンベロープ構造を使用します。

レコードが Amazon Kinesis にマッピングされる方法

Aurora DSQL は、各 CDC レコードを単一の Kinesis レコードとして書き込みます。Kinesis レコードの Data フィールドには、JSON ペイロードが含まれています。Aurora DSQL は、ランダム化された Kinesis パーティションキーを使用して CDC レコードをシャード間で均等に分散します。すべての変更を読み取るには、Kinesis データストリームのすべてのシャードを消費します。レコードが Kinesis レコードのサイズ制限を超えると、Aurora DSQL はそれを複数の Kinesis レコードに分割します。詳細については、「サイズの大きいレコードの処理」を参照してください。

ペイロードのプライマリキー

プライマリキーを持つテーブルの場合、プライマリキー列の値がペイロードに表示されます。

例えば、複合プライマリキーを持つテーブルがあるとします。

CREATE TABLE order_items (
    order_id INT,
    item_id INT,
    quantity INT,
    price NUMERIC,
    PRIMARY KEY (order_id, item_id)
);

このテーブルを削除すると、"before": {"order_id": 1001, "item_id": 42} というペイロードが生成されます。

レコードのペイロード

ペイロードは次の JSON エンベロープ形式を使用します。

INSERT の例

次の例は、挿入オペレーションの CDC レコードを示しています。

{
    "type": "full",
    "op": "c",
    "before": null,
    "after": {"order_id": 1001, "item_id": 42, "quantity": 5, "price": "29.99"},
    "source": {
        "version": "1.0",
        "ts_ms": 1705318200000,
        "ts_ns": 1705318200000000000,
        "txId": "ffthunp5stx6ffs2vyfqoatmfu",
        "schema": "public",
        "table": "order_items",
        "db": "postgres",
        "cluster": "kmabugltfmjdaj2siqr2qbxgju"
    },
    "ts_ms": 1705318200125,
    "ts_ns": 1705318200125483291
}

UPDATE の例

次の例は、Aurora DSQL が op: "u" の出力を開始した後、UPDATE ステートメントによって生成された CDC レコードがどのようになるかを示しています。

{
    "type": "full",
    "op": "u",
    "before": null,
    "after": {"order_id": 1001, "item_id": 42, "quantity": 10, "price": "29.99"},
    "source": {
        "version": "1.0",
        "ts_ms": 1705318300000,
        "ts_ns": 1705318300000000000,
        "txId": "qvtiesgmd55cvlfukm3dfuotji",
        "schema": "public",
        "table": "order_items",
        "db": "postgres",
        "cluster": "kmabugltfmjdaj2siqr2qbxgju"
    },
    "ts_ms": 1705318300125,
    "ts_ns": 1705318300125483291
}

DELETE の例

プライマリキーを持つテーブルの削除の場合、before フィールドには削除された行のプライマリキー値が含まれます。

{
    "type": "full",
    "op": "d",
    "before": {"order_id": 1001, "item_id": 42},
    "after": null,
    "source": {
        "version": "1.0",
        "ts_ms": 1705318400000,
        "ts_ns": 1705318400000000000,
        "txId": "xyzabc123def456ghi789jklmno",
        "schema": "public",
        "table": "order_items",
        "db": "postgres",
        "cluster": "kmabugltfmjdaj2siqr2qbxgju"
    },
    "ts_ms": 1705318400125,
    "ts_ns": 1705318400125483291
}

ペイロードフィールド

フォーマットの詳細

以下の詳細では、Aurora DSQL CDC がレコードをフォーマットする方法について説明します。これらの動作に対応できるようにアプリケーションを設計します。

サイズの大きいレコードの処理

CDC レコードのシリアル化された JSON が 9 MiB を超えると、Aurora DSQL は before および/または after イメージを分割し、複数の Kinesis レコードを配信します。各レコードには、その構造を示す最上位の type フィールドが含まれています。完全なレコードの場合は full、フラグメントを参照するメインレコードの場合は chunked、チャンクイメージの個々の部分の場合は fragment です。チャンク化されたメインレコードの op、source、ts_ms、および ts_ns フィールドは、完全なレコードの場合と同じように動作します。単一の Kinesis レコードに収まるレコードは type が full に設定されており、追加の処理は必要ありません。

chunk_id は再試行間で安定しています。Aurora DSQL がフラグメントを再配信する場合、元の配信と同じ chunk_id を保持するため、アプリケーションは以前の試行の部分的なセットを処理せずに、同じ識別子でバッファリングを続行できます。

メインレコード

チャンク化されたメインレコードは、分割されたイメージの最上位の before または after フィールドを、再構成する方法を説明する chunked オブジェクトに置き換えます。chunked の各エントリには、chunk_id (フラグメントをこのレコードにリンクする識別子)、total_fragments (そのイメージを構成するフラグメントの数)、crc32c (再アセンブルされたイメージテキストに対する 10 進文字列の CRC32C チェックサム) があります。1 つのイメージがインラインで、もう 1 つのイメージがチャンク化されている場合、インラインイメージは依然として最上位レベルに値または null として表示されます。

{
    "type": "chunked",
    "op": "c",
    "before": null,
    "after": null,
    "source": {
        "version": "1.0",
        "ts_ms": 1705318200000,
        "ts_ns": 1705318200000000000,
        "txId": "ffthunp5stx6ffs2vyfqoatmfu",
        "schema": "public",
        "table": "order_items",
        "db": "postgres",
        "cluster": "cluster-id"
    },
    "chunked": {
        "after": {
            "chunk_id": "chunk-id",
            "total_fragments": 3,
            "crc32c": "2073618257"
        }
    },
    "ts_ms": 1705318200125,
    "ts_ns": 1705318200125483291
}

フラグメントレコード

各フラグメントは、type が fragment に設定され、3 つのフィールドを持つ独自の Kinesis レコードです。chunk_id は、メインレコードの対応する chunked.before.chunk_id または chunked.after.chunk_id の値と一致し、index はイメージ内のフラグメントのゼロベースの位置であり、data は UTF-8 文字境界で分割されたイメージの JSON テキストのセグメントです (各フラグメントの data 値は、それ自体が有効な UTF-8 文字列です)。Aurora DSQL CDC は UNORDERED モードとランダム化されたパーティションキーを使用するため、フラグメントとメインレコードは異なるシャードに任意の順序で到着する可能性があります。すべてのフラグメントを読み取るには、Kinesis データストリーム上のすべてのシャードを消費します。配信の順序付けの詳細については、「順序付け」を参照してください。

{
    "type": "fragment",
    "chunk_id": "chunk-id",
    "index": 0,
    "data": "partial-JSON-text"
}

サイズの大きいイメージを再構成するには、type が fragment の各レコードを chunk_id でバッファします。type が chunked のメインレコードを受け取ったら、chunked.before または chunked.after で参照される各 chunk_id に対して、total_fragments のフラグメントが揃うまで待機し、フラグメントを index の昇順にソートして、data 文字列を連結します。連結結果は、元の before または after オブジェクトを JSON テキストとして表したものです。これを解析して列の値にアクセスします。配信の整合性を検証するには、連結された文字列に対して CRC32C を計算し、結果を chunked.before.crc32c または chunked.after.crc32c と比較します。

データ型のシリアル化

次の表は、Aurora DSQL が CDC レコードの各 PostgreSQL データ型をシリアル化する方法を示しています。

整数タイプ

JavaScript 環境では、bigint の値が ±2^53 を超えると精度が低下する可能性があります。このような場合は、BigInt または任意精度のライブラリを使用します。

浮動小数点型

ブール値

文字型

バイナリ

日付と時刻の型

その他の型

NULL 値

どのデータ型でも、NULL 列の値は JSON の null として表されます。

CDC レコードにおけるスキーマの進化

列の追加、削除、名前の変更などによってテーブルのスキーマを変更すると、CDC レコードには DDL の変更をコミットしたトランザクション以降の変更が反映されます。DDL の変更前にコミットされたトランザクションのレコードは、以前のスキーマを使用します。例えば、次のようになります。

各レコードの after および before フィールドに存在する列名を調べて、ダウンストリームのコンシューマーのスキーマの変更を追跡します。各レコードの source.version フィールドは、CDC エンベロープ形式を識別します。