fix(mtree): fetch complex-typed columns as text so their diffs repair

internal/consistency/diff/table_diff.go

@@ -510,18 +510,8 @@ func (t *TableDiffTask) fetchRows(nodeName string, r Range) ([]types.OrderedMap,
 	selectCols = append(selectCols, "pg_xact_commit_timestamp(xmin) as commit_ts", "to_json(pg_xact_commit_timestamp_origin(xmin))->>'roident' as node_origin")
 
 	for _, colName := range t.Cols {
-		colType := colTypes[colName]
 		quotedColName := pgx.Identifier{colName}.Sanitize()
-
-		// We cast user defined types and arrays to TEXT to avoid scan errors with unknown OIDs
-		if strings.HasSuffix(colType, "[]") ||
-			strings.Contains(strings.ToLower(colType), "json") ||
-			strings.Contains(strings.ToLower(colType), "bytea") ||
-			!utils.IsKnownScalarType(colType) {
-			selectCols = append(selectCols, fmt.Sprintf("%s::TEXT AS %s", quotedColName, quotedColName))
-		} else {
-			selectCols = append(selectCols, quotedColName)
-		}
+		selectCols = append(selectCols, utils.SelectColExpr(quotedColName, colTypes[colName]))
 	}
 
 	selectColsStr := strings.Join(selectCols, ", ")

internal/consistency/mtree/merkle.go

@@ -474,6 +474,18 @@ func (m *MerkleTreeTask) processWorkItem(work CompareRangesWorkItem, pool1, pool
 	// so this is not canonicalised (unlike table_diff.pairKeyFor).
 	nodePairKey := fmt.Sprintf("%s/%s", work.Node1["Name"], work.Node2["Name"])
 
+	// Build the row-fetch SELECT list with the same cast policy as the classic
+	// engine (SelectColExpr): complex/unknown types arrive as Postgres text
+	// instead of opaque driver structs, so the diff report stays repairable.
+	refTypes := m.ColTypes["_ref"]
+	colExprs := make([]string, 0, len(m.Cols))
+	for _, c := range m.Cols {
+		colExprs = append(colExprs, utils.SelectColExpr(pgx.Identifier{c}.Sanitize(), refTypes[c]))
+	}
+	selectCols := "pg_xact_commit_timestamp(xmin) as commit_ts, " +
+		"to_json(pg_xact_commit_timestamp_origin(xmin))->>'roident' as node_origin, " +
+		strings.Join(colExprs, ", ")
+
 	if isComposite {
 		for i := 0; i < len(mismatchedComposite); i += fetchBatchSize {
 			if m.pairCapReached(nodePairKey) {
@@ -487,7 +499,7 @@ func (m *MerkleTreeTask) processWorkItem(work CompareRangesWorkItem, pool1, pool
 			}
 			batch := mismatchedComposite[i:end]
 
-			q, qArgs := buildFetchRowsSQLComposite(m.Schema, m.Table, m.Key, orderByStr, batch)
+			q, qArgs := buildFetchRowsSQLComposite(m.Schema, m.Table, m.Key, selectCols, orderByStr, batch)
 
 			r1, err := pool1.Query(m.Ctx, q, qArgs...) // nosemgrep
 			if err != nil {
@@ -524,7 +536,7 @@ func (m *MerkleTreeTask) processWorkItem(work CompareRangesWorkItem, pool1, pool
 			}
 			batch := mismatchedSimple[i:end]
 
-			q, qArgs := buildFetchRowsSQLSimple(m.Schema, m.Table, m.Key[0], orderByStr, batch)
+			q, qArgs := buildFetchRowsSQLSimple(m.Schema, m.Table, m.Key[0], selectCols, orderByStr, batch)
 
 			r1, err := pool1.Query(m.Ctx, q, qArgs...) // nosemgrep
 			if err != nil {
@@ -840,7 +852,7 @@ func splitCompositeKey(k string) []any {
 	return res
 }
 
-func buildFetchRowsSQLSimple(schema, table, pk string, orderBy string, keys []any) (string, []any) {
+func buildFetchRowsSQLSimple(schema, table, pk, selectCols, orderBy string, keys []any) (string, []any) {
 	placeholders := make([]string, len(keys))
 	args := make([]any, len(keys))
 	for i := range keys {
@@ -849,12 +861,11 @@ func buildFetchRowsSQLSimple(schema, table, pk string, orderBy string, keys []an
 	}
 	qualifiedTable := fmt.Sprintf("%s.%s", pgx.Identifier{schema}.Sanitize(), pgx.Identifier{table}.Sanitize())
 	where := fmt.Sprintf("%s IN (%s)", pgx.Identifier{pk}.Sanitize(), strings.Join(placeholders, ","))
-	selectCols := "pg_xact_commit_timestamp(xmin) as commit_ts, to_json(pg_xact_commit_timestamp_origin(xmin))->>'roident' as node_origin, *"
 	q := fmt.Sprintf("SELECT %s FROM %s WHERE %s ORDER BY %s", selectCols, qualifiedTable, where, orderBy)
 	return q, args
 }
 
-func buildFetchRowsSQLComposite(schema, table string, pk []string, orderBy string, keys [][]any) (string, []any) {
+func buildFetchRowsSQLComposite(schema, table string, pk []string, selectCols, orderBy string, keys [][]any) (string, []any) {
 	tupleCols := make([]string, len(pk))
 	for i, k := range pk {
 		tupleCols[i] = pgx.Identifier{k}.Sanitize()
@@ -873,7 +884,6 @@ func buildFetchRowsSQLComposite(schema, table string, pk []string, orderBy strin
 	}
 	qualifiedTable := fmt.Sprintf("%s.%s", pgx.Identifier{schema}.Sanitize(), pgx.Identifier{table}.Sanitize())
 	where := fmt.Sprintf("( %s ) IN ( %s )", strings.Join(tupleCols, ","), strings.Join(tuples, ","))
-	selectCols := "pg_xact_commit_timestamp(xmin) as commit_ts, to_json(pg_xact_commit_timestamp_origin(xmin))->>'roident' as node_origin, *"
 	q := fmt.Sprintf("SELECT %s FROM %s WHERE %s ORDER BY %s", selectCols, qualifiedTable, where, orderBy)
 	return q, args
 }

pkg/common/utils.go

@@ -807,9 +807,13 @@ func ConvertToPgxType(val any, pgType string) (any, error) {
 		return nil, fmt.Errorf("expected UUID string for %s, got %T", pgType, val)
 
 	default:
-		// For all other types, fall back to string representations to keep repairs viable.
+		// String pass-through is the designed path for every type without a
+		// case above: the diff engines fetch such columns as ::TEXT, and
+		// Postgres parses its own text form on input (geometric, network,
+		// range, enum, xml, ...). Only non-string values indicate a value that
+		// bypassed that normalisation and deserve a warning.
 		if s, ok := val.(string); ok {
-			log.Printf("Warning: Passing raw string value '%s' for unknown or complex pgType '%s'", s, pgType)
+			logger.Debug("passing text value through for pgType %s", pgType)
 			return s, nil
 		}
 		if stringer, ok := val.(fmt.Stringer); ok {
@@ -990,6 +994,24 @@ func SafeCut(s string, n int) string {
 	return s[:n]
 }
 
+// SelectColExpr returns the SELECT-list expression for one column of a diff
+// row fetch: the quoted name for known scalar types the driver scans
+// losslessly, or a ::TEXT cast (aliased back to the column name) for arrays,
+// json, bytea and any type the driver would surface as an opaque Go struct
+// (geometric, network, range, enum, xml, ...). Postgres parses its own text
+// form on repair, so text values round-trip for every type without per-type
+// conversion code. Both diff engines must build their row-fetch SELECTs with
+// it so reports stay engine-independent and repairable.
+func SelectColExpr(quotedCol, colType string) string {
+	if strings.HasSuffix(colType, "[]") ||
+		strings.Contains(strings.ToLower(colType), "json") ||
+		strings.Contains(strings.ToLower(colType), "bytea") ||
+		!IsKnownScalarType(colType) {
+		return fmt.Sprintf("%s::TEXT AS %s", quotedCol, quotedCol)
+	}
+	return quotedCol
+}
+
 func IsKnownScalarType(colType string) bool {
 	// "time with time zone" (timetz) is NOT registered in the pgx v5
 	// default type map, so it cannot be scanned into *interface{}. Exclude it

pkg/common/utils_test.go

@@ -123,6 +123,34 @@ func TestConvertToPgxType_Money(t *testing.T) {
 	require.Equal(t, "$532.96", val)
 }
 
+// Complex and unknown types are fetched as ::TEXT so their Postgres text form
+// round-trips through the diff report; known scalars stay native.
+func TestSelectColExpr(t *testing.T) {
+	tests := []struct {
+		colType string
+		want    string
+	}{
+		{"integer", `"c"`},
+		{"timestamp without time zone", `"c"`},
+		{"uuid", `"c"`},
+		{"integer[]", `"c"::TEXT AS "c"`},
+		{"jsonb", `"c"::TEXT AS "c"`},
+		{"bytea", `"c"::TEXT AS "c"`},
+		{"point", `"c"::TEXT AS "c"`},
+		{"int4range", `"c"::TEXT AS "c"`},
+		{"mood_enum", `"c"::TEXT AS "c"`},
+		{"xml", `"c"::TEXT AS "c"`},
+		{"bit(8)", `"c"::TEXT AS "c"`},
+		{"inet", `"c"::TEXT AS "c"`},
+		{"time with time zone", `"c"::TEXT AS "c"`},
+	}
+	for _, tt := range tests {
+		t.Run(tt.colType, func(t *testing.T) {
+			require.Equal(t, tt.want, SelectColExpr(`"c"`, tt.colType))
+		})
+	}
+}
+
 // Concurrent normalisation is race-free (a pgtype.Map must not be shared
 // across goroutines; run with -race).
 func TestNormalizeScannedValue_ConcurrentInterval(t *testing.T) {

tests/integration/mtree_typed_repair_test.go

@@ -23,34 +23,53 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// typedRepairDDL covers the type classes repair must round-trip: temporal,
+// money, uuid, interval, numeric, plus the complex ones the driver scans as
+// opaque structs (geometric, range, bit, network, enum, xml).
+const typedRepairDDL = ` (id BIGINT PRIMARY KEY, name VARCHAR(100), col_time TIME, col_timetz TIMETZ,
+	col_money MONEY, col_uuid UUID, col_interval INTERVAL, col_num NUMERIC(12,4),
+	col_point POINT, col_range INT4RANGE, col_bit BIT(8), col_inet INET,
+	col_xml XML, col_mood typed_repair_mood)`
+
+const typedRepairSeedSQL = ` (id, name, col_time, col_timetz, col_money, col_uuid, col_interval, col_num,
+		col_point, col_range, col_bit, col_inet, col_xml, col_mood)
+	SELECT i,
+	       'name_' || i,
+	       TIME '00:00:00' + (i * 137 || ' seconds')::interval,
+	       TIMETZ '00:00:00+02' + (i * 91 || ' seconds')::interval,
+	       (i * 13.37)::numeric::money,
+	       ('00000000-0000-0000-0000-' || lpad(i::text, 12, '0'))::uuid,
+	       (i || ' hours 30 minutes')::interval,
+	       i * 1.5,
+	       point(i, i * 2),
+	       int4range(i, i + 100),
+	       (i % 256)::bit(8),
+	       ('10.0.' || (i % 256) || '.' || (i % 200 + 1))::inet,
+	       ('<item id="' || i || '"><name>item_' || i || '</name></item>')::xml,
+	       (ARRAY['happy','sad','neutral']::typed_repair_mood[])[(i % 3) + 1]
+	FROM generate_series(1, 100) AS i`
+
 // seedTypedRepairTable creates the typed-columns table on both nodes and seeds
 // rows on n1 only -- the diverged state a disabled subscription leaves behind.
 func seedTypedRepairTable(t *testing.T, ctx context.Context, env *testEnv, safe string) {
 	t.Helper()
 	pools := []*pgxpool.Pool{env.N1Pool, env.N2Pool}
 	for _, pool := range pools {
-		_, err := pool.Exec(ctx, "CREATE TABLE IF NOT EXISTS "+safe+ // nosemgrep
-			" (id BIGINT PRIMARY KEY, name VARCHAR(100), col_time TIME, col_timetz TIMETZ,"+
-			" col_money MONEY, col_uuid UUID, col_interval INTERVAL, col_num NUMERIC(12,4))")
+		_, err := pool.Exec(ctx, "DROP TYPE IF EXISTS typed_repair_mood CASCADE")
+		require.NoError(t, err)
+		_, err = pool.Exec(ctx, "CREATE TYPE typed_repair_mood AS ENUM ('happy','sad','neutral')")
+		require.NoError(t, err)
+		_, err = pool.Exec(ctx, "CREATE TABLE IF NOT EXISTS "+safe+typedRepairDDL) // nosemgrep
 		require.NoError(t, err)
 	}
 	t.Cleanup(func() {
 		for _, pool := range pools {
 			_, _ = pool.Exec(ctx, "DROP TABLE IF EXISTS "+safe+" CASCADE") // nosemgrep
+			_, _ = pool.Exec(ctx, "DROP TYPE IF EXISTS typed_repair_mood CASCADE")
 		}
 	})
 
-	seedSQL := "INSERT INTO " + safe + ` (id, name, col_time, col_timetz, col_money, col_uuid, col_interval, col_num)
-		SELECT i,
-		       'name_' || i,
-		       TIME '00:00:00' + (i * 137 || ' seconds')::interval,
-		       TIMETZ '00:00:00+02' + (i * 91 || ' seconds')::interval,
-		       (i * 13.37)::numeric::money,
-		       ('00000000-0000-0000-0000-' || lpad(i::text, 12, '0'))::uuid,
-		       (i || ' hours 30 minutes')::interval,
-		       i * 1.5
-		FROM generate_series(1, 100) AS i`
-	_, err := env.N1Pool.Exec(ctx, seedSQL) // nosemgrep
+	_, err := env.N1Pool.Exec(ctx, "INSERT INTO "+safe+typedRepairSeedSQL) // nosemgrep
 	require.NoError(t, err)
 	for _, pool := range pools {
 		_, err := pool.Exec(ctx, "ANALYZE "+safe) // nosemgrep
@@ -92,8 +111,9 @@ func typedRepairFingerprint(t *testing.T, ctx context.Context, pool *pgxpool.Poo
 	return
 }
 
-// A diff produced by the Merkle-tree engine repairs cleanly on a table with
-// time, timetz, money, uuid and interval columns, and the nodes converge.
+// A diff produced by the Merkle-tree engine repairs cleanly on a table
+// spanning temporal, money, uuid, interval, numeric, geometric, range, bit,
+// network, enum and xml columns, and the nodes converge.
 func TestMtreeDiffRepairTypedColumns(t *testing.T) {
 	ctx := context.Background()
 	env := newSpockEnv()