use "as binary" conversion to join info schema tables to resolve any collation issues (#4406)

Author: dtunikov

flow/connectors/mysql/cdc.go

@@ -81,32 +81,18 @@ func (c *MySqlConnector) getTableSchemaForTable(
 		return nil, err
 	}
 
-	pkSeqRs, err := c.Execute(ctx, fmt.Sprintf(`
-		select column_name, seq_in_index
-		from information_schema.statistics
-		where table_schema = '%s' and table_name = '%s' and index_name = 'PRIMARY'`,
-		mysql.Escape(qualifiedTable.Namespace), mysql.Escape(qualifiedTable.Table)))
-	if err != nil {
-		return nil, err
-	}
-	pkSeqMap := make(map[string]int64, pkSeqRs.RowNumber())
-	for idx := range pkSeqRs.RowNumber() {
-		name, err := pkSeqRs.GetString(idx, 0)
-		if err != nil {
-			return nil, err
-		}
-		seq, err := pkSeqRs.GetInt(idx, 1)
-		if err != nil {
-			return nil, err
-		}
-		pkSeqMap[name] = seq
-	}
-
+	// CAST(... AS BINARY) forces a case-sensitive, collation-independent comparison on the join keys so it works with
+	// lower_case_table_names=0. The LEFT JOIN leaves seq_in_index NULL for non-primary-key columns.
 	rs, err := c.Execute(ctx, fmt.Sprintf(`
-		select column_name, column_type, is_nullable, numeric_precision, numeric_scale
-		from information_schema.columns
-		where table_schema = '%s' and table_name = '%s'
-		order by ordinal_position`,
+		select c.column_name, c.column_type, c.is_nullable, c.numeric_precision, c.numeric_scale, s.seq_in_index
+		from information_schema.columns c
+		left join information_schema.statistics s
+			on cast(s.table_schema as binary) = cast(c.table_schema as binary)
+			and cast(s.table_name as binary) = cast(c.table_name as binary)
+			and cast(s.column_name as binary) = cast(c.column_name as binary)
+			and s.index_name = 'PRIMARY'
+		where c.table_schema = '%s' and c.table_name = '%s'
+		order by c.ordinal_position`,
 		mysql.Escape(qualifiedTable.Namespace), mysql.Escape(qualifiedTable.Table)))
 	if err != nil {
 		return nil, err
@@ -161,7 +147,15 @@ func (c *MySqlConnector) getTableSchemaForTable(
 		}
 		columns = append(columns, column)
 
-		if seq, ok := pkSeqMap[columnName]; ok {
+		seqIsNull, err := rs.IsNull(idx, 5)
+		if err != nil {
+			return nil, err
+		}
+		if !seqIsNull {
+			seq, err := rs.GetInt(idx, 5)
+			if err != nil {
+				return nil, err
+			}
 			primaryEntries = append(primaryEntries, pkEntry{name: columnName, seqInIndex: seq})
 		}
 	}

flow/connectors/mysql/cdc_test.go

@@ -106,3 +106,55 @@ func TestGetTableSchemaCaseSensitiveIdentifiers(t *testing.T) {
 	assertSchema(schemas[ulTable], []string{"id"}, "id", "ul")
 	assertSchema(schemas[uuTable], []string{"id"}, "id", "uu")
 }
+
+func TestGetTableSchemaPrimaryKeyVariants(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+	connector := newTestConnector(t, ctx)
+
+	dbName := "pk_variants_test"
+	createTestDB(t, ctx, connector, dbName)
+
+	exec := func(sql string) {
+		t.Helper()
+		_, err := connector.Execute(ctx, sql)
+		require.NoError(t, err)
+	}
+
+	assertSchema := func(schema *protos.TableSchema, pk []string, cols ...string) {
+		t.Helper()
+		require.NotNil(t, schema)
+		if len(pk) == 0 {
+			require.Empty(t, schema.PrimaryKeyColumns)
+		} else {
+			require.Equal(t, pk, schema.PrimaryKeyColumns)
+		}
+		require.Len(t, schema.Columns, len(cols))
+		for i, name := range cols {
+			require.Equal(t, name, schema.Columns[i].Name)
+		}
+	}
+
+	compositeTable := dbName + ".composite_pk"
+	noPKTable := dbName + ".no_pk"
+	mixedCaseTable := dbName + ".mixed_case_pk"
+
+	// Composite PK whose key order (b, a) differs from column definition order (a, b, c),
+	// exercising the seq_in_index sort.
+	exec(fmt.Sprintf("CREATE TABLE %s (a INT, b INT, c INT, PRIMARY KEY (b, a))", compositeTable))
+	// Table without a primary key, exercising the LEFT JOIN's NULL seq_in_index path.
+	exec(fmt.Sprintf("CREATE TABLE %s (a INT, b TEXT)", noPKTable))
+	// PK on a mixed-case column name, exercising the case-preserving column_name join.
+	exec(fmt.Sprintf("CREATE TABLE %s (`MyId` INT PRIMARY KEY, val TEXT)", mixedCaseTable))
+
+	schemas, err := connector.GetTableSchema(ctx, nil, shared.InternalVersion_Latest, protos.TypeSystem_Q,
+		[]*protos.TableMapping{
+			{SourceTableIdentifier: compositeTable},
+			{SourceTableIdentifier: noPKTable},
+			{SourceTableIdentifier: mixedCaseTable},
+		})
+	require.NoError(t, err)
+	assertSchema(schemas[compositeTable], []string{"b", "a"}, "a", "b", "c")
+	assertSchema(schemas[noPKTable], nil, "a", "b")
+	assertSchema(schemas[mixedCaseTable], []string{"MyId"}, "MyId", "val")
+}