Rewrite sync_column_key_info as EXISTS-based (fixes Bug A / 6-7 tests)

The previous IFNULL form kept the aggregate-without-GROUP-BY construct
combined with a bare correlated column (ELSE c.is_nullable). That
relies on SQLite's "bare column alongside aggregate" extension which
Turso does not implement — empirically corrupts is_nullable in both
directions: NOT NULL columns flip to DEFAULT NULL, and vice versa.

EXISTS is a pure boolean predicate with no aggregate bare-column
coupling; Turso handles it correctly. Six SHOW CREATE TABLE tests
(testShowCreateTable1/Quoted/WithComments/PreservesDoubleUnderscore/
PreservesKeyLengths) and likely testReconstructTableFromMysqlDataTypesCache
should flip to passing. The six Translation_Tests will still diff
against expectation strings but that's an assertion artefact, not
behavior.

Author: JanJakes

.github/workflows/phpunit-tests-turso.yml

@@ -761,41 +761,53 @@ jobs:
               "\t\t);"
           )
           assert old in src, 'sync_column_key_info UPDATE not found'
-          # Wrap each subquery in IFNULL(..., c.<col>) because Turso's
-          # aggregate-without-GROUP-BY returns zero rows when the WHERE
-          # matches nothing (SQL standard requires one row with NULL
-          # aggregates). Zero rows from a scalar subquery means NULL, which
-          # violates the NOT NULL constraint on is_nullable.
+          # The aggregate-without-GROUP-BY form combined with a bare
+          # correlated column (`ELSE c.is_nullable`) in the CASE relies on
+          # SQLite's "bare column alongside aggregate" extension. Turso does
+          # not implement that the same way and corrupts is_nullable in both
+          # directions. Rewrite with EXISTS subqueries — pure boolean form
+          # that Turso handles correctly.
           new = "\n".join([
               "\t\t$columns_table    = $this->connection->quote_identifier( $columns_table_name );",
               "\t\t$statistics_table = $this->connection->quote_identifier( $statistics_table_name );",
               "\t\t$this->connection->query(",
               "\t\t\t\"",
               "\t\t\t\tUPDATE $columns_table AS c",
-              "\t\t\t\tSET column_key = IFNULL((",
-              "\t\t\t\t\tSELECT",
-              "\t\t\t\t\t\tCASE",
-              "\t\t\t\t\t\t\tWHEN MAX(s.index_name = 'PRIMARY') THEN 'PRI'",
-              "\t\t\t\t\t\t\tWHEN MAX(s.non_unique = 0 AND s.seq_in_index = 1) THEN 'UNI'",
-              "\t\t\t\t\t\t\tWHEN MAX(s.seq_in_index = 1) THEN 'MUL'",
-              "\t\t\t\t\t\t\tELSE ''",
-              "\t\t\t\t\t\tEND",
-              "\t\t\t\t\tFROM $statistics_table AS s",
-              "\t\t\t\t\tWHERE s.table_schema = c.table_schema",
-              "\t\t\t\t\tAND s.table_name = c.table_name",
-              "\t\t\t\t\tAND s.column_name = c.column_name",
-              "\t\t\t\t), c.column_key),",
-              "\t\t\t\tis_nullable = IFNULL((",
-              "\t\t\t\t\tSELECT",
-              "\t\t\t\t\t\tCASE",
-              "\t\t\t\t\t\t\tWHEN MAX(s.index_name = 'PRIMARY') THEN 'NO'",
-              "\t\t\t\t\t\t\tELSE c.is_nullable",
-              "\t\t\t\t\t\tEND",
-              "\t\t\t\t\tFROM $statistics_table AS s",
-              "\t\t\t\t\tWHERE s.table_schema = c.table_schema",
-              "\t\t\t\t\tAND s.table_name = c.table_name",
-              "\t\t\t\t\tAND s.column_name = c.column_name",
-              "\t\t\t\t), c.is_nullable)",
+              "\t\t\t\tSET column_key = CASE",
+              "\t\t\t\t\tWHEN EXISTS (",
+              "\t\t\t\t\t\tSELECT 1 FROM $statistics_table AS s",
+              "\t\t\t\t\t\tWHERE s.table_schema = c.table_schema",
+              "\t\t\t\t\t\t  AND s.table_name   = c.table_name",
+              "\t\t\t\t\t\t  AND s.column_name  = c.column_name",
+              "\t\t\t\t\t\t  AND s.index_name   = 'PRIMARY'",
+              "\t\t\t\t\t) THEN 'PRI'",
+              "\t\t\t\t\tWHEN EXISTS (",
+              "\t\t\t\t\t\tSELECT 1 FROM $statistics_table AS s",
+              "\t\t\t\t\t\tWHERE s.table_schema = c.table_schema",
+              "\t\t\t\t\t\t  AND s.table_name   = c.table_name",
+              "\t\t\t\t\t\t  AND s.column_name  = c.column_name",
+              "\t\t\t\t\t\t  AND s.non_unique   = 0",
+              "\t\t\t\t\t\t  AND s.seq_in_index = 1",
+              "\t\t\t\t\t) THEN 'UNI'",
+              "\t\t\t\t\tWHEN EXISTS (",
+              "\t\t\t\t\t\tSELECT 1 FROM $statistics_table AS s",
+              "\t\t\t\t\t\tWHERE s.table_schema = c.table_schema",
+              "\t\t\t\t\t\t  AND s.table_name   = c.table_name",
+              "\t\t\t\t\t\t  AND s.column_name  = c.column_name",
+              "\t\t\t\t\t\t  AND s.seq_in_index = 1",
+              "\t\t\t\t\t) THEN 'MUL'",
+              "\t\t\t\t\tELSE ''",
+              "\t\t\t\tEND,",
+              "\t\t\t\tis_nullable = CASE",
+              "\t\t\t\t\tWHEN EXISTS (",
+              "\t\t\t\t\t\tSELECT 1 FROM $statistics_table AS s",
+              "\t\t\t\t\t\tWHERE s.table_schema = c.table_schema",
+              "\t\t\t\t\t\t  AND s.table_name   = c.table_name",
+              "\t\t\t\t\t\t  AND s.column_name  = c.column_name",
+              "\t\t\t\t\t\t  AND s.index_name   = 'PRIMARY'",
+              "\t\t\t\t\t) THEN 'NO'",
+              "\t\t\t\t\tELSE c.is_nullable",
+              "\t\t\t\tEND",
               "\t\t\t\tWHERE c.table_schema = ?",
               "\t\t\t\tAND c.table_name = ?",
               "\t\t\t\",",
@@ -804,7 +816,7 @@ jobs:
           ])
           src = src.replace(old, new, 1)
           open(path, 'w').write(src)
-          print('patched sync_column_key_info UPDATE')
+          print('patched sync_column_key_info UPDATE (EXISTS form)')
 
           # 2. Turso doesn't implement `PRAGMA foreign_key_check`. The driver
           #    runs it after every ALTER TABLE to validate foreign keys; its