diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go index 331e626e3..20ce9f92f 100644 --- a/flow/connectors/postgres/cdc.go +++ b/flow/connectors/postgres/cdc.go @@ -7,11 +7,13 @@ import ( "fmt" "log/slog" "maps" + "regexp" "slices" "strings" "sync" "sync/atomic" "time" + "unicode" "github.com/jackc/pglogrepl" "github.com/jackc/pgx/v5" @@ -1315,7 +1317,6 @@ func processRelationMessage[Items model.Items]( } } - var potentiallyNullableAddedColumns []string schemaDelta := &protos.TableSchemaDelta{ SrcTableName: p.srcTableIDNameMapping[currRel.RelationID], DstTableName: p.tableNameMapping[p.srcTableIDNameMapping[currRel.RelationID]].Name, @@ -1333,9 +1334,11 @@ func processRelationMessage[Items model.Items]( return !isExcluded } + addedColumnNames := make([]string, 0) addedColumnTypeOIDs := make([]uint32, 0) for _, column := range currRel.Columns { if isAddedColumnAndNotExcluded(column.Name) { + addedColumnNames = append(addedColumnNames, column.Name) addedColumnTypeOIDs = append(addedColumnTypeOIDs, column.DataType) } } @@ -1345,26 +1348,43 @@ func processRelationMessage[Items model.Items]( return nil, fmt.Errorf("error getting schema names for added column types: %w", err) } + // relation messages carry neither nullability nor defaults, so they come from pg_catalog. That + // reflects the table as it stands now rather than as of the DDL, which only differs if the column + // was altered again in between. + addedColumnCatalogInfo, err := p.fetchAddedColumnCatalogInfo(ctx, currRel.RelationID, addedColumnNames) + if err != nil { + return nil, err + } + for _, column := range currRel.Columns { // not present in previous relation message, but in current one, so added. if isAddedColumnAndNotExcluded(column.Name) { - schemaDelta.AddedColumns = append(schemaDelta.AddedColumns, &protos.FieldDescription{ + catalogInfo := addedColumnCatalogInfo[column.Name] + var defaultExpr *string + // destinations on the PG type system splice the column type in verbatim, so a literal + // rendered for a QValueKind would not fit their DDL + if catalogInfo.defaultExpr != "" && prevSchema.System == protos.TypeSystem_Q { + if literal, ok := defaultExprFromPostgresDefault( + catalogInfo.defaultExpr, types.QValueKind(currRelMap[column.Name]), + ); ok { + defaultExpr = &literal + } + } + addedColumn := &protos.FieldDescription{ Name: column.Name, Type: currRelMap[column.Name], TypeModifier: column.TypeModifier, - Nullable: false, + Nullable: !catalogInfo.notNull, TypeSchemaName: typeSchemaNameMapping[column.DataType], - }) - // pg does not send nullable info, only whether column is part of replica identity - // After loop we will correct this based on pg_catalog, - // but can skip specific scenario where replident is default or index - if currRel.ReplicaIdentity == uint8(ReplicaIdentityFull) || - currRel.ReplicaIdentity == uint8(ReplicaIdentityNothing) || column.Flags == 0 { - potentiallyNullableAddedColumns = append(potentiallyNullableAddedColumns, utils.QuoteLiteral(column.Name)) + DefaultExpr: defaultExpr, } + schemaDelta.AddedColumns = append(schemaDelta.AddedColumns, addedColumn) p.logger.Info("Detected added column", - slog.String("columnName", column.Name), - slog.String("columnType", currRelMap[column.Name]), + slog.String("columnName", addedColumn.Name), + slog.String("columnType", addedColumn.Type), + slog.Bool("nullable", addedColumn.Nullable), + slog.String("sourceDefault", catalogInfo.defaultExpr), + slog.String("default", addedColumn.GetDefaultExpr()), slog.String("relationName", schemaDelta.SrcTableName)) } else if _, inPrevRel := prevRelMap[column.Name]; !inPrevRel { // Column is added but excluded @@ -1396,36 +1416,6 @@ func processRelationMessage[Items model.Items]( schemaDelta.SrcTableName)) } } - if len(potentiallyNullableAddedColumns) > 0 { - p.logger.Info("Checking for potentially nullable columns in table", - slog.String("tableName", schemaDelta.SrcTableName), - slog.Any("potentiallyNullable", potentiallyNullableAddedColumns)) - - rows, err := p.conn.Query( - ctx, - fmt.Sprintf( - "select attname from pg_attribute where attrelid=$1 and attname in (%s) and not attnotnull", - strings.Join(potentiallyNullableAddedColumns, ","), - ), - currRel.RelationID, - ) - if err != nil { - return nil, fmt.Errorf("error looking up column nullable info for schema change: %w", err) - } - - attnames, err := pgx.CollectRows[string](rows, pgx.RowTo) - if err != nil { - return nil, fmt.Errorf("error collecting rows for column nullable info for schema change: %w", err) - } - for _, column := range schemaDelta.AddedColumns { - if slices.Contains(attnames, column.Name) { - column.Nullable = true - p.logger.Info(fmt.Sprintf("Detected column %s in table %s as nullable", - column.Name, schemaDelta.SrcTableName)) - } - } - } - p.relationMessageMapping[currRel.RelationID] = currRel // only log audit if there is actionable delta if len(schemaDelta.AddedColumns) > 0 { @@ -1437,6 +1427,163 @@ func processRelationMessage[Items model.Items]( return nil, nil } +type addedColumnCatalogInfo struct { + // as rendered by pg_get_expr, empty when the column has no default + defaultExpr string + notNull bool +} + +// fetchAddedColumnCatalogInfo reads the nullability and DEFAULT of the named columns from pg_catalog. +func (c *PostgresConnector) fetchAddedColumnCatalogInfo( + ctx context.Context, relID uint32, columnNames []string, +) (map[string]addedColumnCatalogInfo, error) { + if len(columnNames) == 0 { + return nil, nil + } + + // generated columns keep their expression in pg_attrdef too, and it is not a default + rows, err := c.conn.Query(ctx, `SELECT a.attname, a.attnotnull, + CASE WHEN a.attgenerated = '' THEN pg_catalog.pg_get_expr(d.adbin, d.adrelid) END + FROM pg_catalog.pg_attribute a + LEFT JOIN pg_catalog.pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum + WHERE a.attrelid = $1 AND a.attname = ANY($2) AND a.attnum > 0 AND NOT a.attisdropped`, + relID, columnNames) + if err != nil { + return nil, fmt.Errorf("error looking up added column info for schema change: %w", err) + } + + catalogInfo := make(map[string]addedColumnCatalogInfo, len(columnNames)) + var name string + var notNull bool + var defaultExpr pgtype.Text + if _, err := pgx.ForEachRow(rows, []any{&name, ¬Null, &defaultExpr}, func() error { + catalogInfo[name] = addedColumnCatalogInfo{defaultExpr: defaultExpr.String, notNull: notNull} + return nil + }); err != nil { + return nil, fmt.Errorf("error collecting added column info for schema change: %w", err) + } + return catalogInfo, nil +} + +// pg renders numbers as digits with an optional sign, fraction and exponent +var pgNumericLiteralRegex = regexp.MustCompile(`^[+-]?(\d+(\.\d*)?|\.\d+)([eE][+-]?\d+)?$`) + +// defaultExprFromPostgresDefault renders a Postgres column DEFAULT, as returned by pg_get_expr, as a +// SQL literal that destinations can splice into DDL verbatim. Only constants of a type whose text form +// carries over are translated; everything else is declined, leaving the column without a default. +func defaultExprFromPostgresDefault(rendered string, qkind types.QValueKind) (string, bool) { + value, quoted, ok := parsePostgresDefaultConstant(rendered) + if !ok { + return "", false + } + + switch qkind { + case types.QValueKindBoolean: + if quoted || (value != "true" && value != "false") { + return "", false + } + return value, true + + case types.QValueKindInt8, types.QValueKindInt16, types.QValueKindInt32, types.QValueKindInt64, + types.QValueKindInt256, types.QValueKindUInt8, types.QValueKindUInt16, types.QValueKindUInt32, + types.QValueKindUInt64, types.QValueKindUInt256, types.QValueKindFloat32, types.QValueKindFloat64, + types.QValueKindNumeric: + // negative numbers and the wider integers arrive quoted; unquoting keeps the DDL readable. + // Non-finite values such as 'NaN' are left behind, destinations disagree on those. + if !pgNumericLiteralRegex.MatchString(value) { + return "", false + } + return value, true + + case types.QValueKindTimestampTZ: + // our connections run with timezone=UTC so pg renders this offset, and the destination column + // carries no zone of its own, making the offset something it may well refuse to parse + value, ok = strings.CutSuffix(value, "+00") + if !ok { + return "", false + } + return quoteDefaultLiteral(value, quoted) + + case types.QValueKindString, types.QValueKindEnum, types.QValueKindQChar, types.QValueKindUUID, + types.QValueKindJSON, types.QValueKindJSONB, types.QValueKindHStore, types.QValueKindINET, + types.QValueKindCIDR, types.QValueKindMacaddr, types.QValueKindDate, types.QValueKindTimestamp: + return quoteDefaultLiteral(value, quoted) + + default: + // bytea, arrays, interval, time and the geo types all have a text form the destination reads + // differently, if at all + return "", false + } +} + +// quoteDefaultLiteral quotes value for splicing into DDL, declining anything whose escaping is not +// portable across dialects. +func quoteDefaultLiteral(value string, quoted bool) (string, bool) { + if !quoted || strings.ContainsFunc(value, func(r rune) bool { return r == '\\' || r < ' ' }) { + return "", false + } + return "'" + strings.ReplaceAll(value, "'", "''") + "'", true +} + +// parsePostgresDefaultConstant reduces a DEFAULT rendered by pg_get_expr to the constant it holds, +// returning the value with pg's quoting undone and whether it was quoted. Postgres constant folds +// defaults and labels them with their type, so `5`, `(-1)`, `1.50`, `'-32768'::integer` and +// `'abc'::character varying` are all constants, while `now()`, `(id * 2)` and `'{1,2}'::integer[]` +// are not. +func parsePostgresDefaultConstant(rendered string) (string, bool, bool) { + expr := strings.TrimSpace(rendered) + if strings.HasPrefix(expr, "'") { + value, rest, ok := cutPostgresStringLiteral(expr) + return value, true, ok && isPostgresTypeLabel(rest) + } + + value, rest, _ := strings.Cut(expr, "::") + if !isPostgresTypeLabel(rest) { + return "", false, false + } + value = strings.TrimSpace(value) + // an unlabelled signed number is parenthesized, keeping the sign away from the scanner + if strings.HasPrefix(value, "(") && strings.HasSuffix(value, ")") { + value = strings.TrimSpace(value[1 : len(value)-1]) + } + return value, false, value != "" +} + +// cutPostgresStringLiteral consumes the leading single quoted literal of expr, undoing the doubling pg +// uses for embedded quotes, and returns whatever follows it. Our connections run with +// standard_conforming_strings=on, so backslashes carry no meaning here. +func cutPostgresStringLiteral(expr string) (string, string, bool) { + var value strings.Builder + for i := 1; i < len(expr); i++ { + if expr[i] != '\'' { + value.WriteByte(expr[i]) + } else if i+1 < len(expr) && expr[i+1] == '\'' { + value.WriteByte('\'') + i++ + } else { + return value.String(), expr[i+1:], true + } + } + return "", "", false +} + +// isPostgresTypeLabel reports whether rest is the `::type` label pg appends to a constant, and nothing +// more. Refusing anything else is what keeps composite expressions out: `'x'::text || 'y'::text` trips +// on the operator, `'{1,2}'::integer[]` on the brackets. +func isPostgresTypeLabel(rest string) bool { + if rest == "" { + return true + } + name, ok := strings.CutPrefix(rest, "::") + if !ok || name == "" { + return false + } + // names may be schema qualified and quoted, and `timestamp with time zone` brings spaces along + return !strings.ContainsFunc(name, func(r rune) bool { + return !unicode.IsLetter(r) && !unicode.IsDigit(r) && r != '_' && r != '.' && r != '"' && r != ' ' + }) +} + // getParentRelIDIfPartitioned checks if the relation ID is a child table // and returns the parent relation ID if it is. // If the relation ID is not a child table, it returns the original relation ID. diff --git a/flow/connectors/postgres/cdc_test.go b/flow/connectors/postgres/cdc_test.go index 96230dfc4..9a146550a 100644 --- a/flow/connectors/postgres/cdc_test.go +++ b/flow/connectors/postgres/cdc_test.go @@ -8,6 +8,7 @@ import ( "github.com/PeerDB-io/peerdb/flow/model" "github.com/PeerDB-io/peerdb/flow/otel_metrics" + "github.com/PeerDB-io/peerdb/flow/shared/types" ) func newTestCDCSource(t *testing.T) *PostgresCDCSource { @@ -36,3 +37,107 @@ func TestProcessMessageInvalidMessage(t *testing.T) { require.Error(t, err) require.Contains(t, err.Error(), "error parsing logical message (msgType=\"S\", walStart=0/1)") } + +// TestDefaultExprFromPostgresDefault feeds in what pg_get_expr renders for a DEFAULT; want is empty +// for the ones we decline to translate. +func TestDefaultExprFromPostgresDefault(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + rendered string + qkind types.QValueKind + want string + }{ + // integers, which pg quotes once a sign or enough width is involved + {name: "int", rendered: "5", qkind: types.QValueKindInt32, want: "5"}, + {name: "int negative", rendered: "'-1'::integer", qkind: types.QValueKindInt32, want: "-1"}, + {name: "int parenthesized", rendered: "(-1)", qkind: types.QValueKindInt32, want: "-1"}, + {name: "int max", rendered: "2147483647", qkind: types.QValueKindInt32, want: "2147483647"}, + // a smallint default is labelled with the type of the constant pg folded, not the column's + {name: "smallint negative", rendered: "'-32768'::integer", qkind: types.QValueKindInt16, want: "-32768"}, + {name: "bigint", rendered: "'9223372036854775807'::bigint", qkind: types.QValueKindInt64, want: "9223372036854775807"}, + {name: "oid", rendered: "42", qkind: types.QValueKindUInt32, want: "42"}, + + // scale is whatever pg rendered, so DECIMAL(10,2) DEFAULT 1.50 keeps its trailing zero + {name: "numeric", rendered: "1.50", qkind: types.QValueKindNumeric, want: "1.50"}, + {name: "double", rendered: "2.5", qkind: types.QValueKindFloat64, want: "2.5"}, + {name: "real negative", rendered: "'-1.5'::numeric", qkind: types.QValueKindFloat32, want: "-1.5"}, + {name: "double exponent", rendered: "1e-05", qkind: types.QValueKindFloat64, want: "1e-05"}, + + {name: "bool true", rendered: "true", qkind: types.QValueKindBoolean, want: "true"}, + {name: "bool false", rendered: "false", qkind: types.QValueKindBoolean, want: "false"}, + + // strings, requoted with SQL doubling so the literal is dialect neutral + {name: "text", rendered: "'hello'::text", qkind: types.QValueKindString, want: "'hello'"}, + {name: "text empty", rendered: "''::text", qkind: types.QValueKindString, want: "''"}, + {name: "text with quote", rendered: "'it''s'::text", qkind: types.QValueKindString, want: "'it''s'"}, + {name: "text with cast inside", rendered: "'a::b'::text", qkind: types.QValueKindString, want: "'a::b'"}, + {name: "varchar", rendered: "'abc'::character varying", qkind: types.QValueKindString, want: "'abc'"}, + {name: "bpchar", rendered: "'ab'::bpchar", qkind: types.QValueKindString, want: "'ab'"}, + {name: "qchar", rendered: `'x'::"char"`, qkind: types.QValueKindQChar, want: "'x'"}, + {name: "enum", rendered: "'ok'::mood", qkind: types.QValueKindEnum, want: "'ok'"}, + {name: "enum schema qualified", rendered: `'ok'::public."Mood"`, qkind: types.QValueKindEnum, want: "'ok'"}, + { + name: "uuid", rendered: "'00000000-0000-0000-0000-000000000001'::uuid", qkind: types.QValueKindUUID, + want: "'00000000-0000-0000-0000-000000000001'", + }, + {name: "jsonb", rendered: `'{"a": 1}'::jsonb`, qkind: types.QValueKindJSONB, want: `'{"a": 1}'`}, + {name: "inet", rendered: "'10.0.0.1'::inet", qkind: types.QValueKindINET, want: "'10.0.0.1'"}, + {name: "date", rendered: "'2020-01-02'::date", qkind: types.QValueKindDate, want: "'2020-01-02'"}, + { + name: "timestamp", rendered: "'2020-01-02 03:04:05.678'::timestamp without time zone", + qkind: types.QValueKindTimestamp, want: "'2020-01-02 03:04:05.678'", + }, + // pg renders the offset with timezone=UTC, which our connections set; the destination column + // has no zone of its own + { + name: "timestamptz", rendered: "'2020-01-02 01:04:05+00'::timestamp with time zone", + qkind: types.QValueKindTimestampTZ, want: "'2020-01-02 01:04:05'", + }, + + // declined: not a constant + {name: "now", rendered: "now()", qkind: types.QValueKindTimestamp}, + {name: "current timestamp", rendered: "CURRENT_TIMESTAMP", qkind: types.QValueKindTimestamp}, + {name: "current user", rendered: "CURRENT_USER", qkind: types.QValueKindString}, + {name: "sequence", rendered: "nextval('t_id_seq'::regclass)", qkind: types.QValueKindInt32}, + {name: "arithmetic", rendered: "(2 + 3)", qkind: types.QValueKindInt32}, + {name: "generated expression", rendered: "(id * 2)", qkind: types.QValueKindInt32}, + {name: "concatenation", rendered: "('a'::text || 'b'::text)", qkind: types.QValueKindString}, + {name: "concatenation unparenthesized", rendered: "'a'::text || 'b'::text", qkind: types.QValueKindString}, + {name: "function of constant", rendered: "upper('a'::text)", qkind: types.QValueKindString}, + {name: "chained cast", rendered: "'5'::text::integer", qkind: types.QValueKindInt32}, + {name: "null", rendered: "NULL::integer", qkind: types.QValueKindInt32}, + {name: "unterminated literal", rendered: "'oops", qkind: types.QValueKindString}, + + // declined: a constant whose text form does not carry over + {name: "array", rendered: "'{1,2}'::integer[]", qkind: types.QValueKindArrayInt32}, + {name: "bytea", rendered: `'\x0001'::bytea`, qkind: types.QValueKindBytes}, + {name: "interval", rendered: "'1 day'::interval", qkind: types.QValueKindInterval}, + {name: "time", rendered: "'13:14:15'::time without time zone", qkind: types.QValueKindTime}, + {name: "point", rendered: "'(1,2)'::point", qkind: types.QValueKindPoint}, + {name: "numeric nan", rendered: "'NaN'::numeric", qkind: types.QValueKindNumeric}, + {name: "double infinity", rendered: "'Infinity'::double precision", qkind: types.QValueKindFloat64}, + { + name: "timestamptz non utc offset", rendered: "'2020-01-02 01:04:05+02'::timestamp with time zone", + qkind: types.QValueKindTimestampTZ, + }, + // backslashes and control characters escape differently across dialects + {name: "text with backslash", rendered: `'a\b'::text`, qkind: types.QValueKindString}, + {name: "text with newline", rendered: "'a\nb'::text", qkind: types.QValueKindString}, + + // declined: constant of the wrong shape for the column + {name: "quoted bool", rendered: "'true'::boolean", qkind: types.QValueKindBoolean}, + {name: "money", rendered: "1.99", qkind: types.QValueKindString}, + {name: "number for text", rendered: "5", qkind: types.QValueKindString}, + {name: "text for number", rendered: "'abc'::text", qkind: types.QValueKindInt32}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + literal, ok := defaultExprFromPostgresDefault(tc.rendered, tc.qkind) + require.Equal(t, tc.want != "", ok) + require.Equal(t, tc.want, literal) + }) + } +} diff --git a/flow/connectors/postgres/postgres_schema_delta_test.go b/flow/connectors/postgres/postgres_schema_delta_test.go index adacfeef5..1cac4c97d 100644 --- a/flow/connectors/postgres/postgres_schema_delta_test.go +++ b/flow/connectors/postgres/postgres_schema_delta_test.go @@ -247,6 +247,107 @@ func (s PostgresSchemaDeltaTestSuite) testAddDropWhitespaceColumnNames(t *testin require.Equal(t, expectedTableSchema, output[tableName]) } +// TestAddedColumnCatalogInfo pins down what a live Postgres hands back for a column DEFAULT, and how +// much of it survives translation. want is empty for the defaults we decline to carry over. +func (s PostgresSchemaDeltaTestSuite) TestAddedColumnCatalogInfo() { + enumType := s.schema + ".mood" + _, err := s.connector.conn.Exec(s.t.Context(), "CREATE TYPE "+enumType+" AS ENUM ('sad','ok')") + require.NoError(s.t, err) + + // the case name doubles as the column name + for _, tc := range []struct { + name string + colDef string + qkind types.QValueKind + want string + notNull bool + }{ + // integers, which pg quotes once a sign or enough width is involved + {name: "c_int", colDef: "int DEFAULT 5", qkind: types.QValueKindInt32, want: "5"}, + {name: "c_int_neg", colDef: "int DEFAULT -1", qkind: types.QValueKindInt32, want: "-1"}, + {name: "c_int_zero", colDef: "int DEFAULT 0", qkind: types.QValueKindInt32, want: "0"}, + {name: "c_int_nn", colDef: "int NOT NULL DEFAULT 7", qkind: types.QValueKindInt32, want: "7", notNull: true}, + {name: "c_smallint", colDef: "smallint DEFAULT -32768", qkind: types.QValueKindInt16, want: "-32768"}, + { + name: "c_bigint", colDef: "bigint DEFAULT 9223372036854775807", qkind: types.QValueKindInt64, + want: "9223372036854775807", + }, + + // scale is whatever pg rendered, so numeric(10,2) keeps its trailing zero + {name: "c_numeric", colDef: "numeric(10,2) DEFAULT 1.50", qkind: types.QValueKindNumeric, want: "1.50"}, + {name: "c_float8", colDef: "double precision DEFAULT 2.5", qkind: types.QValueKindFloat64, want: "2.5"}, + {name: "c_float4", colDef: "real DEFAULT -1.5", qkind: types.QValueKindFloat32, want: "-1.5"}, + + {name: "c_bool", colDef: "boolean DEFAULT true", qkind: types.QValueKindBoolean, want: "true"}, + {name: "c_bool_false", colDef: "boolean DEFAULT false", qkind: types.QValueKindBoolean, want: "false"}, + + // strings, requoted with SQL doubling so the literal is dialect neutral + {name: "c_text", colDef: "text DEFAULT 'hello'", qkind: types.QValueKindString, want: "'hello'"}, + {name: "c_text_empty", colDef: "text DEFAULT ''", qkind: types.QValueKindString, want: "''"}, + {name: "c_text_quote", colDef: "text DEFAULT 'it''s'", qkind: types.QValueKindString, want: "'it''s'"}, + {name: "c_varchar", colDef: "varchar(10) DEFAULT 'abc'", qkind: types.QValueKindString, want: "'abc'"}, + {name: "c_char", colDef: "char(3) DEFAULT 'abc'", qkind: types.QValueKindString, want: "'abc'"}, + {name: "c_enum", colDef: enumType + " DEFAULT 'ok'", qkind: types.QValueKindEnum, want: "'ok'"}, + { + name: "c_uuid", colDef: "uuid DEFAULT '00000000-0000-0000-0000-000000000001'", qkind: types.QValueKindUUID, + want: "'00000000-0000-0000-0000-000000000001'", + }, + {name: "c_jsonb", colDef: `jsonb DEFAULT '{"a": 1}'`, qkind: types.QValueKindJSONB, want: `'{"a": 1}'`}, + {name: "c_inet", colDef: "inet DEFAULT '10.0.0.1'", qkind: types.QValueKindINET, want: "'10.0.0.1'"}, + {name: "c_date", colDef: "date DEFAULT '2020-01-02'", qkind: types.QValueKindDate, want: "'2020-01-02'"}, + { + name: "c_ts", colDef: "timestamp DEFAULT '2020-01-02 03:04:05.678'", qkind: types.QValueKindTimestamp, + want: "'2020-01-02 03:04:05.678'", + }, + // our connections run with timezone=UTC, so pg shifts the offset it renders to +00 + { + name: "c_tstz", colDef: "timestamptz DEFAULT '2020-01-02 03:04:05+02'", qkind: types.QValueKindTimestampTZ, + want: "'2020-01-02 01:04:05'", + }, + + // declined: not a constant + {name: "c_none", colDef: "int", qkind: types.QValueKindInt32}, + {name: "c_null", colDef: "int DEFAULT NULL", qkind: types.QValueKindInt32}, + {name: "c_now", colDef: "timestamptz DEFAULT now()", qkind: types.QValueKindTimestampTZ}, + {name: "c_expr", colDef: "int DEFAULT (2 + 3)", qkind: types.QValueKindInt32}, + {name: "c_concat", colDef: "text DEFAULT 'a' || 'b'", qkind: types.QValueKindString}, + {name: "c_serial", colDef: "serial", qkind: types.QValueKindInt32, notNull: true}, + {name: "c_identity", colDef: "int GENERATED BY DEFAULT AS IDENTITY", qkind: types.QValueKindInt32, notNull: true}, + // a generated column keeps its expression in pg_attrdef, where a default would sit + {name: "c_generated", colDef: "int GENERATED ALWAYS AS (base * 2) STORED", qkind: types.QValueKindInt32}, + + // declined: a constant whose text form does not carry over + {name: "c_array", colDef: "int[] DEFAULT '{1,2}'", qkind: types.QValueKindArrayInt32}, + {name: "c_bytea", colDef: `bytea DEFAULT '\x0001'`, qkind: types.QValueKindBytes}, + {name: "c_interval", colDef: "interval DEFAULT '1 day'", qkind: types.QValueKindInterval}, + {name: "c_time", colDef: "time DEFAULT '13:14:15'", qkind: types.QValueKindTime}, + {name: "c_nan", colDef: "numeric DEFAULT 'NaN'", qkind: types.QValueKindNumeric}, + // backslashes escape differently across dialects + {name: "c_backslash", colDef: `text DEFAULT E'a\\b'`, qkind: types.QValueKindString}, + } { + s.t.Run(tc.name, func(t *testing.T) { + tableName := fmt.Sprintf("%s.added_column_catalog_info_%s", s.schema, tc.name) + _, err := s.connector.conn.Exec(t.Context(), + fmt.Sprintf("CREATE TABLE %s(base int, %s %s)", tableName, tc.name, tc.colDef)) + require.NoError(t, err) + + var relID uint32 + require.NoError(t, s.connector.conn.QueryRow(t.Context(), "SELECT $1::regclass::oid", tableName).Scan(&relID)) + + catalogInfo, err := s.connector.fetchAddedColumnCatalogInfo(t.Context(), relID, []string{tc.name}) + require.NoError(t, err) + require.Len(t, catalogInfo, 1) + require.Equal(t, tc.notNull, catalogInfo[tc.name].notNull) + + var literal string + if renderedDefault := catalogInfo[tc.name].defaultExpr; renderedDefault != "" { + literal, _ = defaultExprFromPostgresDefault(renderedDefault, tc.qkind) + } + require.Equal(t, tc.want, literal) + }) + } +} + func TestPostgresSchemaDeltaTestSuite(t *testing.T) { e2eshared.RunSuite(t, SetupSuite) } diff --git a/flow/e2e/clickhouse_test.go b/flow/e2e/clickhouse_test.go index 9ea03fcd5..2be0bd69e 100644 --- a/flow/e2e/clickhouse_test.go +++ b/flow/e2e/clickhouse_test.go @@ -2248,6 +2248,157 @@ func (s ClickHouseSuite) Test_Nullable_Schema_Change_Replident_Index() { RequireEnvCanceled(s.t, env) } +type pgAddColumnDefaultCase struct{ name, def, val string } + +// Test_PG_AlterTableAddColumnDefault covers rows that existed before an ADD COLUMN with a DEFAULT. +func (s ClickHouseSuite) Test_PG_AlterTableAddColumnDefault() { + if _, ok := s.source.(*PostgresSource); !ok { + s.t.Skip("only applies to postgres") + } + + srcTableName := "test_add_col_default" + srcFullName := s.attachSchemaSuffix(srcTableName) + dstTableName := "test_add_col_default_dst" + enumType := s.attachSchemaSuffix("add_col_default_mood") + + require.NoError(s.t, s.source.Exec(s.t.Context(), "CREATE TYPE "+enumType+" AS ENUM ('sad','ok')")) + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT PRIMARY KEY)", srcFullName))) + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("INSERT INTO %s (id) VALUES (1)", srcFullName))) + + connectionGen := FlowConnectionGenerationConfig{ + FlowJobName: s.attachSuffix(srcTableName), + TableNameMapping: map[string]string{srcFullName: dstTableName}, + Destination: s.Peer().Name, + } + flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) + flowConnConfig.DoInitialSnapshot = true + + tc := NewTemporalClient(s.t) + env := ExecutePeerflow(s.t, tc, flowConnConfig) + SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) + + // Row 1 must land before the ALTER so it sits in a part predating the new columns; that is what + // makes ClickHouse fall back to the column default when reading it. + EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTableName, dstTableName, "id") + + cols := []pgAddColumnDefaultCase{ + {"c_int", "INT DEFAULT 5", "10"}, + {"c_int_neg", "INT DEFAULT -1", "-20"}, + {"c_int_zero", "INT DEFAULT 0", "1"}, + {"c_int_nn", "INT NOT NULL DEFAULT 7", "30"}, + {"c_smallint", "SMALLINT DEFAULT -32768", "3"}, + {"c_bigint", "BIGINT DEFAULT 9223372036854775807", "1"}, + // exact and approximate numeric; numeric keeps the scale as written + {"c_numeric", "NUMERIC(10,2) DEFAULT 1.50", "9.99"}, + {"c_double", "DOUBLE PRECISION DEFAULT 2.5", "3.5"}, + {"c_bool", "BOOLEAN DEFAULT true", "false"}, + // strings, including the quoting edge cases + {"c_text", "TEXT DEFAULT 'hello'", "'world'"}, + {"c_text_empty", "TEXT DEFAULT ''", "'nonempty'"}, + {"c_text_quote", "TEXT DEFAULT 'it''s'", "'x'"}, + {"c_varchar", "VARCHAR(20) DEFAULT 'dflt'", "'set'"}, + {"c_enum", enumType + " DEFAULT 'ok'", "'sad'"}, + {"c_uuid", "UUID DEFAULT '00000000-0000-0000-0000-000000000001'", "'11111111-1111-1111-1111-111111111111'"}, + {"c_jsonb", `JSONB DEFAULT '{"a": 1}'`, `'{"b": 2}'`}, + // date and time literals; pg renders a timestamptz in UTC, which is the zone our connections run in + {"c_date", "DATE DEFAULT '2020-01-02'", "'2021-02-03'"}, + {"c_ts", "TIMESTAMP DEFAULT '2020-01-02 03:04:05.678'", "'2021-02-03 04:05:06.789'"}, + {"c_tstz", "TIMESTAMPTZ DEFAULT '2020-01-02 03:04:05+02'", "'2021-02-03 04:05:06+03'"}, + } + + adds := make([]string, len(cols)) + names := make([]string, 0, len(cols)+1) + vals := make([]string, len(cols)) + names = append(names, "id") + for i, c := range cols { + adds[i] = "ADD COLUMN " + c.name + " " + c.def + names = append(names, c.name) + vals[i] = c.val + } + + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("ALTER TABLE %s %s", srcFullName, strings.Join(adds, ", ")))) + + // Row 2 arrives through CDC carrying its own values, so the ordinary path is exercised alongside + // row 1's read-time default. + require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf("INSERT INTO %s (%s) VALUES (2, %s)", + srcFullName, strings.Join(names, ", "), strings.Join(vals, ", ")))) + + EnvWaitForEqualTablesWithNames(env, s, "waiting on cdc add column with default", + srcTableName, dstTableName, strings.Join(names, ",")) + + env.Cancel(s.t.Context()) + RequireEnvCanceled(s.t, env) +} + +// Test_PG_AlterTableAddColumnDefaultUntranslated covers defaults deliberately not carried over, plus +// one ClickHouse rejects outright. Neither may stall replication for the table. +func (s ClickHouseSuite) Test_PG_AlterTableAddColumnDefaultUntranslated() { + if _, ok := s.source.(*PostgresSource); !ok { + s.t.Skip("only applies to postgres") + } + + srcTableName := "test_add_col_default_untranslated" + srcFullName := s.attachSchemaSuffix(srcTableName) + dstTableName := "test_add_col_default_untranslated_dst" + + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT PRIMARY KEY)", srcFullName))) + + connectionGen := FlowConnectionGenerationConfig{ + FlowJobName: s.attachSuffix(srcTableName), + TableNameMapping: map[string]string{srcFullName: dstTableName}, + Destination: s.Peer().Name, + } + flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) + flowConnConfig.DoInitialSnapshot = true + + tc := NewTemporalClient(s.t) + env := ExecutePeerflow(s.t, tc, flowConnConfig) + SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) + + cols := []pgAddColumnDefaultCase{ + // Evaluating these on the destination would yield a different value per query, so they are + // declined rather than translated. + {"c_now", "TIMESTAMPTZ DEFAULT now()", "'2021-02-03 04:05:06+00'"}, + {"c_serial", "SERIAL", "9"}, + // Not a constant, pg hands over the whole expression. + {"c_expr", "INT DEFAULT (2 + 3)", "9"}, + // Nullable columns already read back as NULL, so nothing needs to be emitted. + {"c_null", "INT DEFAULT NULL", "42"}, + // Array and binary literals are spelled differently on the destination. + {"c_array", "INT[] DEFAULT '{1,2}'", "'{3,4}'"}, + {"c_bytea", `BYTEA DEFAULT '\x0001'`, `'\x0203'`}, + // Backslashes escape differently across dialects, so the default is declined. + {"c_backslash", `TEXT DEFAULT E'a\\b'`, "'plain'"}, + // Translated, but ClickHouse cannot parse it; the column is then added without a default. + {"c_infinity", "TIMESTAMP DEFAULT 'infinity'", "'2021-02-03 04:05:06'"}, + } + + adds := make([]string, len(cols)) + names := make([]string, 0, len(cols)+1) + vals := make([]string, len(cols)) + names = append(names, "id") + for i, c := range cols { + adds[i] = "ADD COLUMN " + c.name + " " + c.def + names = append(names, c.name) + vals[i] = c.val + } + + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("ALTER TABLE %s %s", srcFullName, strings.Join(adds, ", ")))) + require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf("INSERT INTO %s (%s) VALUES (1, %s)", + srcFullName, strings.Join(names, ", "), strings.Join(vals, ", ")))) + + EnvWaitForEqualTablesWithNames(env, s, "waiting on cdc add column with untranslated default", + srcTableName, dstTableName, strings.Join(names, ",")) + + env.Cancel(s.t.Context()) + RequireEnvCanceled(s.t, env) +} + func (s ClickHouseSuite) Test_Unprivileged_Postgres_Columns() { var pgSource *PostgresSource var ok bool