Skip to content

Commit 8314fd4

Browse files
nmemondnmemond
authored
Implement migration of SERIAL columns from postgres sources, mapping to IDENTITY columns in Spanner (#1233)
* Implement conversion of SERIAL AutoGen in postgres source to Spanner IDENTITY columns (The actual implementation of identifying serial columns will come in later commits). * Identify SERIAL columns from PG dump files * Identify SERIAL columns from Postgres database schema * Update web UI to also show auto gen columns and allow setting IDENTITY options (skip range, start counter with) * Include serial column in postgres integration tests * Update docs * Update integration tests to check that serial columns were migrated as expected * Update DDL syntax to align with PG 9.6 as opposed to latest PG
1 parent 0bf576d commit 8314fd4

18 files changed

Lines changed: 814 additions & 84 deletions

File tree

‎common/constants/constants.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,7 @@ const (
111111
UUID string = "UUID"
112112
SEQUENCE string = "Sequence"
113113
AUTO_INCREMENT string = "Auto Increment"
114+
SERIAL string = "Serial" // For PostgreSQL source DBs (the Postgres dialect uses IDENTITY for auto-gen)
114115
IDENTITY string = "Identity"
115116
// Default gcs path of the Dataflow template.
116117
DEFAULT_TEMPLATE_PATH string = "gs://dataflow-templates/latest/flex/Cloud_Datastream_to_Spanner"

‎docs/data-types/postgres.md‎

Lines changed: 56 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ The Spanner migration tool maps PostgreSQL types to Spanner types as follows:
2828
|--------------------|------------------------|---------------------------------------------------------------|
2929
| `BOOL` | `BOOL` | |
3030
| `BIGINT` | `INT64` | |
31-
| `BIGSERIAL` | `INT64` | dropped autoincrement functionality |
31+
| `BIGSERIAL` | `INT64` | |
3232
| `BYTEA` | `BYTES(MAX)` | |
3333
| `CHAR` | `STRING(1)` | CHAR defaults to length 1 |
3434
| `CHAR(N)` | `STRING(N)` | differences in treatment of fixed-length character types |
@@ -37,8 +37,9 @@ The Spanner migration tool maps PostgreSQL types to Spanner types as follows:
3737
| `INTEGER` | `INT64` | changes in storage size |
3838
| `NUMERIC` | `NUMERIC` | potential changes of precision |
3939
| `REAL` | `FLOAT32` | |
40-
| `SERIAL` | `INT64` | dropped autoincrement functionality , changes in storage size |
40+
| `SERIAL` | `INT64` | changes in storage size |
4141
| `SMALLINT` | `INT64` | changes in storage size |
42+
| `SMALLSERIAL` | `INT64` | changes in storage size |
4243
| `TEXT` | `STRING(MAX)` | |
4344
| `TIMESTAMP` | `TIMESTAMP` | differences in treatment of timezones |
4445
| `TIMESTAMPTZ` | `TIMESTAMP` | |
@@ -57,10 +58,59 @@ up to 29 digits before the decimal point and up to 9 after the decimal point.
5758
PostgreSQL's NUMERIC type can potentially support higher precision that this, so
5859
please verify that Spanner's NUMERIC support meets your application needs.
5960

60-
## BIGSERIAL and SERIAL
61-
62-
Spanner does not support autoincrementing types, so these both map to `INT64`
63-
and the autoincrementing functionality is dropped.
61+
## BIGSERIAL, SERIAL, and SMALLSERIAL
62+
63+
These map to to [Spanner IDENTITY columns](
64+
https://cloud.google.com/spanner/docs/primary-key-default-value#identity-columns)
65+
with type `INT64`.
66+
67+
Users need to set the SKIP RANGE and/or START COUNTER WITH values to avoid duplicate key errors.
68+
69+
The SKIP RANGE and START COUNTER WITH values can be set most via both the web UI (recommended) and the CLI.
70+
71+
The Column tab of the web UI exposes fields to set the SKIP RANGE and START COUNTER WITH values. For more details, see [here](../ui/schema-conv/spanner-draft.md).
72+
73+
To set the SKIP RANGE and/or START COUNTER WITH values via the CLI, the recommended steps are as follows:
74+
- Do a dry-run schema-only migration to generate a session JSON file:
75+
```sh
76+
spanner-migration-tool schema -dry-run ...
77+
```
78+
- Open the resulting session file and find the relevant column definition(s) in the `ColDefs` collection of the table
79+
it belongs to
80+
- Set the appropriate fields in that column's `AutoGen.AutoIncrementOptions` node. All three values are expected to be
81+
strings containing a numeric value. For example:
82+
```json
83+
{
84+
"SpSchema": {
85+
"table1": {
86+
"Name": "SomeTable",
87+
"ColDefs": {
88+
"column1": {
89+
"Name": "some_column",
90+
"AutoGen": {
91+
"Name": "Auto Increment",
92+
"GenerationType": "Auto Increment",
93+
"AutoIncrementOptions": {
94+
"SkipRangeMin": "1000",
95+
"SkipRangeMax": "10000",
96+
"StartCounterWith": "500"
97+
}
98+
},
99+
...
100+
},
101+
...
102+
},
103+
...
104+
},
105+
...
106+
},
107+
...
108+
}
109+
```
110+
- Save the session file and run your desired migration using the updated session file:
111+
```sh
112+
spanner-migration-tool schema -session=<path to session file> ...
113+
```
64114

65115
## TIMESTAMP
66116

‎sources/postgres/infoschema.go‎

Lines changed: 39 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"fmt"
2121
"math/bits"
2222
"reflect"
23+
"slices"
2324
"sort"
2425
"strconv"
2526
"strings"
@@ -290,6 +291,7 @@ func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAnd
290291
ON ((c.table_catalog, c.table_schema, c.table_name, 'TABLE', c.dtd_identifier)
291292
= (e.object_catalog, e.object_schema, e.object_name, e.object_type, e.collection_type_identifier))
292293
where table_schema = $1 and table_name = $2 ORDER BY c.ordinal_position;`
294+
serialCols := isi.getSerialColumns(conv, table)
293295
cols, err := isi.Db.Query(q, table.Schema, table.Name)
294296
if err != nil {
295297
return nil, nil, fmt.Errorf("couldn't get schema for table %s.%s: %s", table.Schema, table.Name, err)
@@ -318,21 +320,48 @@ func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAnd
318320
// Nothing to do here -- these are handled elsewhere.
319321
}
320322
}
321-
ignored.Default = colDefault.Valid
323+
isSerialColumn := slices.Contains(serialCols, colName)
324+
ignored.Default = colDefault.Valid && !isSerialColumn
322325
colId := internal.GenerateColumnId()
323326
c := schema.Column{
324327
Id: colId,
325328
Name: colName,
326329
Type: toType(dataType, elementDataType, charMaxLen, numericPrecision, numericScale),
327330
NotNull: common.ToNotNull(conv, isNullable),
328331
Ignored: ignored,
332+
AutoGen: toAutoGen(isSerialColumn),
329333
}
330334
colDefs[colId] = c
331335
colIds = append(colIds, colId)
332336
}
333337
return colDefs, colIds, nil
334338
}
335339

340+
func (isi InfoSchemaImpl) getSerialColumns(conv *internal.Conv, table common.SchemaAndName) []string {
341+
serialColsQuery := `SELECT a.attname FROM pg_attribute a
342+
WHERE attrelid = $1::regclass AND attnum > 0 AND a.atttypid = ANY ('{int,int8,int2}'::regtype[]) AND EXISTS (
343+
SELECT FROM pg_attrdef ad
344+
WHERE ad.adrelid = a.attrelid
345+
AND ad.adnum = a.attnum
346+
AND pg_get_expr(ad.adbin, ad.adrelid) = 'nextval('''
347+
|| (pg_get_serial_sequence(a.attrelid::regclass::text, a.attname))::regclass
348+
|| '''::regclass)'
349+
);`
350+
serialColsResult, err := isi.Db.Query(serialColsQuery, table.Schema + "." + table.Name)
351+
if err != nil {
352+
conv.Unexpected(fmt.Sprintf("Couldn't get information about serial columns for table %s.%s: %s", table.Schema, table.Name, err))
353+
return []string{}
354+
}
355+
defer serialColsResult.Close()
356+
serialCols := []string{}
357+
var serialColName string
358+
for serialColsResult.Next() {
359+
serialColsResult.Scan(&serialColName)
360+
serialCols = append(serialCols, serialColName)
361+
}
362+
return serialCols
363+
}
364+
336365
// GetConstraints returns a list of primary keys and by-column map of
337366
// other constraints. Note: we need to preserve ordinal order of
338367
// columns in primary key constraints.
@@ -525,6 +554,15 @@ func toType(dataType string, elementDataType sql.NullString, charLen sql.NullInt
525554
}
526555
}
527556

557+
func toAutoGen(isSerial bool) ddl.AutoGenCol {
558+
autoGen := ddl.AutoGenCol{}
559+
if isSerial {
560+
autoGen.Name = constants.SERIAL
561+
autoGen.GenerationType = constants.SERIAL
562+
}
563+
return autoGen;
564+
}
565+
528566
func cvtSQLArray(conv *internal.Conv, srcCd schema.Column, spCd ddl.ColumnDef, val interface{}) (interface{}, error) {
529567
a, ok := val.([]byte)
530568
if !ok {

‎sources/postgres/infoschema_test.go‎

Lines changed: 34 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,11 @@ func TestProcessSchema(t *testing.T) {
7777
{"public", "test", "ref", "id", "fk_test", constants.FK_RESTRICT, constants.FK_CASCADE},
7878
},
7979
},
80+
{
81+
query: "SELECT (.+) FROM pg_attribute (.+)",
82+
args: []driver.Value{"public.user"},
83+
cols: []string{"attname"},
84+
},
8085
{
8186
query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
8287
args: []driver.Value{"public", "user"},
@@ -109,6 +114,11 @@ func TestProcessSchema(t *testing.T) {
109114
{"public", "product", "productid", "product_id", "fk_test2", constants.FK_NO_ACTION, constants.FK_SET_NULL},
110115
{"public", "user", "userid", "user_id", "fk_test3", constants.FK_SET_NULL, constants.FK_RESTRICT}},
111116
},
117+
{
118+
query: "SELECT (.+) FROM pg_attribute (.+)",
119+
args: []driver.Value{"public.cart"},
120+
cols: []string{"attname"},
121+
},
112122
{
113123
query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
114124
args: []driver.Value{"public", "cart"},
@@ -142,6 +152,11 @@ func TestProcessSchema(t *testing.T) {
142152
args: []driver.Value{"public", "product"},
143153
cols: []string{"TABLE_SCHEMA", "REFERENCED_TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME", "ON_DELETE", "ON_UPDATE"},
144154
},
155+
{
156+
query: "SELECT (.+) FROM pg_attribute (.+)",
157+
args: []driver.Value{"public.product"},
158+
cols: []string{"attname"},
159+
},
145160
{
146161
query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
147162
args: []driver.Value{"public", "product"},
@@ -170,12 +185,18 @@ func TestProcessSchema(t *testing.T) {
170185
{"public", "test_ref", "txt", "ref_txt", "fk_test4", constants.FK_CASCADE, constants.FK_NO_ACTION}},
171186
},
172187

188+
{
189+
query: "SELECT (.+) FROM pg_attribute (.+)",
190+
args: []driver.Value{"public.test"},
191+
cols: []string{"attname"},
192+
rows: [][]driver.Value{{"id"}},
193+
},
173194
{
174195
query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
175196
args: []driver.Value{"public", "test"},
176197
cols: []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
177198
rows: [][]driver.Value{
178-
{"id", "bigint", nil, "NO", nil, nil, 64, 0},
199+
{"id", "bigint", nil, "NO", "nextval('public.test_id_seq'::regclass)", nil, 64, 0},
179200
{"aint", "ARRAY", "integer", "YES", nil, nil, nil, nil},
180201
{"atext", "ARRAY", "text", "YES", nil, nil, nil, nil},
181202
{"b", "boolean", nil, "YES", nil, nil, nil, nil},
@@ -216,6 +237,11 @@ func TestProcessSchema(t *testing.T) {
216237
args: []driver.Value{"public", "test_ref"},
217238
cols: []string{"TABLE_SCHEMA", "REFERENCED_TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME", "ON_DELETE", "ON_UPDATE"},
218239
},
240+
{
241+
query: "SELECT (.+) FROM pg_attribute (.+)",
242+
args: []driver.Value{"public.test_ref"},
243+
cols: []string{"attname"},
244+
},
219245
{
220246
query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
221247
args: []driver.Value{"public", "test_ref"},
@@ -285,7 +311,7 @@ func TestProcessSchema(t *testing.T) {
285311
Name: "test",
286312
ColIds: []string{"id", "aint", "atext", "b", "bs", "by", "c", "c_8", "d", "f8", "f4", "i8", "i4", "i2", "num", "s", "ts", "tz", "txt", "vc", "vc6"},
287313
ColDefs: map[string]ddl.ColumnDef{
288-
"id": ddl.ColumnDef{Name: "id", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
314+
"id": ddl.ColumnDef{Name: "id", T: ddl.Type{Name: ddl.Int64}, NotNull: true, AutoGen: ddl.AutoGenCol{Name: constants.IDENTITY, GenerationType: constants.IDENTITY}},
289315
"aint": ddl.ColumnDef{Name: "aint", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength, IsArray: false}},
290316
"atext": ddl.ColumnDef{Name: "atext", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength, IsArray: false}},
291317
"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.Bool}},
@@ -324,6 +350,7 @@ func TestProcessSchema(t *testing.T) {
324350
assert.Equal(t, nil, err)
325351
assert.Equal(t, len(conv.SchemaIssues[cartTableId].ColumnLevelIssues), 0)
326352
expectedIssues := map[string][]internal.SchemaIssue{
353+
"id": []internal.SchemaIssue{internal.IdentitySkipRange},
327354
"aint": []internal.SchemaIssue{internal.Widened, internal.ArrayTypeNotSupported},
328355
"bs": []internal.SchemaIssue{internal.DefaultValue},
329356
"i4": []internal.SchemaIssue{internal.Widened},
@@ -505,6 +532,11 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
505532
args: []driver.Value{"public", "test"},
506533
cols: []string{"TABLE_SCHEMA", "REFERENCED_TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME", "ON_DELETE", "ON_UPDATE"},
507534
},
535+
{
536+
query: "SELECT (.+) FROM pg_attribute (.+)",
537+
args: []driver.Value{"public.test"},
538+
cols: []string{"attname"},
539+
},
508540
{
509541
query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
510542
args: []driver.Value{"public", "test"},

0 commit comments

Comments
 (0)