Skip to content

Commit 070420f

Browse files
ilidemiilidemi
authored
Fix e2e table schema asserts (#4506)
E2E's `CompareTableSchemas` was comparing everything with an `||`, which is not a lot of comparing. Switch all to more strict `RequireEqualTableSchemas`. After trivial cleanups, one test that continued failing was `Test_Schema_Changes_Cutoff_Bug, validating schema changes handling: - in PG we test that the changes are only picked up after we get a row with that column and a schema delta (and a lazy RelationMessage right before) after #2717 - in MySQL we issue a schema delta the moment we see the DDL so the row is optional This is a product decision but funny that a test revealed it. Eagle eye is codex and main investigator is claude with a bunch of nudges
1 parent 8306ce3 commit 070420f

3 files changed

Lines changed: 142 additions & 360 deletions

File tree

‎flow/e2e/clickhouse_mysql_test.go‎

Lines changed: 33 additions & 146 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@ import (
2424
"github.com/PeerDB-io/peerdb/flow/pkg/common"
2525
mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql"
2626
"github.com/PeerDB-io/peerdb/flow/shared"
27-
"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
2827
"github.com/PeerDB-io/peerdb/flow/shared/types"
2928
)
3029

@@ -1188,70 +1187,29 @@ func (s ClickHouseSuite) Test_MySQL_Schema_Changes() {
11881187

11891188
EnvWaitForEqualTablesWithNames(env, s, "normalize reinsert", srcTable, dstTable, "id,c1")
11901189

1191-
expectedTableSchema := &protos.TableSchema{
1192-
TableIdentifier: ExpectedDestinationTableName(s, dstTable),
1193-
Columns: []*protos.FieldDescription{
1194-
{
1195-
Name: ExpectedDestinationIdentifier(s, "id"),
1196-
Type: string(types.QValueKindNumeric),
1197-
TypeModifier: -1,
1198-
},
1199-
{
1200-
Name: ExpectedDestinationIdentifier(s, "c1"),
1201-
Type: string(types.QValueKindNumeric),
1202-
TypeModifier: -1,
1203-
},
1204-
{
1205-
Name: "_PEERDB_IS_DELETED",
1206-
Type: string(types.QValueKindBoolean),
1207-
TypeModifier: -1,
1208-
},
1209-
{
1210-
Name: "_PEERDB_SYNCED_AT",
1211-
Type: string(types.QValueKindTimestamp),
1212-
TypeModifier: -1,
1213-
},
1214-
},
1215-
}
1190+
expectedTableSchema := ExpectedDestinationSchema(s, dstTable, []*protos.FieldDescription{
1191+
{Name: ExpectedDestinationIdentifier(s, "id"), Type: string(types.QValueKindUInt64), TypeModifier: -1},
1192+
{Name: ExpectedDestinationIdentifier(s, "c1"), Type: string(types.QValueKindInt64), TypeModifier: -1},
1193+
})
12161194
output, err := destinationSchemaConnector.GetTableSchema(t.Context(), nil, shared.InternalVersion_Latest, protos.TypeSystem_Q,
12171195
[]*protos.TableMapping{{SourceTableIdentifier: dstTableName}})
12181196
EnvNoError(t, env, err)
1219-
EnvTrue(t, env, CompareTableSchemas(expectedTableSchema, output[dstTableName]))
1197+
EnvTrue(t, env, RequireEqualTableSchemas(t, expectedTableSchema, output[dstTableName]))
12201198

12211199
// alter source table, add column addedColumn and insert another row.
12221200
EnvNoError(t, env, s.Source().Exec(t.Context(), fmt.Sprintf("ALTER TABLE %s ADD COLUMN `addedColumn` BIGINT", srcTableName)))
12231201
// so that the batch finishes, insert a row into the second source table.
12241202
EnvNoError(t, env, s.Source().Exec(t.Context(), fmt.Sprintf(`INSERT INTO %s VALUES(DEFAULT)`, secondSrcTableName)))
12251203
EnvWaitForEqualTablesWithNames(env, s, "normalize altered row", srcTable, dstTable, "id,c1,coalesce(`addedColumn`,0) `addedColumn`")
1226-
expectedTableSchema = &protos.TableSchema{
1227-
TableIdentifier: ExpectedDestinationTableName(s, dstTable),
1228-
Columns: []*protos.FieldDescription{
1229-
{
1230-
Name: ExpectedDestinationIdentifier(s, "id"),
1231-
Type: string(types.QValueKindNumeric),
1232-
TypeModifier: -1,
1233-
},
1234-
{
1235-
Name: ExpectedDestinationIdentifier(s, "c1"),
1236-
Type: string(types.QValueKindNumeric),
1237-
TypeModifier: -1,
1238-
},
1239-
{
1240-
Name: "_PEERDB_SYNCED_AT",
1241-
Type: string(types.QValueKindTimestamp),
1242-
TypeModifier: -1,
1243-
},
1244-
{
1245-
Name: ExpectedDestinationIdentifier(s, "addedColumn"),
1246-
Type: string(types.QValueKindNumeric),
1247-
TypeModifier: -1,
1248-
},
1249-
},
1250-
}
1204+
expectedTableSchema = ExpectedDestinationSchema(s, dstTable, []*protos.FieldDescription{
1205+
{Name: ExpectedDestinationIdentifier(s, "id"), Type: string(types.QValueKindUInt64), TypeModifier: -1},
1206+
{Name: ExpectedDestinationIdentifier(s, "c1"), Type: string(types.QValueKindInt64), TypeModifier: -1},
1207+
{Name: ExpectedDestinationIdentifier(s, "addedColumn"), Type: string(types.QValueKindInt64), TypeModifier: -1},
1208+
})
12511209
output, err = destinationSchemaConnector.GetTableSchema(t.Context(), nil, shared.InternalVersion_Latest, protos.TypeSystem_Q,
12521210
[]*protos.TableMapping{{SourceTableIdentifier: dstTableName}})
12531211
EnvNoError(t, env, err)
1254-
EnvTrue(t, env, CompareTableSchemas(expectedTableSchema, output[dstTableName]))
1212+
EnvTrue(t, env, RequireEqualTableSchemas(t, expectedTableSchema, output[dstTableName]))
12551213

12561214
env.Cancel(t.Context())
12571215
RequireEnvCanceled(t, env)
@@ -1313,35 +1271,14 @@ func (s ClickHouseSuite) Test_MySQL_GhOst_Schema_Changes() {
13131271
EnvWaitForEqualTablesWithNames(env, s, "initial row", srcTable, dstTable, "id,c1")
13141272

13151273
// Verify initial schema
1316-
expectedTableSchema := &protos.TableSchema{
1317-
TableIdentifier: ExpectedDestinationTableName(s, dstTable),
1318-
Columns: []*protos.FieldDescription{
1319-
{
1320-
Name: ExpectedDestinationIdentifier(s, "id"),
1321-
Type: string(types.QValueKindNumeric),
1322-
TypeModifier: -1,
1323-
},
1324-
{
1325-
Name: ExpectedDestinationIdentifier(s, "c1"),
1326-
Type: string(types.QValueKindNumeric),
1327-
TypeModifier: -1,
1328-
},
1329-
{
1330-
Name: "_PEERDB_IS_DELETED",
1331-
Type: string(types.QValueKindBoolean),
1332-
TypeModifier: -1,
1333-
},
1334-
{
1335-
Name: "_PEERDB_SYNCED_AT",
1336-
Type: string(types.QValueKindTimestamp),
1337-
TypeModifier: -1,
1338-
},
1339-
},
1340-
}
1274+
expectedTableSchema := ExpectedDestinationSchema(s, dstTable, []*protos.FieldDescription{
1275+
{Name: ExpectedDestinationIdentifier(s, "id"), Type: string(types.QValueKindUInt64), TypeModifier: -1},
1276+
{Name: ExpectedDestinationIdentifier(s, "c1"), Type: string(types.QValueKindInt64), TypeModifier: -1},
1277+
})
13411278
output, err := destinationSchemaConnector.GetTableSchema(t.Context(), nil, shared.InternalVersion_Latest, protos.TypeSystem_Q,
13421279
[]*protos.TableMapping{{SourceTableIdentifier: dstTableName}})
13431280
EnvNoError(t, env, err)
1344-
EnvTrue(t, env, CompareTableSchemas(expectedTableSchema, output[dstTableName]))
1281+
EnvTrue(t, env, RequireEqualTableSchemas(t, expectedTableSchema, output[dstTableName]))
13451282

13461283
// ============================================
13471284
// Simulate gh-ost migration: add columns to test type detection
@@ -1429,76 +1366,26 @@ func (s ClickHouseSuite) Test_MySQL_GhOst_Schema_Changes() {
14291366
require.Equal(t, fixedBinaryWant, qvalueBytes(dstRows.Records[2][1]))
14301367
require.Equal(t, varBinaryWant, qvalueBytes(dstRows.Records[2][2]))
14311368

1432-
// Verify schema was updated to include new columns with correct types and typmods
1433-
expectedTableSchema = &protos.TableSchema{
1434-
TableIdentifier: ExpectedDestinationTableName(s, dstTable),
1435-
Columns: []*protos.FieldDescription{
1436-
{
1437-
Name: ExpectedDestinationIdentifier(s, "id"),
1438-
Type: string(types.QValueKindNumeric),
1439-
TypeModifier: -1,
1440-
},
1441-
{
1442-
Name: ExpectedDestinationIdentifier(s, "c1"),
1443-
Type: string(types.QValueKindNumeric),
1444-
TypeModifier: -1,
1445-
},
1446-
{
1447-
Name: "_PEERDB_SYNCED_AT",
1448-
Type: string(types.QValueKindTimestamp),
1449-
TypeModifier: -1,
1450-
},
1451-
{
1452-
Name: ExpectedDestinationIdentifier(s, "c2"),
1453-
Type: string(types.QValueKindInt64), // BIGINT
1454-
TypeModifier: -1,
1455-
},
1456-
{
1457-
Name: ExpectedDestinationIdentifier(s, "c3"),
1458-
Type: string(types.QValueKindUInt32), // INT UNSIGNED
1459-
TypeModifier: -1,
1460-
},
1461-
{
1462-
Name: ExpectedDestinationIdentifier(s, "c4"),
1463-
Type: string(types.QValueKindBytes), // BLOB (binary charset)
1464-
TypeModifier: -1,
1465-
},
1466-
{
1467-
Name: ExpectedDestinationIdentifier(s, "c5"),
1468-
Type: string(types.QValueKindString), // TEXT (non-binary charset)
1469-
TypeModifier: -1,
1470-
},
1471-
{
1472-
Name: ExpectedDestinationIdentifier(s, "c6"),
1473-
Type: string(types.QValueKindNumeric), // DECIMAL (default 10,0)
1474-
TypeModifier: datatypes.MakeNumericTypmod(10, 0),
1475-
},
1476-
{
1477-
Name: ExpectedDestinationIdentifier(s, "c7"),
1478-
Type: string(types.QValueKindNumeric), // DECIMAL(10,2)
1479-
TypeModifier: datatypes.MakeNumericTypmod(10, 2),
1480-
},
1481-
{
1482-
Name: ExpectedDestinationIdentifier(s, "c8"),
1483-
Type: string(types.QValueKindNumeric), // DECIMAL(18,6)
1484-
TypeModifier: datatypes.MakeNumericTypmod(18, 6),
1485-
},
1486-
{
1487-
Name: ExpectedDestinationIdentifier(s, "c9"),
1488-
Type: string(types.QValueKindBytes), // BINARY(16)
1489-
TypeModifier: -1,
1490-
},
1491-
{
1492-
Name: ExpectedDestinationIdentifier(s, "c10"),
1493-
Type: string(types.QValueKindBytes), // VARBINARY(16)
1494-
TypeModifier: -1,
1495-
},
1496-
},
1497-
}
1369+
// Verify schema was updated to include the new columns.
1370+
// BLOB/BINARY/VARBINARY are stored in ClickHouse as String,
1371+
// our CH GetTableSchemaForTable always returns TypeModifier -1 for decimals
1372+
expectedTableSchema = ExpectedDestinationSchema(s, dstTable, []*protos.FieldDescription{
1373+
{Name: ExpectedDestinationIdentifier(s, "id"), Type: string(types.QValueKindUInt64), TypeModifier: -1},
1374+
{Name: ExpectedDestinationIdentifier(s, "c1"), Type: string(types.QValueKindInt64), TypeModifier: -1},
1375+
{Name: ExpectedDestinationIdentifier(s, "c2"), Type: string(types.QValueKindInt64), TypeModifier: -1}, // BIGINT
1376+
{Name: ExpectedDestinationIdentifier(s, "c3"), Type: string(types.QValueKindUInt32), TypeModifier: -1}, // INT UNSIGNED
1377+
{Name: ExpectedDestinationIdentifier(s, "c4"), Type: string(types.QValueKindString), TypeModifier: -1}, // BLOB (binary charset)
1378+
{Name: ExpectedDestinationIdentifier(s, "c5"), Type: string(types.QValueKindString), TypeModifier: -1}, // TEXT (non-binary charset)
1379+
{Name: ExpectedDestinationIdentifier(s, "c6"), Type: string(types.QValueKindNumeric), TypeModifier: -1}, // DECIMAL (default 10,0)
1380+
{Name: ExpectedDestinationIdentifier(s, "c7"), Type: string(types.QValueKindNumeric), TypeModifier: -1}, // DECIMAL(10,2)
1381+
{Name: ExpectedDestinationIdentifier(s, "c8"), Type: string(types.QValueKindNumeric), TypeModifier: -1}, // DECIMAL(18,6)
1382+
{Name: ExpectedDestinationIdentifier(s, "c9"), Type: string(types.QValueKindString), TypeModifier: -1}, // BINARY(16)
1383+
{Name: ExpectedDestinationIdentifier(s, "c10"), Type: string(types.QValueKindString), TypeModifier: -1}, // VARBINARY(16)
1384+
})
14981385
output, err = destinationSchemaConnector.GetTableSchema(t.Context(), nil, shared.InternalVersion_Latest, protos.TypeSystem_Q,
14991386
[]*protos.TableMapping{{SourceTableIdentifier: dstTableName}})
15001387
EnvNoError(t, env, err)
1501-
EnvTrue(t, env, CompareTableSchemas(expectedTableSchema, output[dstTableName]))
1388+
EnvTrue(t, env, RequireEqualTableSchemas(t, expectedTableSchema, output[dstTableName]))
15021389

15031390
env.Cancel(t.Context())
15041391
RequireEnvCanceled(t, env)

0 commit comments

Comments
 (0)