Also test the migration in the legacy upgrade tests

Author: rjobanp

misc/python/materialize/mzcompose/__init__.py

@@ -49,7 +49,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)
@@ -113,6 +115,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"
         ),

test/legacy-upgrade/mzcompose.py

@@ -108,20 +108,56 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
         else:
             if parallelism_count == 1 or parallelism_index == 0:
                 test_upgrade_from_version(
-                    c, f"{version}", priors, filter=args.filter, zero_downtime=True
+                    c,
+                    f"{version}",
+                    priors,
+                    filter=args.filter,
+                    zero_downtime=True,
+                    force_source_table_syntax=False,
                 )
             if parallelism_count == 1 or parallelism_index == 1:
                 test_upgrade_from_version(
-                    c, f"{version}", priors, filter=args.filter, zero_downtime=False
+                    c,
+                    f"{version}",
+                    priors,
+                    filter=args.filter,
+                    zero_downtime=False,
+                    force_source_table_syntax=False,
+                )
+                test_upgrade_from_version(
+                    c,
+                    f"{version}",
+                    priors,
+                    filter=args.filter,
+                    zero_downtime=False,
+                    force_source_table_syntax=True,
                 )
 
     if parallelism_count == 1 or parallelism_index == 0:
         test_upgrade_from_version(
-            c, "current_source", priors=[], filter=args.filter, zero_downtime=True
+            c,
+            "current_source",
+            priors=[],
+            filter=args.filter,
+            zero_downtime=True,
+            force_source_table_syntax=False,
         )
     if parallelism_count == 1 or parallelism_index == 1:
         test_upgrade_from_version(
-            c, "current_source", priors=[], filter=args.filter, zero_downtime=False
+            c,
+            "current_source",
+            priors=[],
+            filter=args.filter,
+            zero_downtime=False,
+            force_source_table_syntax=False,
+        )
+        test_upgrade_from_version(
+            c,
+            "current_source",
+            priors=[],
+            filter=args.filter,
+            zero_downtime=False,
+            force_source_table_syntax=True,
         )
 
 
@@ -144,13 +180,14 @@ def test_upgrade_from_version(
     priors: list[MzVersion],
     filter: str,
     zero_downtime: bool,
+    force_source_table_syntax: bool,
 ) -> None:
     print(
         f"+++ Testing {'0dt upgrade' if zero_downtime else 'regular upgrade'} from Materialize {from_version} to current_source."
     )
 
     system_parameter_defaults = get_default_system_parameters(
-        zero_downtime=zero_downtime
+        zero_downtime=zero_downtime,
     )
     deploy_generation = 0
 
@@ -289,6 +326,13 @@ def test_upgrade_from_version(
                     c.rm(mz_service)
 
     print(f"{'0dt-' if zero_downtime else ''}Upgrading to final version")
+    system_parameter_defaults = get_default_system_parameters(
+        zero_downtime=zero_downtime,
+        # We can only force the syntax on the final version so that the migration to convert
+        # sources to the new model can be applied without preventing sources from being
+        # created in the old syntax on the older version.
+        force_source_table_syntax=force_source_table_syntax,
+    )
     mz_to = Materialized(
         name=mz_service,
         options=list(mz_options.values()),