Function workflow_migration

test/pg-cdc-old-syntax/mzcompose.py:422–490

(c: Composition, parser: WorkflowArgumentParser)

def workflow_migration(c: Composition, parser: WorkflowArgumentParser) -> None:
    parser.add_argument(
        "filter",
        nargs="*",
        default=["*.td"],
        help="limit to only the files matching filter",
    )
    args = parser.parse_args()

    sharded_files = get_sharded_files(args.filter)
    print(f"Files: {sharded_files}")

    ssl_args_dict = get_testdrive_ssl_args(c)
    testdrive_ssl_args = ssl_args_dict["testdrive_args"]
    volumes_extra = ssl_args_dict["volumes_extra"]

    testdrive_args = (
        testdrive_ssl_args + get_default_testdrive_size_args() + ["--no-reset"]
    )

    pg_version = get_targeted_pg_version(parser)

    for file in sharded_files:
        mz_old = Materialized(
            name="materialized",
            volumes_extra=volumes_extra,
            external_metadata_store=True,
            external_blob_store=True,
            additional_system_parameter_defaults={
                "log_filter": "mz_storage::source::postgres=trace,info"
            },
            default_replication_factor=2,
        )

        mz_new = Materialized(
            name="materialized",
            volumes_extra=volumes_extra,
            external_metadata_store=True,
            external_blob_store=True,
            additional_system_parameter_defaults={
                "log_filter": "mz_storage::source::postgres=trace,info",
                "force_source_table_syntax": "true",
            },
            default_replication_factor=2,
        )
        with c.override(mz_old, create_postgres(pg_version=pg_version)):
            c.up("materialized", "test-certs", "postgres")

            print(f"Running {file} with mz_old")

            c.run_testdrive_files(
                *testdrive_args,
                file,
            )
            c.kill("materialized", wait=True)

            with c.override(mz_new):
                c.up("materialized")