Skip to content

Commit 146e61d

Browse files
jgao54jgao54
authored
move pg -> qvalue logic to shared for type conversion (#3012)
Follow-up to #3011. Move Postgres -> qvalue type mapping logic to shared package so we can determined qvalue. This gives us a way to use shared package to directly derive: 1) qvalue kind from postgres type using `PostgresOIDToQValueKind` 2) supported destination types from qvalue using `ListSupportedTypeConversions` Test: CI + locally tested a simple flow.
1 parent 50fdfa0 commit 146e61d

25 files changed

Lines changed: 168 additions & 142 deletions

‎flow/connectors/bigquery/qrep.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,10 +8,10 @@ import (
88

99
"cloud.google.com/go/bigquery"
1010

11-
"github.com/PeerDB-io/peerdb/flow/datatypes"
1211
"github.com/PeerDB-io/peerdb/flow/generated/protos"
1312
"github.com/PeerDB-io/peerdb/flow/model"
1413
"github.com/PeerDB-io/peerdb/flow/shared"
14+
"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
1515
"github.com/PeerDB-io/peerdb/flow/shared/types"
1616
)
1717

‎flow/connectors/bigquery/qvalue_convert.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,8 @@ import (
55

66
"cloud.google.com/go/bigquery"
77

8-
"github.com/PeerDB-io/peerdb/flow/datatypes"
98
"github.com/PeerDB-io/peerdb/flow/generated/protos"
9+
"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
1010
"github.com/PeerDB-io/peerdb/flow/shared/types"
1111
)
1212

‎flow/connectors/mysql/cdc.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,12 @@ import (
2222
"github.com/PeerDB-io/peerdb/flow/alerting"
2323
"github.com/PeerDB-io/peerdb/flow/connectors/utils"
2424
"github.com/PeerDB-io/peerdb/flow/connectors/utils/monitoring"
25-
"github.com/PeerDB-io/peerdb/flow/datatypes"
2625
"github.com/PeerDB-io/peerdb/flow/generated/protos"
2726
"github.com/PeerDB-io/peerdb/flow/internal"
2827
"github.com/PeerDB-io/peerdb/flow/model"
2928
"github.com/PeerDB-io/peerdb/flow/otel_metrics"
3029
"github.com/PeerDB-io/peerdb/flow/shared"
30+
"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
3131
"github.com/PeerDB-io/peerdb/flow/shared/types"
3232
)
3333

‎flow/connectors/mysql/qvalue_convert.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,9 @@ import (
1515
"github.com/shopspring/decimal"
1616
geom "github.com/twpayne/go-geos"
1717

18-
"github.com/PeerDB-io/peerdb/flow/datatypes"
1918
"github.com/PeerDB-io/peerdb/flow/generated/protos"
2019
"github.com/PeerDB-io/peerdb/flow/shared"
20+
"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
2121
"github.com/PeerDB-io/peerdb/flow/shared/types"
2222
)
2323

‎flow/connectors/postgres/cdc.go‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,14 @@ import (
2323
connmetadata "github.com/PeerDB-io/peerdb/flow/connectors/external_metadata"
2424
"github.com/PeerDB-io/peerdb/flow/connectors/utils"
2525
"github.com/PeerDB-io/peerdb/flow/connectors/utils/monitoring"
26-
geo "github.com/PeerDB-io/peerdb/flow/datatypes"
2726
"github.com/PeerDB-io/peerdb/flow/generated/protos"
2827
"github.com/PeerDB-io/peerdb/flow/internal"
2928
"github.com/PeerDB-io/peerdb/flow/model"
3029
"github.com/PeerDB-io/peerdb/flow/otel_metrics"
3130
"github.com/PeerDB-io/peerdb/flow/shared"
31+
geo "github.com/PeerDB-io/peerdb/flow/shared/datatypes"
3232
"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
33+
"github.com/PeerDB-io/peerdb/flow/shared/postgres"
3334
"github.com/PeerDB-io/peerdb/flow/shared/types"
3435
)
3536

@@ -282,7 +283,7 @@ func (p *PostgresCDCSource) decodeColumnData(
282283
} else if dataType == uint32(oid.T_timetz) { // ugly TIMETZ workaround for CDC decoding.
283284
return p.parseFieldFromPostgresOID(dataType, string(data), customTypeMapping)
284285
} else if typeData, ok := customTypeMapping[dataType]; ok {
285-
customQKind := customTypeToQKind(typeData)
286+
customQKind := postgres.CustomTypeToQKind(typeData)
286287
switch customQKind {
287288
case types.QValueKindGeography, types.QValueKindGeometry:
288289
wkt, err := geo.GeoValidate(string(data))
@@ -911,7 +912,7 @@ func processRelationMessage[Items model.Items](
911912
if qKind == types.QValueKindInvalid {
912913
typeName, ok := customTypeMapping[column.DataType]
913914
if ok {
914-
qKind = customTypeToQKind(typeName)
915+
qKind = postgres.CustomTypeToQKind(typeName)
915916
}
916917
}
917918
currRelMap[column.Name] = string(qKind)

‎flow/connectors/postgres/client.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,10 @@ import (
1616
"github.com/lib/pq/oid"
1717

1818
"github.com/PeerDB-io/peerdb/flow/connectors/utils"
19-
numeric "github.com/PeerDB-io/peerdb/flow/datatypes"
2019
"github.com/PeerDB-io/peerdb/flow/generated/protos"
2120
"github.com/PeerDB-io/peerdb/flow/model"
2221
"github.com/PeerDB-io/peerdb/flow/shared"
22+
numeric "github.com/PeerDB-io/peerdb/flow/shared/datatypes"
2323
)
2424

2525
const (

‎flow/connectors/postgres/qrep_query_executor.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,9 +10,9 @@ import (
1010
"github.com/jackc/pgx/v5/pgtype"
1111
"go.temporal.io/sdk/log"
1212

13-
"github.com/PeerDB-io/peerdb/flow/datatypes"
1413
"github.com/PeerDB-io/peerdb/flow/model"
1514
"github.com/PeerDB-io/peerdb/flow/shared"
15+
"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
1616
"github.com/PeerDB-io/peerdb/flow/shared/types"
1717
)
1818

‎flow/connectors/postgres/qvalue_convert.go‎

Lines changed: 12 additions & 125 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,9 @@ import (
1515
"github.com/lib/pq/oid"
1616
"github.com/shopspring/decimal"
1717

18-
datatypes "github.com/PeerDB-io/peerdb/flow/datatypes"
1918
"github.com/PeerDB-io/peerdb/flow/shared"
19+
"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
20+
"github.com/PeerDB-io/peerdb/flow/shared/postgres"
2021
"github.com/PeerDB-io/peerdb/flow/shared/types"
2122
)
2223

@@ -51,106 +52,17 @@ func (c *PostgresConnector) postgresOIDToQValueKind(
5152
recvOID uint32,
5253
customTypeMapping map[uint32]shared.CustomDataType,
5354
) types.QValueKind {
54-
switch recvOID {
55-
case pgtype.BoolOID:
56-
return types.QValueKindBoolean
57-
case pgtype.Int2OID:
58-
return types.QValueKindInt16
59-
case pgtype.Int4OID:
60-
return types.QValueKindInt32
61-
case pgtype.Int8OID:
62-
return types.QValueKindInt64
63-
case pgtype.Float4OID:
64-
return types.QValueKindFloat32
65-
case pgtype.Float8OID:
66-
return types.QValueKindFloat64
67-
case pgtype.QCharOID:
68-
return types.QValueKindQChar
69-
case pgtype.TextOID, pgtype.VarcharOID, pgtype.BPCharOID:
70-
return types.QValueKindString
71-
case pgtype.ByteaOID:
72-
return types.QValueKindBytes
73-
case pgtype.JSONOID:
74-
return types.QValueKindJSON
75-
case pgtype.JSONBOID:
76-
return types.QValueKindJSONB
77-
case pgtype.UUIDOID:
78-
return types.QValueKindUUID
79-
case pgtype.TimeOID:
80-
return types.QValueKindTime
81-
case pgtype.DateOID:
82-
return types.QValueKindDate
83-
case pgtype.CIDROID:
84-
return types.QValueKindCIDR
85-
case pgtype.MacaddrOID:
86-
return types.QValueKindMacaddr
87-
case pgtype.InetOID:
88-
return types.QValueKindINET
89-
case pgtype.TimestampOID:
90-
return types.QValueKindTimestamp
91-
case pgtype.TimestamptzOID:
92-
return types.QValueKindTimestampTZ
93-
case pgtype.NumericOID:
94-
return types.QValueKindNumeric
95-
case pgtype.Int2ArrayOID:
96-
return types.QValueKindArrayInt16
97-
case pgtype.Int4ArrayOID:
98-
return types.QValueKindArrayInt32
99-
case pgtype.Int8ArrayOID:
100-
return types.QValueKindArrayInt64
101-
case pgtype.PointOID:
102-
return types.QValueKindPoint
103-
case pgtype.Float4ArrayOID:
104-
return types.QValueKindArrayFloat32
105-
case pgtype.Float8ArrayOID:
106-
return types.QValueKindArrayFloat64
107-
case pgtype.BoolArrayOID:
108-
return types.QValueKindArrayBoolean
109-
case pgtype.DateArrayOID:
110-
return types.QValueKindArrayDate
111-
case pgtype.TimestampArrayOID:
112-
return types.QValueKindArrayTimestamp
113-
case pgtype.TimestamptzArrayOID:
114-
return types.QValueKindArrayTimestampTZ
115-
case pgtype.UUIDArrayOID:
116-
return types.QValueKindArrayUUID
117-
case pgtype.TextArrayOID, pgtype.VarcharArrayOID, pgtype.BPCharArrayOID:
118-
return types.QValueKindArrayString
119-
case pgtype.JSONArrayOID:
120-
return types.QValueKindArrayJSON
121-
case pgtype.JSONBArrayOID:
122-
return types.QValueKindArrayJSONB
123-
case pgtype.IntervalOID:
124-
return types.QValueKindInterval
125-
case pgtype.TstzrangeOID:
126-
return types.QValueKindTSTZRange
127-
default:
128-
if typeName, ok := c.typeMap.TypeForOID(recvOID); ok {
129-
colType := types.QValueKindString
130-
if typeData, ok := customTypeMapping[recvOID]; ok {
131-
colType = customTypeToQKind(typeData)
132-
}
133-
if _, warned := c.hushWarnOID[recvOID]; !warned {
134-
c.logger.Warn("unsupported field type",
135-
slog.Int64("oid", int64(recvOID)), slog.String("typeName", typeName.Name), slog.String("mapping", string(colType)))
136-
c.hushWarnOID[recvOID] = struct{}{}
137-
}
138-
return colType
139-
} else {
140-
// workaround for some types not being defined by pgtype
141-
switch oid.Oid(recvOID) {
142-
case oid.T_timetz:
143-
return types.QValueKindTimeTZ
144-
case oid.T_point:
145-
return types.QValueKindPoint
146-
default:
147-
if typeData, ok := customTypeMapping[recvOID]; ok {
148-
return customTypeToQKind(typeData)
149-
}
150-
return types.QValueKindString
151-
}
152-
}
55+
colType, err := postgres.PostgresOIDToQValueKind(recvOID, customTypeMapping, c.typeMap)
56+
_, warned := c.hushWarnOID[recvOID]
57+
if err != nil && !warned {
58+
c.logger.Warn(
59+
"unsupported field type",
60+
slog.Int64("oid", int64(recvOID)),
61+
slog.String("typeName", err.Error()),
62+
slog.String("mapping", string(colType)))
63+
c.hushWarnOID[recvOID] = struct{}{}
15364
}
65+
return colType
15466
}
15567

15668
func qValueKindToPostgresType(colTypeStr string) string {
@@ -616,31 +528,6 @@ func numericToDecimal(numVal pgtype.Numeric) (types.QValue, error) {
616528
}
617529
}
618530

619-
func customTypeToQKind(typeData shared.CustomDataType) types.QValueKind {
620-
if typeData.Type == 'e' {
621-
if typeData.Delim != 0 {
622-
return types.QValueKindArrayEnum
623-
} else {
624-
return types.QValueKindEnum
625-
}
626-
}
627-
628-
if typeData.Delim != 0 {
629-
return types.QValueKindArrayString
630-
}
631-
632-
switch typeData.Name {
633-
case "geometry":
634-
return types.QValueKindGeometry
635-
case "geography":
636-
return types.QValueKindGeography
637-
case "hstore":
638-
return types.QValueKindHStore
639-
default:
640-
return types.QValueKindString
641-
}
642-
}
643-
644531
// Postgres does not like timestamps of the form 2006-01-02 15:04:05 +0000 UTC
645532
// in tstzrange.
646533
// convertTimeRangeBound removes the +0000 UTC part

‎flow/connectors/snowflake/get_schema_for_tests.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,8 @@ import (
44
"context"
55
"slices"
66

7-
"github.com/PeerDB-io/peerdb/flow/datatypes"
87
"github.com/PeerDB-io/peerdb/flow/generated/protos"
8+
"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
99
"github.com/PeerDB-io/peerdb/flow/shared/types"
1010
)
1111

‎flow/connectors/snowflake/merge_stmt_generator.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,10 @@ import (
66
"strings"
77

88
"github.com/PeerDB-io/peerdb/flow/connectors/utils"
9-
numeric "github.com/PeerDB-io/peerdb/flow/datatypes"
109
"github.com/PeerDB-io/peerdb/flow/generated/protos"
1110
"github.com/PeerDB-io/peerdb/flow/model/qvalue"
1211
"github.com/PeerDB-io/peerdb/flow/shared"
12+
numeric "github.com/PeerDB-io/peerdb/flow/shared/datatypes"
1313
"github.com/PeerDB-io/peerdb/flow/shared/types"
1414
)
1515

0 commit comments

Comments
 (0)