source-postgres: Treat enum primary keys as unpredictable

Previously we treated enum primary keys as having a predictable
ordering, which fails the moment somebody actually tries to
capture a table with an enum primary key where the cases are
not listed in alphabetical order because we just serialize the
string representation of the enum value as the backfill row key.

This commit adds a test and the discovery tweak that makes it pass.

source-postgres/.snapshots/TestEnumScanKey

@@ -0,0 +1,205 @@
+####################################
+### Capture from Start
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"red","data":"data","id":0}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":1,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FAJyZWQA"}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FAJyZWQA"
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"green","data":"data","id":0}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":2,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FAJncmVlbgA="}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FAJncmVlbgA="
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"blue","data":"data","id":0}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":3,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FAJibHVlAA=="}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FAJibHVlAA=="
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"red","data":"data","id":1}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":4,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQECcmVkAA=="}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQECcmVkAA=="
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"green","data":"data","id":1}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":5,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQECZ3JlZW4A"}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQECZ3JlZW4A"
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"blue","data":"data","id":1}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":6,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQECYmx1ZQA="}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQECYmx1ZQA="
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"red","data":"data","id":2}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":7,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQICcmVkAA=="}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQICcmVkAA=="
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"green","data":"data","id":2}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":8,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQICZ3JlZW4A"}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQICZ3JlZW4A"
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"blue","data":"data","id":2}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":9,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQICYmx1ZQA="}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQICYmx1ZQA="
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"red","data":"data","id":3}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":10,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQMCcmVkAA=="}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQMCcmVkAA=="
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"green","data":"data","id":4}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":11,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQQCZ3JlZW4A"}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQQCZ3JlZW4A"
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"blue","data":"data","id":5}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":12,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQUCYmx1ZQA="}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQUCYmx1ZQA="
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"green","data":"data","id":6}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":13,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQYCZ3JlZW4A"}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQYCZ3JlZW4A"
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"red","data":"data","id":7}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":14,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQcCcmVkAA=="}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQcCcmVkAA=="
+####################################
+# ================================
+# Collection "acmeCo/test/test_enumscankey_97825976": 1 Documents
+# ================================
+{"_meta":{"op":"c","source":{"schema":"test","snapshot":true,"table":"enumscankey_97825976","loc":[11111111,11111111,11111111]}},"color":"blue","data":"data","id":8}
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":15,"key_columns":["id","color"],"mode":"UnfilteredBackfill","scanned":"FQgCYmx1ZQA="}},"cursor":"0/1111111"}
+
+
+####################################
+### Capture from Key "FQgCYmx1ZQA="
+####################################
+# ================================
+# Final State Checkpoint
+# ================================
+{"bindingStateV1":{"test%2Fenumscankey_97825976":{"backfilled":15,"key_columns":["id","color"],"mode":"Active"}},"cursor":"0/1111111"}
+
+
+

source-postgres/datatype_test.go

@@ -214,3 +214,27 @@ func TestScanKeyTypes(t *testing.T) {
 		})
 	}
 }
+
+func TestEnumScanKey(t *testing.T) {
+	var tb, ctx = postgresTestBackend(t), context.Background()
+
+	tb.Query(ctx, t, `DROP TYPE IF EXISTS UserEnum CASCADE`)
+	tb.Query(ctx, t, `CREATE TYPE UserEnum AS ENUM ('red', 'green', 'blue')`)
+	t.Cleanup(func() { tb.Query(ctx, t, `DROP TYPE UserEnum CASCADE`) })
+
+	var uniqueID = "97825976"
+	var tableName = tb.CreateTable(ctx, t, uniqueID, "(id INTEGER, color UserEnum, data TEXT, PRIMARY KEY (id, color))")
+
+	tb.Insert(ctx, t, tableName, [][]any{
+		{1, "green", "data"}, {8, "blue", "data"}, {3, "red", "data"},
+		{0, "blue", "data"}, {1, "blue", "data"}, {0, "red", "data"},
+		{7, "red", "data"}, {6, "green", "data"}, {2, "blue", "data"},
+		{1, "red", "data"}, {2, "red", "data"}, {0, "green", "data"},
+		{4, "green", "data"}, {2, "green", "data"}, {5, "blue", "data"},
+	})
+
+	var cs = tb.CaptureSpec(ctx, t, regexp.MustCompile(uniqueID))
+	cs.EndpointSpec.(*Config).Advanced.BackfillChunkSize = 1
+	var summary, _ = tests.RestartingBackfillCapture(ctx, t, cs)
+	cupaloy.SnapshotT(t, summary)
+}

source-postgres/discovery.go

@@ -184,7 +184,10 @@ func predictableColumnOrder(colType any) bool {
 	// https://github.com/estuary/connectors/issues/1343 for more details.
 	if colType == "varchar" || colType == "bpchar" || colType == "text" {
 		return false
+	} else if _, ok := colType.(postgresEnumType); ok {
+		return false
 	}
+
 	return true
 }