diff --git a/internal/consistency/diff/table_diff.go b/internal/consistency/diff/table_diff.go index c88e2c8..47f7f2c 100644 --- a/internal/consistency/diff/table_diff.go +++ b/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, ", ") diff --git a/internal/consistency/mtree/merkle.go b/internal/consistency/mtree/merkle.go index f7775b0..1127ed6 100644 --- a/internal/consistency/mtree/merkle.go +++ b/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 } diff --git a/pkg/common/utils.go b/pkg/common/utils.go index 5c53c32..95883f1 100644 --- a/pkg/common/utils.go +++ b/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 diff --git a/pkg/common/utils_test.go b/pkg/common/utils_test.go index 4f8ca14..b033535 100644 --- a/pkg/common/utils_test.go +++ b/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) { diff --git a/tests/integration/mtree_typed_repair_test.go b/tests/integration/mtree_typed_repair_test.go index 970ebfa..b8edf6a 100644 --- a/tests/integration/mtree_typed_repair_test.go +++ b/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_' || i || '')::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()