diff --git a/ci/test/pipeline.template.yml b/ci/test/pipeline.template.yml
index b659ff74085a0..b1378dccdf26a 100644
--- a/ci/test/pipeline.template.yml
+++ b/ci/test/pipeline.template.yml
@@ -378,20 +378,38 @@ steps:
       # Too slow on aarch64
       queue: hetzner-x86-64-4cpu-8gb
 
-  - id: sql-server-cdc
-    label: "SQL Server CDC tests"
-    depends_on: build-x86_64
-    timeout_in_minutes: 30
-    inputs: [test/sql-server-cdc]
-    parallelism: 3
-    plugins:
-      - ./ci/plugins/mzcompose:
-          composition: sql-server-cdc
-    agents:
-      # The SQL Server Docker image isn't available on ARM.
-      #
-      # See: <https://github.com/microsoft/mssql-docker/issues/864>
-      queue: hetzner-x86-64-8cpu-16gb
+  - group: "SQL Server tests"
+    key: sql-server-tests
+    steps:
+      - id: sql-server-cdc
+        label: "SQL Server CDC tests"
+        depends_on: build-x86_64
+        timeout_in_minutes: 30
+        inputs: [test/sql-server-cdc]
+        parallelism: 3
+        plugins:
+          - ./ci/plugins/mzcompose:
+              composition: sql-server-cdc
+        agents:
+          # The SQL Server Docker image isn't available on ARM.
+          #
+          # See: <https://github.com/microsoft/mssql-docker/issues/864>
+          queue: hetzner-x86-64-8cpu-16gb
+
+      - id: sql-server-cdc-old-syntax
+        label: "SQL Server CDC old syntax tests"
+        depends_on: build-x86_64
+        timeout_in_minutes: 30
+        inputs: [test/sql-server-cdc-old-syntax]
+        parallelism: 6
+        plugins:
+          - ./ci/plugins/mzcompose:
+              composition: sql-server-cdc-old-syntax
+        agents:
+          # The SQL Server Docker image isn't available on ARM.
+          #
+          # See: <https://github.com/microsoft/mssql-docker/issues/864>
+          queue: hetzner-x86-64-8cpu-16gb
 
   - group: "Connection tests"
     key: connection-tests
diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py
index 71b519edd47b8..eab801a5ebd4c 100644
--- a/misc/python/materialize/mzcompose/__init__.py
+++ b/misc/python/materialize/mzcompose/__init__.py
@@ -110,6 +110,7 @@ def get_minimal_system_parameters(
         "enable_reduce_mfp_fusion": "true",
         "enable_refresh_every_mvs": "true",
         "enable_cluster_schedule_refresh": "true",
+        "enable_sql_server_source": "true",
         "enable_statement_lifecycle_logging": "true",
         "enable_compute_temporal_bucketing": "true",
         "enable_variadic_left_join_lowering": "true",
diff --git a/test/sql-server-cdc-old-syntax/alter-source.td b/test/sql-server-cdc-old-syntax/alter-source.td
new file mode 100644
index 0000000000000..b980e2c518175
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/alter-source.td
@@ -0,0 +1,393 @@
+# 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.
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+CREATE TABLE table_a (pk INTEGER PRIMARY KEY, f2 VARCHAR(128));
+INSERT INTO table_a VALUES (1, 'one');
+INSERT INTO table_a VALUES (2, 'two');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'table_a', @role_name = 'SA', @supports_net_changes = 0;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+> DROP SOURCE dummy;
+
+> SELECT * FROM table_a;
+1 one
+2 two
+
+$ sql-server-execute name=sql-server
+DROP TABLE table_a;
+
+! ALTER SOURCE mz_source ADD SUBSOURCE table_b;
+contains:reference to table_b not found in source
+
+# Adding a table with the same name as a running table does not allow you to add
+# the new table, even though its OID is the different.
+
+$ sql-server-execute name=sql-server
+CREATE TABLE table_a (pk INTEGER PRIMARY KEY, f2 VARCHAR(128));
+EXEC sys.sp_cdc_disable_table @source_schema = 'dbo', @source_name = 'table_a', @capture_instance = 'all';
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'table_a', @role_name = 'SA', @supports_net_changes = 0, @capture_instance = 'table_a2';
+INSERT INTO table_a VALUES (9, 'nine');
+
+# Current table_a is not new table_a. Note that this only works right now
+# because we are bad at detecting dropped tables.
+> SELECT * FROM table_a;
+1 one
+2 two
+
+# We are not aware that the new table_a is different
+!ALTER SOURCE mz_source ADD SUBSOURCE table_a;
+contains:catalog item 'table_a' already exists
+
+> DROP SOURCE mz_source CASCADE;
+
+# Re-populate tables for rest of test.
+
+$ sql-server-execute name=sql-server
+DELETE FROM table_a;
+INSERT INTO table_a VALUES (1, 'one');
+INSERT INTO table_a VALUES (2, 'two');
+
+CREATE TABLE table_b (pk INTEGER PRIMARY KEY, f2 VARCHAR(128));
+INSERT INTO table_b VALUES (1, 'one');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'table_b', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO table_b VALUES (2, 'two');
+
+CREATE TABLE table_c (pk INTEGER PRIMARY KEY, f2 VARCHAR(128));
+INSERT INTO table_c VALUES (1, 'one');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'table_c', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO table_c VALUES (2, 'two');
+
+CREATE TABLE table_d (pk INTEGER PRIMARY KEY, f2 VARCHAR(128));
+INSERT INTO table_d VALUES (1, 'one');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'table_d', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO table_d VALUES (2, 'two');
+
+CREATE TABLE table_e (pk INTEGER PRIMARY KEY, f2 VARCHAR(128));
+INSERT INTO table_e VALUES (1, 'one');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'table_e', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO table_e VALUES (2, 'two');
+
+CREATE TABLE table_f (pk INTEGER PRIMARY KEY, f2 VARCHAR(10));
+INSERT INTO table_f VALUES (1, 'var0');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'table_f', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO table_f VALUES (2, 'var1');
+
+CREATE TABLE table_g (pk INTEGER PRIMARY KEY, f2 VARCHAR(128));
+INSERT INTO table_g VALUES (1, 'one');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'table_g', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO table_g VALUES (2, 'two');
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection (
+    TEXT COLUMNS (dbo.table_f.f2)
+  )
+  FOR ALL TABLES;
+
+> DROP SOURCE dummy;
+
+> SHOW SUBSOURCES ON mz_source
+mz_source_progress    progress
+table_a               subsource
+table_b               subsource
+table_c               subsource
+table_d               subsource
+table_e               subsource
+table_f               subsource
+table_g               subsource
+
+> SELECT regexp_match(create_sql, 'TEXT COLUMNS = \((.*?)\)')[1] FROM (SHOW CREATE SOURCE mz_source);
+dbo.table_f.f2
+
+> SELECT regexp_match(create_sql, 'TEXT COLUMNS = \((.*?)\)')[1] FROM (SHOW CREATE SOURCE table_f);
+"f2"
+
+#
+# State checking
+#
+> DROP SOURCE table_a
+
+> SELECT * FROM table_b;
+1 one
+2 two
+
+> SHOW SUBSOURCES ON mz_source
+mz_source_progress    progress
+table_b               subsource
+table_c               subsource
+table_d               subsource
+table_e               subsource
+table_f               subsource
+table_g               subsource
+
+! SELECT * FROM table_a;
+contains: unknown catalog item 'table_a'
+
+# Makes progress after dropping subsources
+$ sql-server-execute name=sql-server
+INSERT INTO table_b VALUES (3, 'three');
+
+> SELECT * FROM table_b;
+1 one
+2 two
+3 three
+
+# IF EXISTS works
+> DROP SOURCE IF EXISTS table_a;
+
+# Multiple, repetitive tables work
+> DROP SOURCE table_b, table_c, table_b, table_c, table_b, table_c;
+
+# IF EXISTS works with multiple tables
+> DROP SOURCE IF EXISTS table_c, table_d;
+
+> CREATE MATERIALIZED VIEW mv_e AS SELECT pk + 1 FROM table_e;
+> CREATE MATERIALIZED VIEW mv_f AS SELECT pk + 1 FROM table_f;
+
+# Makes progress after dropping subsources
+$ sql-server-execute name=sql-server
+INSERT INTO table_e VALUES (3, 'three');
+
+> SELECT * FROM mv_e;
+2
+3
+4
+
+> SHOW MATERIALIZED VIEWS
+mv_e quickstart ""
+mv_f quickstart ""
+
+# RESTRICT works
+! DROP SOURCE table_e RESTRICT;
+contains:cannot drop source "table_e": still depended upon by materialized view "mv_e"
+
+# CASCADE works
+> DROP SOURCE table_e CASCADE;
+
+# IF NOT EXISTS + CASCADE works
+> DROP SOURCE IF EXISTS table_e, table_f CASCADE;
+
+# TEXT COLUMNS removed from table_f
+> SELECT regexp_match(create_sql, 'TEXT COLUMNS = \((.*?)\)')[1] FROM (SHOW CREATE SOURCE mz_source);
+<null>
+
+> SHOW SUBSOURCES ON mz_source
+mz_source_progress    progress
+table_g               subsource
+
+> SHOW MATERIALIZED VIEWS
+
+> DROP SOURCE table_g;
+
+> SHOW SUBSOURCES ON mz_source
+mz_source_progress    progress
+
+#
+# Add subsources
+
+> ALTER SOURCE mz_source ADD SUBSOURCE table_g;
+
+! ALTER SOURCE mz_source ADD SUBSOURCE table_g;
+contains:catalog item 'table_g' already exists
+
+> ALTER SOURCE mz_source ADD SUBSOURCE table_a, table_b AS tb;
+
+> SELECT * FROM table_a;
+1 one
+2 two
+
+! ALTER SOURCE mz_source ADD SUBSOURCE table_a;
+contains:catalog item 'table_a' already exists
+
+> SELECT * FROM tb;
+1 one
+2 two
+3 three
+
+!SELECT * FROM table_b;
+contains:unknown catalog item
+
+# We can add tables that didn't exist at the time of publication
+$ sql-server-execute name=sql-server
+CREATE TABLE table_h (pk INTEGER PRIMARY KEY, f2 VARCHAR(128));
+INSERT INTO table_h VALUES (1, 'one');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'table_h', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO table_h VALUES (2, 'two');
+
+> ALTER SOURCE mz_source ADD SUBSOURCE table_h;
+
+> SELECT * FROM table_h;
+1 one
+2 two
+
+> SHOW SUBSOURCES ON mz_source
+mz_source_progress progress
+table_a            subsource
+table_g            subsource
+table_h            subsource
+tb                 subsource
+
+#
+# Complex subsource operations
+
+# If your schema change breaks the subsource, you can fix it.
+$ sql-server-execute name=sql-server
+ALTER TABLE table_a DROP COLUMN f2;
+
+# TODO: This should fail
+> SELECT * FROM table_a;
+1 one
+2 two
+
+#! SELECT * FROM table_a;
+#contains:incompatible schema change
+#
+#> SELECT error ~~ '%incompatible schema change%' FROM mz_internal.mz_source_statuses WHERE name = 'table_a';
+#true
+
+# Subsource errors not propagated to primary source
+> SELECT error IS NULL FROM mz_internal.mz_source_statuses WHERE name = 'mz_source';
+true
+
+> DROP SOURCE table_a;
+> ALTER SOURCE mz_source ADD SUBSOURCE table_a;
+
+> SELECT * FROM table_a;
+1
+2
+
+# If you add columns you can re-ingest them
+$ sql-server-execute name=sql-server
+ALTER TABLE table_a ADD f2 VARCHAR(128);
+INSERT INTO table_a VALUES (3, 'three');
+
+> SELECT * FROM table_a;
+1
+2
+3
+
+> DROP SOURCE table_a;
+> ALTER SOURCE mz_source ADD SUBSOURCE table_a;
+
+> SELECT * FROM table_a;
+1 <null>
+2 <null>
+3 three
+
+# If you add a NOT NULL constraint, you can propagate it.
+$ sql-server-execute name=sql-server
+ALTER TABLE table_a ADD f3 int DEFAULT 1 NOT NULL;
+INSERT INTO table_a VALUES (4, 'four', 4);
+
+> DROP SOURCE table_a;
+> ALTER SOURCE mz_source ADD SUBSOURCE table_a;
+
+> SELECT * FROM table_a;
+1 <null> 1
+2 <null> 1
+3 three 1
+4 four 4
+
+? EXPLAIN OPTIMIZED PLAN AS VERBOSE TEXT FOR SELECT * FROM table_a WHERE f3 IS NULL;
+Explained Query (fast path):
+  Constant <empty>
+
+Target cluster: quickstart
+
+# Can add tables with text columns
+! ALTER SOURCE mz_source ADD SUBSOURCE table_f WITH (TEXT COLUMNS (dbo.table_f.f2, dbo.table_f.f2));
+contains: invalid TEXT COLUMNS option value: unexpected multiple references to dbo.table_f.f2
+
+> ALTER SOURCE mz_source ADD SUBSOURCE table_f WITH (TEXT COLUMNS (dbo.table_f.f2));
+
+> SELECT * FROM table_f
+1 var0
+2 var1
+
+> SELECT regexp_match(create_sql, 'TEXT COLUMNS = \((.*?)\)')[1] FROM (SHOW CREATE SOURCE mz_source);
+dbo.table_f.f2
+
+# Drop a table that's in the publication, which shuffles the tables' output
+# indexes, then add a table to the publication and ensure it can be added.
+$ sql-server-execute name=sql-server
+DROP TABLE table_c, table_d;
+
+CREATE TABLE table_i (pk INTEGER PRIMARY KEY, f2 VARCHAR(128));
+INSERT INTO table_i VALUES (1, 'var0');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'table_i', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO table_i VALUES (2, 'var1');
+
+INSERT INTO table_f VALUES (3, 'var1');
+
+> ALTER SOURCE mz_source ADD SUBSOURCE table_i WITH (TEXT COLUMNS (dbo.table_i.f2));
+
+# TODO: This should work
+# > SELECT regexp_match(create_sql, 'TEXT COLUMNS = \((.*?)\)')[1] FROM (SHOW CREATE SOURCE mz_source);
+# "dbo.table_f.f2, dbo.table_i.f2"
+#
+# > SELECT regexp_match(create_sql, 'TEXT COLUMNS = \((.*?)\)')[1] FROM (SHOW CREATE SOURCE table_i);
+# f2
+
+> SELECT * FROM table_f
+1 var0
+2 var1
+3 var1
+
+> DROP SOURCE table_f, table_i;
+
+> SELECT regexp_match(create_sql, 'TEXT COLUMNS = \((.*?)\)')[1] FROM (SHOW CREATE SOURCE mz_source);
+<null>
+
+! ALTER SOURCE mz_source ADD SUBSOURCE table_e WITH (TEXT COLUMNS (dbo.table_z.a));
+contains:SQL SERVER source validation: TEXT COLUMNS refers to table not currently being added
+
+! ALTER SOURCE mz_source ADD SUBSOURCE table_e WITH (TEXT COLUMNS (dbo.table_f.f2));
+contains:SQL SERVER source validation: TEXT COLUMNS refers to table not currently being added
+detail:the following columns are referenced but not added: dbo.table_f.f2
+
+# Test adding text cols w/o original text columns
+
+> CREATE SOURCE "mz_source_wo_init_text_cols"
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (table_a AS t_a);
+
+> SELECT regexp_match(create_sql, 'TEXT COLUMNS = \((.*?)\)')[1] FROM (SHOW CREATE SOURCE mz_source_wo_init_text_cols);
+<null>
+
+> ALTER SOURCE mz_source_wo_init_text_cols ADD SUBSOURCE table_f AS t_f WITH (TEXT COLUMNS (dbo.table_f.f2));
+
+> SELECT regexp_match(create_sql, 'TEXT COLUMNS = \((.*?)\)')[1] FROM (SHOW CREATE SOURCE mz_source_wo_init_text_cols);
+dbo.table_f.f2
+
+# add a table after having created the source
+$ sql-server-execute name=sql-server
+CREATE TABLE t2 (f1 BIT);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't2', @role_name = 'SA', @supports_net_changes = 0;
+
+! SELECT COUNT(*) > 0 FROM t2;
+contains:unknown catalog item 't2'
diff --git a/test/sql-server-cdc-old-syntax/alter-table-after-source.td b/test/sql-server-cdc-old-syntax/alter-table-after-source.td
new file mode 100644
index 0000000000000..539d6f6541f9c
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/alter-table-after-source.td
@@ -0,0 +1,323 @@
+# 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.
+
+#
+# Test ALTER TABLE -- source will error out for tables which existed when the source was created
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}';
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE;
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE TABLE add_columns (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'add_columns', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO add_columns VALUES (1);
+
+CREATE TABLE remove_column (f1 INTEGER, f2 VARCHAR(2));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'remove_column', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO remove_column VALUES (2, 'ab');
+
+CREATE TABLE alter_column (f1 INTEGER, f2 VARCHAR(2));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_column', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_column VALUES (2, 'ab');
+
+CREATE TABLE alter_drop_nullability (f1 INTEGER NOT NULL);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_drop_nullability', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_drop_nullability VALUES (1);
+
+CREATE TABLE alter_add_nullability (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_add_nullability', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_add_nullability VALUES (1);
+
+CREATE TABLE alter_add_pk (f1 INTEGER NOT NULL);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_add_pk', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_add_pk VALUES (1);
+
+CREATE TABLE alter_drop_unique (f1 INTEGER, CONSTRAINT alter_drop_unique_f1_key UNIQUE (f1));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_drop_unique', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_drop_unique VALUES (1);
+
+CREATE TABLE alter_add_unique (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_add_unique', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_add_unique VALUES (1);
+
+CREATE TABLE alter_extend_column (f1 VARCHAR(2));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_extend_column', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_extend_column VALUES ('ab');
+
+CREATE TABLE alter_decimal (f1 DECIMAL(5,2));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_decimal', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_decimal VALUES (123.45);
+
+CREATE TABLE alter_table_rename (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_table_rename', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_table_rename VALUES (1);
+
+CREATE TABLE alter_table_rename_column (f1 VARCHAR(10), f2 VARCHAR(10));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_table_rename_column', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_table_rename_column (f1, f2) VALUES ('f1_orig','f2_orig');
+
+CREATE TABLE alter_table_change_attnum (f1 VARCHAR(10), f2 VARCHAR(10));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_table_change_attnum', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_table_change_attnum (f1, f2) VALUES ('f1_orig','f2_orig');
+
+CREATE TABLE alter_table_supported (f1 int, f2 int);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alter_table_supported', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alter_table_supported (f1, f2) VALUES (1, 1);
+
+CREATE TABLE truncate_table (f1 int, f2 int);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'truncate_table', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO truncate_table (f1, f2) VALUES (1, 1);
+
+CREATE TABLE drop_table (f1 int, f2 int);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'drop_table', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO drop_table (f1, f2) VALUES (1, 1);
+
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+#
+# Add column
+
+> SELECT * FROM add_columns;
+1
+
+$ sql-server-execute name=sql-server
+ALTER TABLE add_columns ADD f2 varchar(2);
+INSERT INTO add_columns VALUES (2, 'ab');
+
+> SELECT * from add_columns;
+1
+2
+
+
+#
+# Remove column
+
+> SELECT * from remove_column;
+2 ab
+
+$ sql-server-execute name=sql-server
+ALTER TABLE remove_column DROP COLUMN f2;
+
+INSERT INTO remove_column VALUES (3);
+
+# TODO: This should fail
+#> SELECT * from remove_column;
+#2
+#3
+
+
+#
+# Alter column type
+
+> SELECT * from alter_column;
+2 ab
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_column ALTER COLUMN f2 CHAR(2);
+INSERT INTO alter_column VALUES (3, 'cd');
+
+# TODO: This should fail
+> SELECT * from alter_column;
+2 ab
+3 cd
+
+
+#
+# Drop NOT NULL
+
+> SELECT * from alter_drop_nullability
+1
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_drop_nullability ALTER COLUMN f1 INTEGER NULL;
+
+# TODO: This should fail
+# ! SELECT * FROM alter_drop_nullability WHERE f1 IS NOT NULL;
+# contains:altered
+
+# We have guaranteed that this column is not null so the optimizer eagerly
+# returns the empty set.
+> SELECT * FROM alter_drop_nullability WHERE f1 IS NULL;
+
+
+#
+# Add NOT NULL
+
+> SELECT * from alter_add_nullability
+1
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_add_nullability ALTER COLUMN f1 INTEGER NOT NULL;
+INSERT INTO alter_add_nullability VALUES (1);
+
+> SELECT * FROM alter_add_nullability;
+1
+1
+
+
+#
+# Add PK
+
+> SELECT * from alter_add_pk
+1
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_add_pk ADD PRIMARY KEY(f1);
+INSERT INTO alter_add_pk VALUES (2);
+
+> SELECT * FROM alter_add_pk;
+1
+2
+
+
+#
+# Drop unique
+
+> SELECT * from alter_drop_unique
+1
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_drop_unique DROP CONSTRAINT alter_drop_unique_f1_key;
+
+# TODO: This should fail
+# ! SELECT f1 FROM alter_drop_unique;
+# contains:altered
+
+
+#
+# Add unique
+> SELECT * from alter_add_unique
+1
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_add_unique ADD UNIQUE(f1);
+INSERT INTO alter_add_unique VALUES (2);
+
+> SELECT * FROM alter_add_unique;
+1
+2
+
+
+#
+# Extend column
+
+
+> SELECT * from alter_extend_column
+ab
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_extend_column ALTER COLUMN f1 VARCHAR(20);
+
+# TODO: This should fail
+# ! SELECT * FROM alter_extend_column;
+# contains:altered
+
+
+#
+# Alter decimal
+> SELECT * from alter_decimal
+123.45
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_decimal ALTER COLUMN f1 DECIMAL(6,1);
+
+# TODO: This should fail
+# ! SELECT * FROM alter_decimal;
+# contains:altered
+
+
+#
+# Alter table rename
+
+> SELECT * from alter_table_rename;
+1
+
+$ sql-server-execute name=sql-server
+EXEC sp_rename 'alter_table_rename', 'alter_table_renamed';
+
+# TODO: Should fail
+# ! SELECT * FROM alter_table_rename;
+# contains:altered
+
+#
+# Change column attnum
+
+> SELECT * from alter_table_change_attnum;
+f1_orig f2_orig
+
+# Ensure simple name swap doesn't fool schema detection
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_table_change_attnum DROP COLUMN f2;
+ALTER TABLE alter_table_change_attnum ADD f2 VARCHAR(10);
+
+# TODO: Should fail
+# ! SELECT * FROM alter_table_change_attnum;
+# contains:altered
+
+> SELECT * from alter_table_supported;
+1 1
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_table_supported ADD f3 int;
+INSERT INTO alter_table_supported (f1, f2, f3) VALUES (2, 2, 2);
+
+> SELECT * from alter_table_supported;
+1 1
+2 2
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_table_supported DROP COLUMN f3;
+INSERT INTO alter_table_supported (f1, f2) VALUES (3, 3);
+
+> SELECT * from alter_table_supported;
+1 1
+2 2
+3 3
+
+$ sql-server-execute name=sql-server
+ALTER TABLE alter_table_supported DROP COLUMN f2;
+
+# TODO: This should fail
+# ! SELECT * from alter_table_supported;
+# contains:altered
+
+
+#
+# Drop table
+
+> SELECT * from drop_table;
+1 1
+
+$ sql-server-execute name=sql-server
+DROP TABLE drop_table;
+
+# Table is dropped
+# TODO: This should fail
+# ! SELECT * FROM drop_table;
+# regex:(table was dropped|incompatible schema change)
diff --git a/test/sql-server-cdc-old-syntax/alter-table-irrelevant.td b/test/sql-server-cdc-old-syntax/alter-table-irrelevant.td
new file mode 100644
index 0000000000000..abc66b052a4e5
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/alter-table-irrelevant.td
@@ -0,0 +1,108 @@
+# 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.
+
+#
+# Test ALTER TABLE -- tables that are created after the source are irrelevant
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE TABLE base_table (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'base_table', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO base_table VALUES (1);
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT * FROM base_table;
+1
+
+# Create "irrelevant" table after the materialized source took a snapshot of the publication
+$ sql-server-execute name=sql-server
+CREATE TABLE irrelevant_table (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'irrelevant_table', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO irrelevant_table VALUES (1);
+INSERT INTO base_table VALUES (2);
+
+# A new table arriving does not prevent queries on existing views for this materialized source
+> SELECT * FROM base_table;
+1
+2
+
+# Alter the irrelevant table and insert a row to force a second relation message that would be incompatible
+$ sql-server-execute name=sql-server
+ALTER TABLE irrelevant_table ADD f2 varchar(2);
+ALTER TABLE irrelevant_table DROP COLUMN f1;
+INSERT INTO irrelevant_table VALUES ('ab');
+
+# Query still works because the relation was ignored for being irrelevant
+> SELECT * FROM base_table;
+1
+2
+
+# Recreate the source and views to verify the irrelevant_table is part of the publication
+> DROP SOURCE mz_source CASCADE;
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+# Check the first view still works
+> SELECT * FROM base_table;
+1
+2
+
+# Confirm the second table now has a corresponding view and it has the expected data
+> SELECT * FROM irrelevant_table
+<null>
+ab
+
+# Alter the irrelevant_table now that it is relevant
+$ sql-server-execute name=sql-server
+ALTER TABLE irrelevant_table ADD f3 char(2);
+INSERT INTO irrelevant_table VALUES ('bc', 'de');
+
+> SELECT * FROM base_table;
+1
+2
+
+# TODO: Fails with: Decode error: column 'f2' was invalid when getting as type 'varchar' (original text: Row { columns: [Column { name: "__$start_lsn", column_type: BigBinary }, Column { name: "__$seqval", column_type: BigBinary }, Column { name: "__$operation", column_type: Int4 }, Column { name: "__$update_mask", column_type: BigVarBin }, Column { name: "f1", column_type: Int4 }], data: TokenRow { data: [Binary(Some([0, 0, 0, 44, 0, 0, 5, 232, 0, 3])), Binary(Some([0, 0, 0, 44, 0, 0, 5, 232, 0, 2])), I32(Some(2)), Binary(Some([1])), I32(None)] }, result_index: 0 }
+# > SELECT * FROM irrelevant_table
+# <null>
+# ab
+# bc
+
+# Alter in an incompatible way and ensure replication error does not occur
+$ sql-server-execute name=sql-server
+ALTER TABLE irrelevant_table DROP COLUMN f2;
+INSERT INTO irrelevant_table VALUES ('gh');
+
+> SELECT * FROM base_table;
+1
+2
+
+> DROP SOURCE irrelevant_table CASCADE;
diff --git a/test/sql-server-cdc-old-syntax/constraints.td b/test/sql-server-cdc-old-syntax/constraints.td
new file mode 100644
index 0000000000000..1efe29399406a
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/constraints.td
@@ -0,0 +1,117 @@
+# 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.
+
+# IMPORTANT: The Postgres server has a custom pg_hba.conf that only
+# accepts connections from specific users. You will have to update
+# pg_hba.conf if you modify the existing user names or add new ones.
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password};Database=test
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE TABLE pk (f1 INT, f2 INT, f3 INT, PRIMARY KEY (f1, f2));
+INSERT INTO pk VALUES (1,1,null);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'pk', @role_name = 'SA', @supports_net_changes = 0;
+
+CREATE TABLE unique_not_null (f1 INT NOT NULL, f2 INT NOT NULL, f3 INT, UNIQUE (f1, f2));
+INSERT INTO unique_not_null VALUES (1,1,null);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'unique_not_null', @role_name = 'SA', @supports_net_changes = 0;
+
+CREATE TABLE unique_nullable (f1 INT, f2 INT, f3 INT, UNIQUE (f1, f2));
+INSERT INTO unique_nullable VALUES (1,1,null), (2,2,null), (null,null,null);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'unique_nullable', @role_name = 'SA', @supports_net_changes = 0;
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> CREATE DEFAULT INDEX ON pk;
+> CREATE DEFAULT INDEX ON unique_not_null;
+> CREATE DEFAULT INDEX ON unique_nullable;
+
+# TODO: This should be f1,f2 instead of f1,f2,f3
+#> SELECT key FROM (SHOW INDEXES ON pk);
+#{f1,f2}
+
+# TODO: This should be f1,f2 instead of f1,f2,f3
+#> SELECT key FROM (SHOW INDEXES ON unique_not_null);
+#{f1,f2}
+
+> SELECT key FROM (SHOW INDEXES ON unique_nullable);
+{f1,f2,f3}
+
+> SELECT * FROM unique_nullable
+1 1 <null>
+2 2 <null>
+<null> <null> <null>
+
+# Not null constraint optimizes like this:
+? EXPLAIN OPTIMIZED PLAN AS VERBOSE TEXT FOR SELECT * FROM pk WHERE f1 IS NULL OR f2 IS NULL;
+Explained Query (fast path):
+  Constant <empty>
+
+Target cluster: quickstart
+
+? EXPLAIN OPTIMIZED PLAN AS VERBOSE TEXT FOR SELECT * FROM unique_not_null WHERE f1 IS NULL OR f2 IS NULL;
+Explained Query (fast path):
+  Constant <empty>
+
+Target cluster: quickstart
+
+#
+# Unique converted to keys
+? EXPLAIN OPTIMIZED PLAN AS VERBOSE TEXT FOR SELECT DISTINCT f1, f2 FROM pk
+Explained Query (fast path):
+  Project (#0, #1)
+    ReadIndex on=materialize.public.pk pk_primary_idx=[*** full scan ***]
+
+Used Indexes:
+  - materialize.public.pk_primary_idx (*** full scan ***)
+
+Target cluster: quickstart
+
+
+#
+# Unique converted to keys
+? EXPLAIN OPTIMIZED PLAN AS VERBOSE TEXT FOR SELECT DISTINCT f1, f2 FROM unique_not_null
+Explained Query:
+  Distinct project=[#0, #1]
+    Project (#0, #1)
+      ReadIndex on=unique_not_null unique_not_null_primary_idx=[*** full scan ***]
+
+Used Indexes:
+  - materialize.public.unique_not_null_primary_idx (*** full scan ***)
+
+Target cluster: quickstart
+
+
+# Unique NOT converted to keys because values are nullable
+? EXPLAIN OPTIMIZED PLAN AS VERBOSE TEXT FOR SELECT DISTINCT f1, f2 FROM unique_nullable
+Explained Query:
+  Distinct project=[#0, #1]
+    Project (#0, #1)
+      ReadIndex on=unique_nullable unique_nullable_primary_idx=[*** full scan ***]
+
+Used Indexes:
+  - materialize.public.unique_nullable_primary_idx (*** full scan ***)
+
+Target cluster: quickstart
diff --git a/test/sql-server-cdc-old-syntax/correctness-property-2.td b/test/sql-server-cdc-old-syntax/correctness-property-2.td
new file mode 100644
index 0000000000000..32a0b26f637e0
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/correctness-property-2.td
@@ -0,0 +1,60 @@
+# 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.
+
+$ nop
+
+# FIXME(ptravers): https://github.com/MaterializeInc/database-issues/issues/9575
+# $ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+# ALTER SYSTEM SET unsafe_enable_unorchestrated_cluster_replicas = true
+#
+# $ sql-server-connect name=sql-server
+# server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password};Database=test
+#
+# > CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+#
+# > DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+# > CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+#     HOST 'sql-server',
+#     PORT 1433,
+#     DATABASE test,
+#     USER '${arg.default-sql-server-user}',
+#     PASSWORD = SECRET sql_server_pass
+#   );
+#
+# > VALIDATE CONNECTION sql_server_test_connection;
+#
+# $ sql-server-execute name=sql-server
+# CREATE TABLE t1 (pk INT IDENTITY(1,1) PRIMARY KEY, f2 BIGINT);
+# EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+#
+# WITH numbers AS (SELECT 1 AS x UNION ALL SELECT x + 1 FROM numbers WHERE x + 1 <= 10000) INSERT INTO t1 (f2) SELECT x FROM numbers OPTION (MAXRECURSION 0);
+#
+# # Create a cluster with no replicas so that we have time to submit queries at the minimum frontier.
+# > CREATE CLUSTER storage REPLICAS ()
+#
+# > CREATE SOURCE mz_source
+#   FROM SQL SERVER CONNECTION sql_server_test_connection
+#   FOR ALL TABLES
+#   WITH (RETAIN HISTORY = FOR '365000 days');
+#
+# # Grab a cursor at timestamp 0
+# > BEGIN
+#
+# > DECLARE c CURSOR FOR SELECT 1, COUNT(*) FROM t1 AS OF 0
+#
+# # Start ingestion by adding a replica to the cluster. We must do this from a
+# # different connection to not disturbe the transaction we're in.
+# $ postgres-execute connection=postgres://materialize:materialize@${testdrive.materialize-sql-addr}
+# CREATE CLUSTER REPLICA storage.r1 SIZE = 'scale=1,workers=1';
+#
+# # Verify that at timestamp 0 there is only one record whose value is the final value
+# > FETCH 1 c WITH (timeout = '1d');
+# 1 10000
+#
+# > COMMIT
diff --git a/test/sql-server-cdc-old-syntax/create-views-from-source-in-schema.td b/test/sql-server-cdc-old-syntax/create-views-from-source-in-schema.td
new file mode 100644
index 0000000000000..342f38ed99168
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/create-views-from-source-in-schema.td
@@ -0,0 +1,66 @@
+# 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.
+
+#
+# Test that CREATE SOURCE can be used to direct the tables to another schema
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t1 VALUES (1);
+
+CREATE TABLE t2 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't2', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t2 VALUES (2);
+
+> DROP SCHEMA IF EXISTS schema1
+
+> CREATE SCHEMA schema1
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1 AS schema1.t1, t2 AS schema1.t2);
+
+# Wait for snapshot to be complete
+> SELECT COUNT(*) > 0 FROM schema1.t1;
+true
+
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+INSERT INTO t2 SELECT * FROM t2;
+
+> SELECT * FROM schema1.t1;
+1
+1
+
+> SELECT * FROM schema1.t2;
+2
+2
+
+> DROP SCHEMA schema1 CASCADE;
diff --git a/test/sql-server-cdc-old-syntax/data-utf8.td b/test/sql-server-cdc-old-syntax/data-utf8.td
new file mode 100644
index 0000000000000..323925832c267
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/data-utf8.td
@@ -0,0 +1,58 @@
+# 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.
+
+#
+# Test UTF-8 data both pre- and -post snapshot
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 NVARCHAR(128));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1 VALUES (N'това е текст'), (N'това ''е'' "текст"');
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+# Insert the same data post-snapshot
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+
+> SELECT pg_typeof(f1) FROM t1 LIMIT 1;
+"text"
+
+> SELECT * FROM t1;
+"това е текст"
+"това \'е\' \"текст\""
+"това е текст"
+"това \'е\' \"текст\""
diff --git a/test/sql-server-cdc-old-syntax/dropped-slot-errors.td b/test/sql-server-cdc-old-syntax/dropped-slot-errors.td
new file mode 100644
index 0000000000000..61d5c5776d90f
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/dropped-slot-errors.td
@@ -0,0 +1,57 @@
+# 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.
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET unsafe_enable_unorchestrated_cluster_replicas = true
+
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (pk INT IDENTITY(1,1) PRIMARY KEY, f2 varchar(128));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+> CREATE CLUSTER storage REPLICAS (r1 (SIZE 'scale=1,workers=1'))
+
+> CREATE SOURCE mz_source
+  IN CLUSTER storage
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES
+
+# Wait for the initial snapshot to be ingested
+> SELECT * FROM t1
+
+# Stop ingestion by dropping the replica
+> DROP CLUSTER REPLICA storage.r1;
+
+# Now drop the replication slot
+$ sql-server-execute name=sql-server
+EXEC sys.sp_cdc_disable_db;
+
+# Resume the ingestion by adding a replica to the cluster.
+> CREATE CLUSTER REPLICA storage.r1 SIZE 'scale=1,workers=1';
+
+> SELECT error ~~ 'sql-server: into results: Token error: ''Invalid object name ''cdc.lsn_time_mapping''.'' on server % executing  on line 1 (code: %, state: %, class: %)' FROM mz_internal.mz_source_statuses WHERE name = 't1';
+true
diff --git a/test/sql-server-cdc-old-syntax/gh-10981.td b/test/sql-server-cdc-old-syntax/gh-10981.td
new file mode 100644
index 0000000000000..367a962a37c0c
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/gh-10981.td
@@ -0,0 +1,41 @@
+# 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.
+
+#
+# Test for issue database-issues#3215
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password};Database=test
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (id int, name varchar(128), id2 int);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t1 VALUES (1, NULL, 1);
+
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1);
+
+> SELECT * FROM t1;
+1 <null> 1
diff --git a/test/sql-server-cdc-old-syntax/max-slot-wal-keep-size.td b/test/sql-server-cdc-old-syntax/max-slot-wal-keep-size.td
new file mode 100644
index 0000000000000..86dc1e5574479
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/max-slot-wal-keep-size.td
@@ -0,0 +1,57 @@
+# 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.
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password};Database=test
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET unsafe_enable_unorchestrated_cluster_replicas = true
+
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (pk INT IDENTITY(1,1) PRIMARY KEY, f2 varchar(2048));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+> CREATE CLUSTER storage REPLICAS (r1 (SIZE 'scale=1,workers=1'))
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+# Wait for the initial snapshot to be ingested
+> SELECT * FROM t1
+
+# Stop ingestion by dropping the replica
+> DROP CLUSTER REPLICA storage.r1;
+
+# Now generate more than 10MB of WAL data so that the slot gets invalidated
+$set count=200000
+$ sql-server-execute name=sql-server
+WITH numbers AS (SELECT 1 AS x UNION ALL SELECT x + 1 FROM numbers WHERE x + 1 <= ${count}) INSERT INTO t1 (f2) SELECT REPLICATE('0123456789abcdef', 64) FROM numbers OPTION (MAXRECURSION 0);
+UPDATE t1 SET f2 = CONCAT(f2, pk);
+UPDATE t1 SET f2 = CONCAT(f2, pk);
+
+# Resume the ingestion by adding a replica to the cluster.
+> CREATE CLUSTER REPLICA storage.r1 SIZE = 'scale=1,workers=1';
+
+$ set-sql-timeout duration=60s
+> SELECT count(*) FROM t1;
+200000
diff --git a/test/sql-server-cdc-old-syntax/mzcompose b/test/sql-server-cdc-old-syntax/mzcompose
new file mode 100755
index 0000000000000..1f866645dabc8
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/mzcompose
@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+
+# 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.
+#
+# mzcompose — runs Docker Compose with Materialize customizations.
+
+exec "$(dirname "$0")"/../../bin/pyactivate -m materialize.cli.mzcompose "$@"
diff --git a/test/sql-server-cdc-old-syntax/mzcompose.py b/test/sql-server-cdc-old-syntax/mzcompose.py
new file mode 100644
index 0000000000000..478c3ab2f148f
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/mzcompose.py
@@ -0,0 +1,84 @@
+# 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.
+
+"""
+Native SQL Server source tests, functional.
+"""
+
+import glob
+import pathlib
+import random
+
+from materialize import MZ_ROOT, buildkite
+from materialize.mzcompose.composition import (
+    Composition,
+    WorkflowArgumentParser,
+)
+from materialize.mzcompose.services.materialized import Materialized
+from materialize.mzcompose.services.mz import Mz
+from materialize.mzcompose.services.sql_server import SqlServer
+from materialize.mzcompose.services.testdrive import Testdrive
+
+SERVICES = [
+    Mz(app_password=""),
+    Materialized(
+        additional_system_parameter_defaults={
+            "log_filter": "mz_storage::source::sql_server=trace,mz_storage::source::sql_server::replication=trace,mz_sql_server_util=debug,info",
+            "max_credit_consumption_rate": "2000",
+        },
+    ),
+    Testdrive(),
+    SqlServer(),
+]
+
+
+def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
+    parser.add_argument(
+        "filter",
+        nargs="*",
+        default=["*.td"],
+        help="limit to only the files matching filter",
+    )
+    args = parser.parse_args()
+
+    matching_files: list[str] = []
+    for filter in args.filter:
+        matching_files.extend(
+            glob.glob(
+                filter,
+                root_dir=MZ_ROOT / "test" / "sql-server-cdc-old-syntax",
+            )
+        )
+    matching_files = sorted(matching_files)
+    sharded_files: list[str] = buildkite.shard_list(
+        sorted(matching_files), lambda file: file
+    )
+    print(f"Filter: {args.filter} Files: {sharded_files}")
+
+    # Start with a fresh state
+    c.kill("sql-server")
+    c.rm("sql-server")
+    c.kill("materialized")
+    c.rm("materialized")
+
+    c.up("materialized", "sql-server")
+    seed = random.getrandbits(16)
+
+    def run(file: pathlib.Path | str) -> None:
+        c.run_testdrive_files(
+            "--max-errors=1",
+            f"--seed={seed}",
+            f"--var=default-replica-size=scale={Materialized.Size.DEFAULT_SIZE},workers={Materialized.Size.DEFAULT_SIZE}",
+            f"--var=default-sql-server-user={SqlServer.DEFAULT_USER}",
+            f"--var=default-sql-server-password={SqlServer.DEFAULT_SA_PASSWORD}",
+            "setup/setup.td",
+            str(file),
+        )
+
+    c.test_parts(sharded_files, run)
diff --git a/test/sql-server-cdc-old-syntax/privileges.td b/test/sql-server-cdc-old-syntax/privileges.td
new file mode 100644
index 0000000000000..5a6a4796b7c22
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/privileges.td
@@ -0,0 +1,82 @@
+# 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.
+
+# Setup SQL Server state.
+#
+# Create a table that has CDC enabled.
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+CREATE TABLE t1_privileges (c1 int, c2 int);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1_privileges', @role_name = 'SA', @supports_net_changes = 0;
+
+CREATE LOGIN authorization_user_column_perms WITH PASSWORD = '${arg.default-sql-server-password}';
+CREATE USER authorization_user_column_perms FOR LOGIN authorization_user_column_perms;
+ALTER ROLE db_datareader ADD MEMBER authorization_user_column_perms;
+
+DENY SELECT ON OBJECT::dbo.t1_privileges(c2) TO authorization_user_column_perms;
+
+CREATE LOGIN authorization_user_column_perms_capture_instance WITH PASSWORD = '${arg.default-sql-server-password}';
+CREATE USER authorization_user_column_perms_capture_instance FOR LOGIN authorization_user_column_perms_capture_instance;
+ALTER ROLE db_datareader ADD MEMBER authorization_user_column_perms_capture_instance;
+
+DENY SELECT ON OBJECT::cdc.dbo_t1_privileges_ct(__$start_lsn) TO authorization_user_column_perms_capture_instance;
+
+CREATE LOGIN authorization_user_table_perms WITH PASSWORD = '${arg.default-sql-server-password}';
+CREATE USER authorization_user_table_perms FOR LOGIN authorization_user_table_perms;
+ALTER ROLE db_datareader ADD MEMBER authorization_user_table_perms;
+
+DENY SELECT ON OBJECT::dbo.t1_privileges TO authorization_user_table_perms;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> CREATE CONNECTION sql_server_privileges_connection_table TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER 'authorization_user_table_perms',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+! CREATE SOURCE t1_privileges_sql_server_failure
+  FROM SQL SERVER CONNECTION sql_server_privileges_connection_table
+  FOR TABLES (dbo.t1_privileges);
+contains:reference to dbo.t1_privileges not found in source
+
+> CREATE CONNECTION sql_server_privileges_connection_column TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER 'authorization_user_column_perms',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+# Column exclusion has no effect on rejection for column level permissions.
+! CREATE SOURCE t1_privileges_sql_server_failure
+  FROM SQL SERVER CONNECTION sql_server_privileges_connection_column (
+    EXCLUDE COLUMNS (dbo.t1_privileges.c2)
+  )
+  FOR TABLES (dbo.t1_privileges);
+contains:insufficient permissions for tables [dbo.t1_privileges] or capture instances []
+
+> CREATE CONNECTION sql_server_privileges_connection_capture_instance TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER 'authorization_user_column_perms_capture_instance',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+! CREATE SOURCE t1_privileges_sql_server_failure
+  FROM SQL SERVER CONNECTION sql_server_privileges_connection_capture_instance
+  FOR TABLES (dbo.t1_privileges);
+contains:insufficient permissions for tables [] or capture instances [dbo_t1_privileges]
diff --git a/test/sql-server-cdc-old-syntax/replica-identity.td b/test/sql-server-cdc-old-syntax/replica-identity.td
new file mode 100644
index 0000000000000..0667358c44a07
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/replica-identity.td
@@ -0,0 +1,50 @@
+# 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.
+
+# IMPORTANT: The Postgres server has a custom pg_hba.conf that only
+# accepts connections from specific users. You will have to update
+# pg_hba.conf if you modify the existing user names or add new ones.
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE TABLE tbl_a (pk INTEGER PRIMARY KEY);
+INSERT INTO tbl_a VALUES (1);
+
+! CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES(tbl_a);
+contains:reference to tbl_a not found in source
+
+$ sql-server-execute name=sql-server
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'tbl_a', @role_name = 'SA', @supports_net_changes = 0;
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES(tbl_a);
+
+> SELECT * FROM tbl_a;
+1
diff --git a/test/sql-server-cdc-old-syntax/setup/setup.td b/test/sql-server-cdc-old-syntax/setup/setup.td
new file mode 100644
index 0000000000000..41c27f7a3bda5
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/setup/setup.td
@@ -0,0 +1,58 @@
+# 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.
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server split-lines=false
+IF EXISTS (SELECT name FROM sys.databases WHERE name = N'test')
+BEGIN
+    ALTER DATABASE test SET SINGLE_USER WITH ROLLBACK IMMEDIATE;
+    DROP DATABASE test;
+END;
+
+$ sql-server-execute name=sql-server split-lines=false
+USE msdb;
+IF EXISTS (
+    SELECT 1
+    FROM msdb.dbo.sysjobs
+    WHERE name = N'DummyTicker'
+)
+BEGIN
+    EXEC sp_delete_job @job_name = N'DummyTicker';
+END;
+
+$ sql-server-execute name=sql-server
+CREATE DATABASE test;
+USE test;
+
+EXEC sys.sp_cdc_enable_db;
+ALTER DATABASE test SET ALLOW_SNAPSHOT_ISOLATION ON;
+
+CREATE TABLE dummy (t datetime);
+INSERT INTO dummy VALUES (CURRENT_TIMESTAMP);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'dummy', @role_name = 'SA', @supports_net_changes = 0;
+
+$ sql-server-execute name=sql-server split-lines=false
+CREATE OR ALTER PROCEDURE dbo.DummyTicker AS
+BEGIN
+    SET NOCOUNT ON;
+    WHILE 1 = 1
+    BEGIN
+        UPDATE dbo.dummy SET t = CURRENT_TIMESTAMP;
+        WAITFOR DELAY '00:00:01';
+    END
+END;
+
+$ sql-server-execute name=sql-server
+USE msdb;
+EXEC sp_add_job @job_name = N'DummyTicker', @enabled = 1;
+EXEC sp_add_jobstep @job_name = N'DummyTicker', @step_name = N'run', @subsystem = N'TSQL', @database_name = N'test', @command = N'EXEC dbo.DummyTicker;';
+EXEC sp_add_jobserver @job_name = N'DummyTicker';
+EXEC msdb.dbo.sp_start_job N'DummyTicker';
diff --git a/test/sql-server-cdc-old-syntax/statistics.td b/test/sql-server-cdc-old-syntax/statistics.td
new file mode 100644
index 0000000000000..febc9577342ea
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/statistics.td
@@ -0,0 +1,157 @@
+# 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.
+
+#
+# Test progress statistics
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password};Datbase=test
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET storage_statistics_collection_interval = 1000
+ALTER SYSTEM SET storage_statistics_interval = 2000
+ALTER SYSTEM SET max_credit_consumption_rate = 2000
+
+$ sql-server-execute name=sql-server
+USE test;
+CREATE TABLE t1 (f1 VARCHAR(128));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t1 VALUES ('one');
+INSERT INTO t1 VALUES ('two');
+
+> CREATE CLUSTER stats_cluster SIZE 'scale=1,workers=1'
+
+> CREATE SOURCE mz_source
+  IN CLUSTER stats_cluster
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES ("t1")
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+> SELECT
+    s.name,
+    SUM(u.offset_committed) > 0,
+    SUM(u.offset_known) >= SUM(u.offset_committed),
+    SUM(u.snapshot_records_known),
+    SUM(u.snapshot_records_staged)
+  FROM mz_sources s
+  JOIN mz_internal.mz_source_statistics_raw u ON s.id = u.id
+  WHERE s.name IN ('mz_source')
+  GROUP BY s.name
+  ORDER BY s.name
+mz_source true true 2 2
+
+$ set-from-sql var=previous-offset-committed
+SELECT
+    (SUM(u.offset_committed))::text
+  FROM mz_sources s
+  JOIN mz_internal.mz_source_statistics_raw u ON s.id = u.id
+  WHERE s.name IN ('mz_source')
+
+$ sql-server-execute name=sql-server
+INSERT INTO t1 VALUES ('three');
+
+> SELECT
+    s.name,
+    SUM(u.offset_committed) > ${previous-offset-committed},
+    SUM(u.offset_known) >= SUM(u.offset_committed),
+    SUM(u.snapshot_records_known),
+    SUM(u.snapshot_records_staged)
+  FROM mz_sources s
+  JOIN mz_internal.mz_source_statistics_raw u ON s.id = u.id
+  WHERE s.name IN ('mz_source')
+  GROUP BY s.name
+  ORDER BY s.name
+mz_source true true 2 2
+
+$ set-from-sql var=pre-restart-offset-committed
+SELECT
+    (SUM(u.offset_committed))::text
+  FROM mz_sources s
+  JOIN mz_internal.mz_source_statistics_raw u ON s.id = u.id
+  WHERE s.name IN ('mz_source')
+
+> ALTER CLUSTER stats_cluster SET (REPLICATION FACTOR 0)
+
+$ sql-server-execute name=sql-server
+INSERT INTO t1 VALUES ('four');
+
+> ALTER CLUSTER stats_cluster SET (REPLICATION FACTOR 1)
+
+# Ensure the snapshot stats stay there, and don't change.
+> SELECT
+    s.name,
+    SUM(u.offset_committed) > ${pre-restart-offset-committed},
+    SUM(u.offset_known) >= SUM(u.offset_committed),
+    SUM(u.snapshot_records_known),
+    SUM(u.snapshot_records_staged)
+  FROM mz_sources s
+  JOIN mz_internal.mz_source_statistics_raw u ON s.id = u.id
+  WHERE s.name IN ('mz_source')
+  GROUP BY s.name
+  ORDER BY s.name
+mz_source true true 2 2
+
+$ sql-server-execute name=sql-server
+CREATE TABLE alt (f1 VARCHAR(128));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'alt', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO alt VALUES ('one');
+
+> ALTER SOURCE mz_source ADD SUBSOURCE alt;
+
+> SELECT COUNT(*) > 0 FROM alt;
+true
+
+# FIXME(ptravers): https://github.com/MaterializeInc/database-issues/issues/9583
+# Ensure snapshot stats are overridden when we add a new table
+# > SELECT
+#     s.name,
+#     SUM(u.snapshot_records_known),
+#     SUM(u.snapshot_records_staged)
+#   FROM mz_sources s
+#   JOIN mz_internal.mz_source_statistics_raw u ON s.id = u.id
+#   WHERE s.name IN ('mz_source')
+#   GROUP BY s.name
+#   ORDER BY s.name
+# mz_source 1 1
+
+# Ensure subsource stats show up, and then are removed when we drop subsources.
+> SELECT
+    s.name,
+    SUM(u.updates_committed) > 0
+  FROM mz_sources s
+  JOIN mz_internal.mz_source_statistics_raw u ON s.id = u.id
+  WHERE s.name IN ('alt')
+  GROUP BY s.name
+  ORDER BY s.name
+alt true
+
+> DROP SOURCE alt;
+
+> SELECT
+    count(*)
+  FROM mz_sources s
+  JOIN mz_internal.mz_source_statistics_raw u ON s.id = u.id
+  WHERE s.name IN ('alt')
+0
diff --git a/test/sql-server-cdc-old-syntax/subsource-names.td b/test/sql-server-cdc-old-syntax/subsource-names.td
new file mode 100644
index 0000000000000..c4ac7d710be66
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/subsource-names.td
@@ -0,0 +1,91 @@
+# 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.
+
+#
+# Test that subsources are created in the source schema if provided
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE SCHEMA subsource_test;
+CREATE TABLE subsource_test.t1 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'subsource_test', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO subsource_test.t1 VALUES (1);
+
+CREATE TABLE subsource_test.t2 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'subsource_test', @source_name = 't2', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO subsource_test.t2 VALUES (5);
+
+> CREATE SCHEMA a;
+> CREATE SOURCE a.mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1);
+
+> SELECT * FROM a.t1;
+1
+
+> CREATE SCHEMA another;
+> CREATE SOURCE another.mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1, t2);
+
+> SELECT * FROM another.t1;
+1
+
+> SELECT * FROM another.t2;
+5
+
+> CREATE SCHEMA foo;
+> CREATE SCHEMA bar;
+> CREATE SOURCE foo.mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1 as table1, t2 as bar.table2);
+
+! SELECT * FROM table1;
+contains: unknown
+
+# table1 gets created in source schema foo because it doesn't have any prefix
+> SELECT * FROM foo.table1;
+1
+
+! SELECT * FROM foo.table2;
+contains: unknown
+
+# table2 gets created in mentioned bar because it does have a prefix
+> SELECT * FROM bar.table2;
+5
+
+> CREATE SCHEMA baz;
+> CREATE SOURCE baz.mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR SCHEMAS (subsource_test);
+
+> SELECT * FROM baz.t1;
+1
+
+> SELECT * FROM baz.t2;
+5
diff --git a/test/sql-server-cdc-old-syntax/subsource-resolution-empty.td b/test/sql-server-cdc-old-syntax/subsource-resolution-empty.td
new file mode 100644
index 0000000000000..a9230df920f23
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/subsource-resolution-empty.td
@@ -0,0 +1,68 @@
+# 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.
+
+#
+# Tests that empty publication and missing for tables clauses errors correctly
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server split-lines=false
+IF EXISTS (SELECT name FROM sys.databases WHERE name = N'test')
+BEGIN
+    ALTER DATABASE test SET SINGLE_USER WITH ROLLBACK IMMEDIATE;
+    DROP DATABASE test;
+END;
+CREATE DATABASE test;
+USE test;
+EXEC sys.sp_cdc_enable_db;
+ALTER DATABASE test SET ALLOW_SNAPSHOT_ISOLATION ON;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+USE test;
+CREATE TABLE t (f1 int);
+
+! CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+contains:SQL Server source must ingest at least one table, but FOR ALL TABLES matched none
+
+! CREATE SOURCE mz_source_empty
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1);
+contains:reference to t1 not found in source
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET enable_create_table_from_source = false
+
+! CREATE SOURCE "mz_source_empty"
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+contains:missing TABLES specification
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET enable_create_table_from_source = true
+
+! CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR SCHEMAS (dne);
+contains:no tables found in referenced schemas: "dne"
diff --git a/test/sql-server-cdc-old-syntax/support-multiple-materializations.td b/test/sql-server-cdc-old-syntax/support-multiple-materializations.td
new file mode 100644
index 0000000000000..077297affa28a
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/support-multiple-materializations.td
@@ -0,0 +1,77 @@
+# 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.
+
+#
+# Test that any replication slot can only be materialized once
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (id INT IDENTITY(1,1) PRIMARY KEY, f1 BIT);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+CREATE TABLE t2 (id INT IDENTITY(1,1) PRIMARY KEY, t1_id INT REFERENCES t1(id), name VARCHAR(256));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't2', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1(f1) VALUES ('true'),('false');
+
+INSERT INTO t2(t1_id, name) VALUES (1, 'example');
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> CREATE MATERIALIZED VIEW t1_mat AS
+  SELECT * FROM t1
+
+> SELECT id, f1 FROM t1_mat;
+1 true
+2 false
+
+> CREATE MATERIALIZED VIEW t1_mat_dupe AS
+  SELECT * FROM t1
+
+> DROP MATERIALIZED VIEW t1_mat;
+
+> DROP SOURCE mz_source CASCADE;
+
+# verify that dropping things allows recreation
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> CREATE MATERIALIZED VIEW joiner AS
+  SELECT t2.id, t1.f1, t2.name
+  FROM t1
+  JOIN t2
+  ON   t1.id = t2.t1_id
+
+> SELECT * FROM joiner;
+1 true example
diff --git a/test/sql-server-cdc-old-syntax/transactions-multi-conn.td b/test/sql-server-cdc-old-syntax/transactions-multi-conn.td
new file mode 100644
index 0000000000000..1f82060787361
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/transactions-multi-conn.td
@@ -0,0 +1,228 @@
+# 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.
+
+#
+# Test that transactions work properly
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+
+$ sql-server-execute name=sql-server
+
+$ sql-server-connect name=sql-server-1
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+$ sql-server-connect name=sql-server-2
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+$ sql-server-connect name=sql-server-3
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server-1
+USE test;
+CREATE TABLE t1 (a INT);
+CREATE TABLE t2 (a INT);
+CREATE TABLE t3 (a INT);
+
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't2', @role_name = 'SA', @supports_net_changes = 0;
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't3', @role_name = 'SA', @supports_net_changes = 0;
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT count(*) FROM t1;
+0
+
+> SELECT count(*) FROM t2;
+0
+
+> SELECT count(*) FROM t3;
+0
+
+$ sql-server-execute name=sql-server-1
+USE test;
+BEGIN TRANSACTION;
+INSERT INTO t1 VALUES (1000);
+INSERT INTO t2 VALUES (1000);
+INSERT INTO t3 VALUES (1000);
+
+$ sql-server-execute name=sql-server-2
+USE test;
+BEGIN TRANSACTION;
+INSERT INTO t1 VALUES (2000);
+INSERT INTO t2 VALUES (2000);
+INSERT INTO t3 VALUES (2000);
+
+$ sql-server-execute name=sql-server-3
+USE test;
+BEGIN TRANSACTION;
+INSERT INTO t1 VALUES (3000);
+INSERT INTO t2 VALUES (3000);
+INSERT INTO t3 VALUES (3000);
+
+$ sql-server-execute name=sql-server-1
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+$ sql-server-execute name=sql-server-3
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+> SELECT * FROM t1;
+1000
+3000
+
+> SELECT * FROM t2;
+1000
+3000
+
+> SELECT * FROM t3;
+1000
+3000
+
+$ sql-server-execute name=sql-server-2
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+> SELECT * FROM t1;
+1000
+2000
+3000
+
+# delete and insert statements cannot be done in multiple transactions on the same table even with fine-grained where condition
+
+$ sql-server-execute name=sql-server-1
+INSERT INTO t1 VALUES (1001)
+INSERT INTO t2 VALUES (1001)
+INSERT INTO t3 VALUES (1001)
+
+$ sql-server-execute name=sql-server-2
+INSERT INTO t1 VALUES (2001)
+INSERT INTO t2 VALUES (2001)
+INSERT INTO t3 VALUES (2001)
+
+$ sql-server-execute name=sql-server-3
+INSERT INTO t1 VALUES (3001)
+INSERT INTO t2 VALUES (3001)
+INSERT INTO t3 VALUES (3001)
+
+$ sql-server-execute name=sql-server-1
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+$ sql-server-execute name=sql-server-3
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+> SELECT * FROM t1;
+1000
+1001
+2000
+3000
+3001
+
+> SELECT * FROM t2;
+1000
+1001
+2000
+3000
+3001
+
+> SELECT * FROM t3;
+1000
+1001
+2000
+3000
+3001
+
+$ sql-server-execute name=sql-server-2
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+INSERT INTO t1 VALUES (2002);
+
+$ sql-server-execute name=sql-server-1
+INSERT INTO t1 VALUES (1002);
+DELETE FROM t2 WHERE a = 2000;
+
+$ sql-server-execute name=sql-server-2
+DELETE FROM t3 WHERE a = 2000;
+
+$ sql-server-execute name=sql-server-1
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+> SELECT * FROM t1;
+1000
+1001
+1002
+2000
+2001
+3000
+3001
+
+> SELECT * FROM t2;
+1000
+1001
+2001
+3000
+3001
+
+> SELECT * FROM t3;
+1000
+1001
+2000
+2001
+3000
+3001
+
+$ sql-server-execute name=sql-server-2
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+> SELECT * FROM t1;
+1000
+1001
+1002
+2000
+2001
+2002
+3000
+3001
+
+> SELECT * FROM t2;
+1000
+1001
+2001
+3000
+3001
+
+> SELECT * FROM t3;
+1000
+1001
+2001
+3000
+3001
diff --git a/test/sql-server-cdc-old-syntax/transactions.td b/test/sql-server-cdc-old-syntax/transactions.td
new file mode 100644
index 0000000000000..2ea09525079e0
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/transactions.td
@@ -0,0 +1,122 @@
+# 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.
+
+#
+# Test that transactions work properly
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password};Database=test
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (a INT);
+INSERT INTO t1 VALUES (3);
+INSERT INTO t1 VALUES (4);
+
+CREATE TABLE t2 (a INT);
+
+CREATE TABLE t3 (a INT);
+
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't2', @role_name = 'SA', @supports_net_changes = 0;
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't3', @role_name = 'SA', @supports_net_changes = 0;
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT * FROM t1;
+3
+4
+
+$ sql-server-execute name=sql-server
+BEGIN TRANSACTION;
+DELETE FROM t1;
+INSERT INTO t1 VALUES (5);
+
+> SELECT * FROM t1;
+3
+4
+
+$ sql-server-execute name=sql-server
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+> SELECT * FROM t1;
+5
+
+$ sql-server-execute name=sql-server
+INSERT INTO t1 VALUES (6);
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+INSERT INTO t2 VALUES (20);
+INSERT INTO t3 VALUES (300);
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+DELETE FROM t1;
+ROLLBACK TRANSACTION;
+BEGIN TRANSACTION;
+
+UPDATE t1 SET a = a + 8;
+DELETE FROM t3 WHERE a = 300;
+INSERT INTO t2 SELECT * FROM t1;
+UPDATE t1 SET a = 100 WHERE a = 14;
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+UPDATE t2 SET a = a + 10;
+INSERT INTO t3 VALUES (500), (600);
+INSERT INTO t2 VALUES (44);
+DELETE FROM t3 WHERE a = 500;
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+INSERT INTO t2 VALUES (99);
+
+> SELECT * FROM t1;
+13
+100
+
+> SELECT * FROM t2;
+30
+23
+24
+44
+
+> SELECT * FROM t3;
+600
+
+$ sql-server-execute name=sql-server
+COMMIT TRANSACTION;
+BEGIN TRANSACTION;
+
+> SELECT * FROM t2;
+30
+23
+24
+44
+99
diff --git a/test/sql-server-cdc-old-syntax/two-destination-schemas.td b/test/sql-server-cdc-old-syntax/two-destination-schemas.td
new file mode 100644
index 0000000000000..36ff898d08008
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/two-destination-schemas.td
@@ -0,0 +1,80 @@
+# 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.
+
+#
+# Test that identically-named tables in two destination schemas can be
+# successfully disambiguated and replicated
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t1 VALUES (1);
+
+CREATE TABLE t2 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't2', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t2 VALUES (2);
+
+
+> DROP SCHEMA IF EXISTS schema1 CASCADE;
+> DROP SCHEMA IF EXISTS schema2 CASCADE;
+
+> CREATE SCHEMA schema1;
+> CREATE SCHEMA schema2;
+
+> DROP SOURCE IF EXISTS mz_source CASCADE;
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1 AS schema1.t1, t2 AS schema2.t1);
+
+> SELECT * FROM schema1.t1;
+1
+
+> SELECT * FROM schema2.t1;
+2
+
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+INSERT INTO t2 SELECT * FROM t2;
+
+> SELECT * FROM schema1.t1;
+1
+1
+
+> SELECT * FROM schema2.t1;
+2
+2
+
+$ sql-server-execute name=sql-server
+DROP TABLE t1;
+DROP TABLE t2;
+
+> DROP SOURCE mz_source CASCADE;
+> DROP SCHEMA schema1 CASCADE;
+> DROP SCHEMA schema2 CASCADE;
diff --git a/test/sql-server-cdc-old-syntax/two-publications-one-table.td b/test/sql-server-cdc-old-syntax/two-publications-one-table.td
new file mode 100644
index 0000000000000..2ca9f29e3585d
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/two-publications-one-table.td
@@ -0,0 +1,67 @@
+# 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.
+
+#
+# Test that two publications can replicate the same table
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t1 VALUES (1);
+
+> DROP SCHEMA IF EXISTS schema1
+
+> CREATE SCHEMA schema1
+
+> CREATE SOURCE mz_source1
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1 AS t1_1);
+
+> CREATE SOURCE mz_source2
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1 AS t1_2);
+
+> SELECT * FROM t1_1;
+1
+
+> SELECT * FROM t1_2;
+1
+
+$ sql-server-execute name=sql-server
+INSERT INTO t1 VALUES (2);
+
+> SELECT * FROM t1_1;
+1
+2
+
+> SELECT * FROM t1_2;
+1
+2
+
+> DROP SCHEMA schema1 CASCADE;
diff --git a/test/sql-server-cdc-old-syntax/two-source-schemas.td b/test/sql-server-cdc-old-syntax/two-source-schemas.td
new file mode 100644
index 0000000000000..6bc8ef6cd52a9
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/two-source-schemas.td
@@ -0,0 +1,83 @@
+# 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.
+
+#
+# Test that identically-named tables in two source schemas can be
+# successfully disambiguated and replicated
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+DROP SCHEMA IF EXISTS schema1;
+CREATE SCHEMA schema1;
+
+CREATE TABLE schema1.t1 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'schema1', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO schema1.t1 VALUES (1);
+
+DROP SCHEMA IF EXISTS schema2;
+CREATE SCHEMA schema2;
+
+CREATE TABLE schema2.t1 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'schema2', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO schema2.t1 VALUES (2);
+
+$ sql-server-execute name=sql-server
+INSERT INTO schema1.t1 SELECT * FROM schema1.t1;
+INSERT INTO schema2.t1 SELECT * FROM schema2.t1;
+
+! CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+contains:multiple subsources would be named t1
+
+! CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1);
+contains:reference t1 is ambiguous, consider specifying an additional layer of qualification
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (schema1.t1);
+
+> SELECT * FROM t1;
+1
+1
+
+> DROP SOURCE mz_source CASCADE;
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (schema1.t1 AS t1_1, schema2.t1 AS t1_2);
+
+> SELECT * FROM t1_1;
+1
+1
+
+> SELECT * FROM t1_2;
+2
+2
diff --git a/test/sql-server-cdc-old-syntax/two-sources-one-publication.td b/test/sql-server-cdc-old-syntax/two-sources-one-publication.td
new file mode 100644
index 0000000000000..b8a3bbddd6ec2
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/two-sources-one-publication.td
@@ -0,0 +1,92 @@
+# 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.
+
+#
+# Test that two sources reading from one publication is OK
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t1 VALUES (1);
+
+CREATE TABLE t2 (f1 INTEGER);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't2', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t2 VALUES (5);
+
+> DROP SCHEMA IF EXISTS schema1
+
+> CREATE SCHEMA schema1
+
+> CREATE SOURCE mz_source1
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1 AS t1_1, t2 AS t2_1);
+
+> CREATE SOURCE mz_source2
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR TABLES (t1 AS t1_2, t2 AS t2_2);
+
+> SELECT * FROM t1_1;
+1
+
+> SELECT * FROM t2_1;
+5
+
+> SELECT * FROM t1_2;
+1
+
+> SELECT * FROM t2_2;
+5
+
+$ sql-server-execute name=sql-server
+INSERT INTO t1 VALUES (2);
+INSERT INTO t1 VALUES (3);
+INSERT INTO t2 VALUES (6);
+INSERT INTO t2 VALUES (7);
+
+> SELECT * FROM t1_1;
+1
+2
+3
+
+> SELECT * FROM t2_1;
+5
+6
+7
+
+> SELECT * FROM t1_2;
+1
+2
+3
+
+> SELECT * FROM t2_2;
+5
+6
+7
+
+> DROP SCHEMA schema1 CASCADE;
diff --git a/test/sql-server-cdc-old-syntax/types-boolean.td b/test/sql-server-cdc-old-syntax/types-boolean.td
new file mode 100644
index 0000000000000..a5e9aa65443c2
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-boolean.td
@@ -0,0 +1,58 @@
+# 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.
+
+#
+# Test the BIT data type
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 BIT);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1 VALUES ('true'),('false');
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+# Insert the same data post-snapshot
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+
+> SELECT pg_typeof(f1) FROM t1 LIMIT 1;
+boolean
+
+> SELECT * FROM t1;
+true
+false
+true
+false
diff --git a/test/sql-server-cdc-old-syntax/types-bytea.td b/test/sql-server-cdc-old-syntax/types-bytea.td
new file mode 100644
index 0000000000000..93aa959ad800f
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-bytea.td
@@ -0,0 +1,64 @@
+# 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.
+
+#
+# Test the BYTEA data type
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 varbinary(48));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1 VALUES (CAST(NULL AS VARBINARY(48))), (CAST('' AS VARBINARY(48))), (CONVERT(VARBINARY(48), '0x00', 1)), (CONVERT(VARBINARY(48), '0xABCDEF1230', 1));
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+# Insert the same data post-snapshot
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+
+> SELECT pg_typeof(f1) FROM t1 LIMIT 1;
+bytea
+
+# Note that there is some expected loss of precision on the very large numbers
+
+> SELECT * FROM t1;
+""
+""
+<null>
+<null>
+"\\x00"
+"\\x00"
+"\\xab\\xcd\\xef\\x120"
+"\\xab\\xcd\\xef\\x120"
diff --git a/test/sql-server-cdc-old-syntax/types-character.td b/test/sql-server-cdc-old-syntax/types-character.td
new file mode 100644
index 0000000000000..6d3306bce2308
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-character.td
@@ -0,0 +1,72 @@
+# 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.
+
+#
+# Test the various character data types
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 VARCHAR(10), f2 CHAR(10), f3 VARCHAR(128), f4 VARCHAR(256));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1 VALUES ('abc', 'abc', 'abc', 'abc');
+INSERT INTO t1 VALUES ('abc ', 'abc ', 'abc ', 'abc ');
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+# Insert the same data post-snapshot
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+
+> SELECT pg_typeof(f1), pg_typeof(f2), pg_typeof(f3), pg_typeof(f4) FROM t1 LIMIT 1;
+"character varying" "character" "character varying" "character varying"
+
+> SELECT * FROM t1;
+"abc" "abc       " "abc" "abc"
+"abc" "abc       " "abc" "abc"
+"abc " "abc       " "abc " "abc "
+"abc " "abc       " "abc " "abc "
+
+# NB(ptravers): String comparison in SQL Server pads trailing whitespace for comparison
+# operators. The `LIKE` operator doesn't pad when performing comparisons. Padded 'abc '
+# must be run before unpadded 'abc' otherwise unpadded will match both.
+$ sql-server-execute name=sql-server
+UPDATE t1 SET f1 = 'xyz ', f2 = 'xyz ', f3 = 'xyz ', f4 = 'xyz ' WHERE f1 LIKE 'abc ';
+UPDATE t1 SET f1 = 'klm', f2 = 'klm', f3 = 'klm', f4 = 'klm' WHERE f1 LIKE 'abc';
+
+> SELECT * FROM t1;
+"klm" "klm       " "klm" "klm"
+"klm" "klm       " "klm" "klm"
+"xyz " "xyz       " "xyz " "xyz "
+"xyz " "xyz       " "xyz " "xyz "
diff --git a/test/sql-server-cdc-old-syntax/types-decimal.td b/test/sql-server-cdc-old-syntax/types-decimal.td
new file mode 100644
index 0000000000000..f892588eda82a
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-decimal.td
@@ -0,0 +1,59 @@
+# 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.
+
+#
+# Test various variants of DECIMAL
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 DECIMAL(20, 0), f2 DECIMAL(20, 20), f3 DECIMAL(20,10), f4 NUMERIC(20,10));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1 VALUES ('99999999999999999999', '0.99999999999999999999', '9999999999.9999999999', '9999999999.9999999999');
+INSERT INTO t1 VALUES ('-99999999999999999999', '-0.99999999999999999999', '-9999999999.9999999999', '-9999999999.9999999999');
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+# Insert the same data post-snapshot
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+
+> SELECT pg_typeof(f1), pg_typeof(f2), pg_typeof(f3), pg_typeof(f4) FROM t1 LIMIT 1;
+numeric numeric numeric numeric
+
+> SELECT * FROM t1;
+99999999999999999999 0.99999999999999999999 9999999999.9999999999 9999999999.9999999999
+-99999999999999999999 -0.99999999999999999999 -9999999999.9999999999 -9999999999.9999999999
+99999999999999999999 0.99999999999999999999 9999999999.9999999999 9999999999.9999999999
+-99999999999999999999 -0.99999999999999999999 -9999999999.9999999999 -9999999999.9999999999
diff --git a/test/sql-server-cdc-old-syntax/types-domain.td b/test/sql-server-cdc-old-syntax/types-domain.td
new file mode 100644
index 0000000000000..357be43582b75
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-domain.td
@@ -0,0 +1,46 @@
+# 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.
+
+#
+# Test domain types
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TYPE posint FROM int;
+CREATE RULE PositiveInt AS @value > 0;
+
+CREATE TABLE t1 (f1 POSINT);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1 VALUES (1);
+INSERT INTO t1 VALUES (2);
+
+! CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+contains:SQL SERVER source validation: column dbo.t1.f1 of type posint is not supported
diff --git a/test/sql-server-cdc-old-syntax/types-double.td b/test/sql-server-cdc-old-syntax/types-double.td
new file mode 100644
index 0000000000000..fc8bdf5f9c662
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-double.td
@@ -0,0 +1,70 @@
+# 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.
+
+#
+# Test the floating-point types
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 REAL);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1 VALUES (NULL);
+INSERT INTO t1 VALUES ('−16777216');
+INSERT INTO t1 VALUES (16777216);
+INSERT INTO t1 VALUES ('0.000000000000000000000000000001');
+INSERT INTO t1 VALUES ('-1111111111111111111111111111111');
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+# Insert the same data post-snapshot
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+
+> SELECT pg_typeof(f1) FROM t1 LIMIT 1;
+"real"
+
+# Note that there is some expected loss of precision on the very large numbers
+
+> SELECT * FROM t1;
+-1111111100000000000000000000000
+-1111111100000000000000000000000
+-16777216
+-16777216
+0.000000000000000000000000000001
+0.000000000000000000000000000001
+16777216
+16777216
+<null>
+<null>
diff --git a/test/sql-server-cdc-old-syntax/types-geometric.td b/test/sql-server-cdc-old-syntax/types-geometric.td
new file mode 100644
index 0000000000000..4bdadb4159596
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-geometric.td
@@ -0,0 +1,43 @@
+# 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.
+
+#
+# Test the geometric data types
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (polygon geometry);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1 VALUES (geometry::STGeomFromText('POLYGON ((0 0, 150 0, 150 150, 0 150, 0 0))', 0));
+
+! CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+contains:SQL SERVER source validation: column dbo.t1.polygon of type geometry is not supported
diff --git a/test/sql-server-cdc-old-syntax/types-integer.td b/test/sql-server-cdc-old-syntax/types-integer.td
new file mode 100644
index 0000000000000..7c183219ef418
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-integer.td
@@ -0,0 +1,58 @@
+# 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.
+
+#
+# Test various variants of integer types
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 SMALLINT, f2 INT, f3 BIGINT, f4 TINYINT);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1 VALUES (-32768, -2147483648, -9223372036854775808, 0);
+INSERT INTO t1 VALUES (32767, 2147483647, 9223372036854775807, 255);
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+# Insert the same data post-snapshot
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+
+> SELECT pg_typeof(f1), pg_typeof(f2), pg_typeof(f3), pg_typeof(f4) FROM t1 LIMIT 1;
+smallint integer bigint smallint
+
+> SELECT * FROM t1;
+-32768 -2147483648 -9223372036854775808 0
+32767 2147483647 9223372036854775807 255
+-32768 -2147483648 -9223372036854775808 0
+32767 2147483647 9223372036854775807 255
diff --git a/test/sql-server-cdc-old-syntax/types-money.td b/test/sql-server-cdc-old-syntax/types-money.td
new file mode 100644
index 0000000000000..dbfa58fd86fd3
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-money.td
@@ -0,0 +1,59 @@
+# 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.
+
+#
+# MONEY data type supported only as TEXT
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 MONEY);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+INSERT INTO t1 VALUES ('-922337203685477.58'), ('922337203685477.58');
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+# Insert the same data post-snapshot
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+
+# SQL Server diverges from Postgres by returning numeric money.
+> SELECT pg_typeof(f1) FROM t1 LIMIT 1;
+"numeric"
+
+> SELECT * FROM t1;
+"922337203685477.58"
+"922337203685477.58"
+"-922337203685477.58"
+"-922337203685477.58"
diff --git a/test/sql-server-cdc-old-syntax/types-numeric-arbitrary-precision.td b/test/sql-server-cdc-old-syntax/types-numeric-arbitrary-precision.td
new file mode 100644
index 0000000000000..cd3c7ec25c7a2
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-numeric-arbitrary-precision.td
@@ -0,0 +1,62 @@
+# 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.
+
+#
+# Arbitrary-precision NUMERIC supported only as TEXT
+#
+$ nop
+
+# FIXME(ptravers): https://github.com/MaterializeInc/database-issues/issues/9574
+# $ sql-server-connect name=sql-server
+# server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+# $ sql-server-execute name=sql-server
+# USE test;
+
+# > CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+# > DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+# > CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+#     HOST 'sql-server',
+#     PORT 1433,
+#     DATABASE test,
+#     USER '${arg.default-sql-server-user}',
+#     PASSWORD = SECRET sql_server_pass
+#   );
+
+# > VALIDATE CONNECTION sql_server_test_connection;
+
+# # Insert data pre-snapshot
+# $ sql-server-execute name=sql-server
+# CREATE TABLE t1 (f1 NUMERIC(28,10));
+# EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+
+# INSERT INTO t1 VALUES (NULL), ('0.000000000000000000000000000001'), ('-111111111111111111');
+
+# > CREATE SOURCE mz_source
+#   FROM SQL SERVER CONNECTION sql_server_test_connection
+#   FOR ALL TABLES;
+
+# > SELECT COUNT(*) > 0 FROM t1;
+# true
+
+# # Insert the same data post-snapshot
+# $ sql-server-execute name=sql-server
+# INSERT INTO t1 SELECT * FROM t1;
+
+# > SELECT pg_typeof(f1) FROM t1 LIMIT 1;
+# "numeric"
+
+# > SELECT * FROM t1;
+# <null>
+# <null>
+# -111111111111111111
+# -111111111111111111
+# 0.000000000000000000000000000001
+# 0.000000000000000000000000000001
diff --git a/test/sql-server-cdc-old-syntax/types-serial.td b/test/sql-server-cdc-old-syntax/types-serial.td
new file mode 100644
index 0000000000000..b88e02fc01195
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-serial.td
@@ -0,0 +1,54 @@
+# 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.
+
+#
+# Test various variants of INT IDENTITY(1,1)
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password};Database=test
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (f1 SMALLINT IDENTITY(1,1));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+SET IDENTITY_INSERT t1 ON;
+
+INSERT INTO t1 (f1) VALUES (32767);
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+# Insert the same data post-snapshot
+$ sql-server-execute name=sql-server
+INSERT INTO t1 (f1) SELECT * FROM t1;
+
+> SELECT pg_typeof(f1) FROM t1 LIMIT 1;
+smallint
+
+> SELECT * FROM t1;
+32767
+32767
diff --git a/test/sql-server-cdc-old-syntax/types-temporal-with-tz.td b/test/sql-server-cdc-old-syntax/types-temporal-with-tz.td
new file mode 100644
index 0000000000000..df98844eca908
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-temporal-with-tz.td
@@ -0,0 +1,52 @@
+# 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.
+
+#
+# Test the temporal data types with time zone
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE timestamp_table (f1 DATETIMEOFFSET);
+INSERT INTO timestamp_table VALUES (TODATETIMEOFFSET('2011-11-11 02:11:11.123456', '-00:00'));
+INSERT INTO timestamp_table VALUES (TODATETIMEOFFSET('2011-11-11 02:11:11.123456', '-00:00'));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'timestamp_table', @role_name = 'SA', @supports_net_changes = 0;
+
+> CREATE SOURCE mz_source_timestamp
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM timestamp_table;
+true
+
+> SELECT pg_typeof(f1) FROM timestamp_table LIMIT 1;
+"timestamp with time zone"
+
+> SELECT * FROM timestamp_table;
+"2011-11-11 02:11:11.123456 UTC"
+"2011-11-11 02:11:11.123456 UTC"
diff --git a/test/sql-server-cdc-old-syntax/types-temporal.td b/test/sql-server-cdc-old-syntax/types-temporal.td
new file mode 100644
index 0000000000000..4fe3a4fd3ba63
--- /dev/null
+++ b/test/sql-server-cdc-old-syntax/types-temporal.td
@@ -0,0 +1,52 @@
+# 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.
+
+#
+# Test the temporal data types
+#
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password};Database=test
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_connection CASCADE
+> CREATE CONNECTION sql_server_test_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+> VALIDATE CONNECTION sql_server_test_connection;
+
+# Insert data pre-snapshot
+$ sql-server-execute name=sql-server
+CREATE TABLE t1 (date_col DATE, time_col TIME, timestamp_col DATETIME2);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't1', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t1 VALUES (CAST('2011-11-11' AS DATE), CAST('11:11:11.123456' AS TIME), CAST('2011-11-11 11:11:11.123456' AS DATETIME2));
+
+> CREATE SOURCE mz_source
+  FROM SQL SERVER CONNECTION sql_server_test_connection
+  FOR ALL TABLES;
+
+> SELECT COUNT(*) > 0 FROM t1;
+true
+
+# Insert the same data post-snapshot
+$ sql-server-execute name=sql-server
+INSERT INTO t1 SELECT * FROM t1;
+
+> SELECT pg_typeof(date_col), pg_typeof(time_col), pg_typeof(timestamp_col) FROM t1 LIMIT 1;
+date time "timestamp without time zone"
+
+> SELECT * FROM t1;
+2011-11-11 11:11:11.123456 "2011-11-11 11:11:11.123456"
+2011-11-11 11:11:11.123456 "2011-11-11 11:11:11.123456"
diff --git a/test/sql-server-cdc/mzcompose.py b/test/sql-server-cdc/mzcompose.py
index 96a92ab639e3e..f37c366220708 100644
--- a/test/sql-server-cdc/mzcompose.py
+++ b/test/sql-server-cdc/mzcompose.py
@@ -185,7 +185,7 @@ def workflow_snapshot_consistency(
     update_val_offset = 100000
     insert_delete = lambda i: dedent(
         f"""
-        INSERT INTO t1 VALUES (999999999,666666666), ({i+update_id_offset}, {i+update_val_offset});
+        INSERT INTO t1 VALUES (999999999,666666666), ({i + update_id_offset}, {i + update_val_offset});
         DELETE FROM t1 WHERE id = 999999999;
         """
     )
@@ -234,7 +234,7 @@ def concurrent_updates(c: Composition) -> None:
         args=["--no-reset"],
         input=dedent(
             f"""
-            > SELECT COUNT(*) >= {update_rows+initial_rows}, MIN(id), MAX(id) >= {update_rows + update_id_offset - 1} FROM t1;
+            > SELECT COUNT(*) >= {update_rows + initial_rows}, MIN(id), MAX(id) >= {update_rows + update_id_offset - 1} FROM t1;
             true 1 true
             """
         ),