storage/adapter: Opt-in migration of sources to the new table model

ci/nightly/pipeline.template.yml

@@ -568,6 +568,17 @@ steps:
         agents:
           queue: hetzner-aarch64-4cpu-8gb
 
+      - id: mysql-cdc-migration
+        label: MySQL CDC source-versioning migration tests
+        depends_on: build-aarch64
+        timeout_in_minutes: 360
+        plugins:
+          - ./ci/plugins/mzcompose:
+              composition: mysql-cdc-old-syntax
+              run: migration
+        agents:
+          queue: hetzner-aarch64-4cpu-8gb
+
       - id: mysql-cdc-resumption-old-syntax
         label: MySQL CDC resumption tests (before source versioning)
         depends_on: build-aarch64
@@ -599,6 +610,17 @@ steps:
           queue: hetzner-aarch64-4cpu-8gb
         # the mzbuild postgres version will be used, which depends on the Dockerfile specification
 
+      - id: pg-cdc-migration
+        label: Postgres CDC source-versioning migration tests
+        depends_on: build-aarch64
+        timeout_in_minutes: 360
+        plugins:
+          - ./ci/plugins/mzcompose:
+              composition: pg-cdc-old-syntax
+              run: migration
+        agents:
+          queue: hetzner-aarch64-4cpu-8gb
+
       - id: pg-cdc-resumption-old-syntax
         label: Postgres CDC resumption tests (before source versioning)
         depends_on: build-aarch64
@@ -629,6 +651,17 @@ steps:
         agents:
           queue: hetzner-aarch64-8cpu-16gb
 
+      - id: testdrive-kafka-migration
+        label: "Testdrive %N migration tests"
+        depends_on: build-aarch64
+        timeout_in_minutes: 180
+        parallelism: 8
+        plugins:
+          - ./ci/plugins/mzcompose:
+              composition: testdrive-old-kafka-src-syntax
+              run: migration
+        agents:
+          queue: hetzner-aarch64-8cpu-16gb
 
   - group: AWS
     key: aws

misc/python/materialize/checks/all_checks/source_tables.py

@@ -16,7 +16,6 @@
 
 
 class TableFromSourceBase(Check):
-
     def _can_run(self, e: Executor) -> bool:
         return self.base_version >= MzVersion.parse_mz("v0.116.0-dev")
 
@@ -34,7 +33,6 @@ class TableFromPgSource(TableFromSourceBase):
     suffix = "tbl_from_pg_source"
 
     def initialize(self) -> Testdrive:
-
         return Testdrive(
             self.generic_setup()
             + dedent(

misc/python/materialize/checks/all_checks/upsert.py

@@ -179,3 +179,59 @@ def validate(self) -> Testdrive:
            """
             )
         )
+
+
+class UpsertLegacy(Check):
+    """
+    An upsert source test that uses the legacy syntax to create the source
+    on all versions to ensure the source is properly migrated with the
+    ActivateSourceVersioningMigration scenario
+    """
+
+    def initialize(self) -> Testdrive:
+        return Testdrive(
+            schemas()
+            + dedent(
+                """
+                $ kafka-create-topic topic=upsert-legacy-syntax
+
+                $ kafka-ingest format=avro key-format=avro topic=upsert-legacy-syntax key-schema=${keyschema} schema=${schema} repeat=10000
+                {"key1": "A${kafka-ingest.iteration}"} {"f1": "A${kafka-ingest.iteration}"}
+
+                > CREATE SOURCE upsert_insert_legacy
+                  FROM KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-upsert-legacy-syntax-${testdrive.seed}')
+                  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn
+                  ENVELOPE UPSERT
+
+                > CREATE MATERIALIZED VIEW upsert_insert_legacy_view AS SELECT COUNT(DISTINCT key1 || ' ' || f1) FROM upsert_insert_legacy;
+                """
+            )
+        )
+
+    def manipulate(self) -> list[Testdrive]:
+        return [
+            Testdrive(schemas() + dedent(s))
+            for s in [
+                """
+                $ kafka-ingest format=avro key-format=avro topic=upsert-legacy-syntax key-schema=${keyschema} schema=${schema} repeat=10000
+                {"key1": "A${kafka-ingest.iteration}"} {"f1": "A${kafka-ingest.iteration}"}
+                """,
+                """
+                $ kafka-ingest format=avro key-format=avro topic=upsert-legacy-syntax key-schema=${keyschema} schema=${schema} repeat=10000
+                {"key1": "A${kafka-ingest.iteration}"} {"f1": "A${kafka-ingest.iteration}"}
+                """,
+            ]
+        ]
+
+    def validate(self) -> Testdrive:
+        return Testdrive(
+            dedent(
+                """
+                > SELECT COUNT(*), COUNT(DISTINCT key1), COUNT(DISTINCT f1) FROM upsert_insert_legacy
+                10000 10000 10000
+
+                > SELECT * FROM upsert_insert_legacy_view;
+                10000
+           """
+            )
+        )

misc/python/materialize/checks/scenarios_upgrade.py

@@ -391,3 +391,49 @@ def actions(self) -> list[Action]:
             ),
             Validate(self),
         ]
+
+
+class ActivateSourceVersioningMigration(Scenario):
+    """
+    Starts MZ, initializes and manipulates, then forces the migration
+    of sources to the new table model (introducing Source Versioning).
+    """
+
+    def base_version(self) -> MzVersion:
+        return get_last_version()
+
+    def actions(self) -> list[Action]:
+        print(f"Upgrading from tag {self.base_version()}")
+        return [
+            StartMz(
+                self,
+                tag=self.base_version(),
+            ),
+            Initialize(self),
+            Manipulate(self, phase=1),
+            KillMz(
+                capture_logs=True
+            ),  #  We always use True here otherwise docker-compose will lose the pre-upgrade logs
+            StartMz(
+                self,
+                tag=None,
+                # Activate the `force_source_table_syntax` flag
+                # which should trigger the migration of sources
+                # using the old syntax to the new table model.
+                additional_system_parameter_defaults={
+                    "force_source_table_syntax": "true",
+                },
+            ),
+            Manipulate(self, phase=2),
+            Validate(self),
+            # A second restart while already on the new version
+            KillMz(capture_logs=True),
+            StartMz(
+                self,
+                tag=None,
+                additional_system_parameter_defaults={
+                    "force_source_table_syntax": "true",
+                },
+            ),
+            Validate(self),
+        ]

misc/python/materialize/cli/ci_annotate_errors.py

@@ -98,6 +98,8 @@
     | (FAIL|TIMEOUT)\s+\[\s*\d+\.\d+s\]
     # parallel-workload
     | worker_.*\ still\ running: [\s\S]* Threads\ have\ not\ stopped\ within\ 5\ minutes,\ exiting\ hard
+    # source-table migration
+    | source-table-migration\ issue
     )
     .* $
     """,

misc/python/materialize/mzcompose/__init__.py

@@ -53,7 +53,9 @@
 
 
 def get_default_system_parameters(
-    version: MzVersion | None = None, zero_downtime: bool = False
+    version: MzVersion | None = None,
+    zero_downtime: bool = False,
+    force_source_table_syntax: bool = False,
 ) -> dict[str, str]:
     """For upgrade tests we only want parameters set when all environmentd /
     clusterd processes have reached a specific version (or higher)
@@ -118,6 +120,7 @@ def get_default_system_parameters(
         "enable_table_keys": "true",
         "enable_variadic_left_join_lowering": "true",
         "enable_worker_core_affinity": "true",
+        "force_source_table_syntax": "true" if force_source_table_syntax else "false",
         "persist_batch_columnar_format": (
             "both_v2" if version >= MzVersion.parse_mz("v0.112.0-dev") else "row"
         ),

misc/python/materialize/source_table_migration.py

@@ -0,0 +1,71 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+"""Utilities for testing the source table migration"""
+from materialize.mz_version import MzVersion
+from materialize.mzcompose.composition import Composition
+
+
+def verify_sources_after_source_table_migration(
+    c: Composition, file: str, fail: bool = False
+) -> None:
+    source_names_rows = c.sql_query(
+        "SELECT sm.name || '.' || src.name FROM mz_sources src INNER JOIN mz_schemas sm ON src.schema_id = sm.id WHERE src.id LIKE 'u%';"
+    )
+    source_names = [row[0] for row in source_names_rows]
+
+    print(f"Sources created in {file} are: {source_names}")
+
+    c.sql("SET statement_timeout = '20s'")
+
+    for source_name in source_names:
+        _verify_source(c, file, source_name, fail=fail)
+
+
+def _verify_source(
+    c: Composition, file: str, source_name: str, fail: bool = False
+) -> None:
+    try:
+        print(f"Checking source: {source_name}")
+
+        # must not crash
+        statement = f"SELECT count(*) FROM {source_name};"
+        print(statement)
+        c.sql_query(statement)
+
+        statement = f"SHOW CREATE SOURCE {source_name};"
+        print(statement)
+        result = c.sql_query(statement)
+        sql = result[0][1]
+        assert "FOR TABLE" not in sql, f"FOR TABLE found in: {sql}"
+        assert "FOR ALL TABLES" not in sql, f"FOR ALL TABLES found in: {sql}"
+
+        if not source_name.endswith("_progress"):
+            assert "CREATE SUBSOURCE" not in sql, f"CREATE SUBSOURCE found in: {sql}"
+
+        print("OK.")
+    except Exception as e:
+        print(f"source-table-migration issue in {file}: {str(e)}")
+
+        if fail:
+            raise e
+
+
+def check_source_table_migration_test_sensible() -> None:
+    assert MzVersion.parse_cargo() < MzVersion.parse_mz(
+        "v0.130.0"
+    ), "migration test probably no longer needed"
+
+
+def get_old_image_for_source_table_migration_test() -> str:
+    return "materialize/materialized:v0.122.0"
+
+
+def get_new_image_for_source_table_migration_test() -> str | None:
+    return None