[WIP] introduce direct source to destination type conversion

flow/connectors/mysql/qvalue_convert.go

@@ -15,7 +15,6 @@ import (
 	"github.com/shopspring/decimal"
 	geom "github.com/twpayne/go-geos"
 
-	"github.com/PeerDB-io/peerdb/flow/datatypes"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/model/qvalue"
 	"github.com/PeerDB-io/peerdb/flow/shared"
@@ -177,7 +176,7 @@ func QRecordSchemaFromMysqlFields(tableSchema *protos.TableSchema, fields []*mys
 		if col, ok := tableColumns[name]; ok {
 			qkind = qvalue.QValueKind(col.Type)
 			if qkind == qvalue.QValueKindNumeric {
-				precision, scale = datatypes.ParseNumericTypmod(col.TypeModifier)
+				precision, scale = shared.ParseNumericTypmod(col.TypeModifier)
 			}
 		} else {
 			var err error

flow/connectors/mysql/schema.go

@@ -100,10 +100,11 @@ func (c *MySqlConnector) GetColumns(ctx context.Context, schema string, table st
 			return nil, err
 		}
 		columns = append(columns, &protos.ColumnsItem{
-			Name:  columnName,
-			Type:  columnType,
-			IsKey: columnKey == "PRI",
-			Qkind: string(qkind),
+			Name:     columnName,
+			Type:     columnType,
+			IsKey:    columnKey == "PRI",
+			Qkind:    string(qkind),
+			TypeInfo: &protos.ColumnsItem_NullTypeInfo{},
 		})
 	}
 	return &protos.TableColumnsResponse{Columns: columns}, nil

flow/connectors/postgres/client.go

@@ -16,7 +16,6 @@ import (
 	"github.com/lib/pq/oid"
 
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
-	numeric "github.com/PeerDB-io/peerdb/flow/datatypes"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/shared"
@@ -471,7 +470,7 @@ func generateCreateTableSQLForNormalizedTable(
 			pgColumnType = qValueKindToPostgresType(pgColumnType)
 		}
 		if column.Type == "numeric" && column.TypeModifier != -1 {
-			precision, scale := numeric.ParseNumericTypmod(column.TypeModifier)
+			precision, scale := shared.ParseNumericTypmod(column.TypeModifier)
 			pgColumnType = fmt.Sprintf("numeric(%d,%d)", precision, scale)
 		}
 		var notNull string

flow/connectors/postgres/qrep_query_executor.go

@@ -10,7 +10,6 @@ import (
 	"github.com/jackc/pgx/v5/pgtype"
 	"go.temporal.io/sdk/log"
 
-	"github.com/PeerDB-io/peerdb/flow/datatypes"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/model/qvalue"
 	"github.com/PeerDB-io/peerdb/flow/shared"
@@ -76,7 +75,7 @@ func (qe *QRepQueryExecutor) fieldDescriptionsToSchema(fds []pgconn.FieldDescrip
 		ctype := qe.postgresOIDToQValueKind(fd.DataTypeOID, qe.customTypeMapping)
 		// there isn't a way to know if a column is nullable or not
 		if ctype == qvalue.QValueKindNumeric {
-			precision, scale := datatypes.ParseNumericTypmod(fd.TypeModifier)
+			precision, scale := shared.ParseNumericTypmod(fd.TypeModifier)
 			qfields[i] = qvalue.QField{
 				Name:      fd.Name,
 				Type:      ctype,

flow/connectors/postgres/qvalue_convert.go

@@ -14,14 +14,24 @@ import (
 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/lib/pq/oid"
 	"github.com/shopspring/decimal"
+	"go.temporal.io/sdk/log"
 
 	datatypes "github.com/PeerDB-io/peerdb/flow/datatypes"
 	"github.com/PeerDB-io/peerdb/flow/model/qvalue"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 )
 
 func (c *PostgresConnector) postgresOIDToName(recvOID uint32, customTypeMapping map[uint32]shared.CustomDataType) (string, error) {
-	if ty, ok := c.typeMap.TypeForOID(recvOID); ok {
+	return postgresOidToName(recvOID, customTypeMapping, c.typeMap)
+}
+
+// for unit testing
+func postgresOidToName(
+	recvOID uint32,
+	customTypeMapping map[uint32]shared.CustomDataType,
+	typeMap *pgtype.Map,
+) (string, error) {
+	if ty, ok := typeMap.TypeForOID(recvOID); ok {
 		return ty.Name, nil
 	}
 	// workaround for some types not being defined by pgtype
@@ -50,6 +60,17 @@ func (c *PostgresConnector) postgresOIDToName(recvOID uint32, customTypeMapping
 func (c *PostgresConnector) postgresOIDToQValueKind(
 	recvOID uint32,
 	customTypeMapping map[uint32]shared.CustomDataType,
+) qvalue.QValueKind {
+	return postgresOIDToQValueKind(recvOID, customTypeMapping, c.typeMap, c.hushWarnOID, c.logger)
+}
+
+// for unit testing
+func postgresOIDToQValueKind(
+	recvOID uint32,
+	customTypeMapping map[uint32]shared.CustomDataType,
+	typeMap *pgtype.Map,
+	hushWarnOID map[uint32]struct{},
+	logger log.Logger,
 ) qvalue.QValueKind {
 	switch recvOID {
 	case pgtype.BoolOID:
@@ -125,15 +146,15 @@ func (c *PostgresConnector) postgresOIDToQValueKind(
 	case pgtype.TstzrangeOID:
 		return qvalue.QValueKindTSTZRange
 	default:
-		if typeName, ok := c.typeMap.TypeForOID(recvOID); ok {
+		if typeName, ok := typeMap.TypeForOID(recvOID); ok {
 			colType := qvalue.QValueKindString
 			if typeData, ok := customTypeMapping[recvOID]; ok {
 				colType = customTypeToQKind(typeData)
 			}
-			if _, warned := c.hushWarnOID[recvOID]; !warned {
-				c.logger.Warn("unsupported field type",
+			if _, warned := hushWarnOID[recvOID]; !warned {
+				logger.Warn("unsupported field type",
 					slog.Int64("oid", int64(recvOID)), slog.String("typeName", typeName.Name), slog.String("mapping", string(colType)))
-				c.hushWarnOID[recvOID] = struct{}{}
+				hushWarnOID[recvOID] = struct{}{}
 			}
 			return colType
 		} else {

flow/connectors/postgres/qvalue_convert_test.go

@@ -0,0 +1,174 @@
+package connpostgres
+
+import (
+	"strconv"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/stretchr/testify/require"
+
+	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/internal"
+	"github.com/PeerDB-io/peerdb/flow/shared"
+	"github.com/PeerDB-io/peerdb/flow/shared/clickhouse"
+)
+
+// TODO: turn this test into an integration test and use TestContainer to query
+// an actual postgres database to fetch an exhaustive list of all pg types, to
+// avoid explicitly listing all types.
+var pgTypeInfo = []uint32{
+	pgtype.BoolOID,
+	pgtype.ByteaOID,
+	pgtype.QCharOID,
+	pgtype.NameOID,
+	pgtype.Int8OID,
+	pgtype.Int2OID,
+	pgtype.Int4OID,
+	pgtype.TextOID,
+	pgtype.OIDOID,
+	pgtype.TIDOID,
+	pgtype.XIDOID,
+	pgtype.CIDOID,
+	pgtype.JSONOID,
+	pgtype.XMLOID,
+	pgtype.XMLArrayOID,
+	pgtype.JSONArrayOID,
+	pgtype.XID8ArrayOID,
+	pgtype.PointOID,
+	pgtype.LsegOID,
+	pgtype.PathOID,
+	pgtype.BoxOID,
+	pgtype.PolygonOID,
+	pgtype.LineOID,
+	pgtype.LineArrayOID,
+	pgtype.CIDROID,
+	pgtype.CIDRArrayOID,
+	pgtype.Float4OID,
+	pgtype.Float8OID,
+	pgtype.CircleOID,
+	pgtype.CircleArrayOID,
+	pgtype.UnknownOID,
+	pgtype.Macaddr8OID,
+	pgtype.MacaddrOID,
+	pgtype.InetOID,
+	pgtype.BoolArrayOID,
+	pgtype.QCharArrayOID,
+	pgtype.NameArrayOID,
+	pgtype.Int2ArrayOID,
+	pgtype.Int4ArrayOID,
+	pgtype.TextArrayOID,
+	pgtype.TIDArrayOID,
+	pgtype.ByteaArrayOID,
+	pgtype.XIDArrayOID,
+	pgtype.CIDArrayOID,
+	pgtype.BPCharArrayOID,
+	pgtype.VarcharArrayOID,
+	pgtype.Int8ArrayOID,
+	pgtype.PointArrayOID,
+	pgtype.LsegArrayOID,
+	pgtype.PathArrayOID,
+	pgtype.BoxArrayOID,
+	pgtype.Float4ArrayOID,
+	pgtype.Float8ArrayOID,
+	pgtype.PolygonArrayOID,
+	pgtype.OIDArrayOID,
+	pgtype.ACLItemOID,
+	pgtype.ACLItemArrayOID,
+	pgtype.MacaddrArrayOID,
+	pgtype.InetArrayOID,
+	pgtype.BPCharOID,
+	pgtype.VarcharOID,
+	pgtype.DateOID,
+	pgtype.TimeOID,
+	pgtype.TimestampOID,
+	pgtype.TimestampArrayOID,
+	pgtype.DateArrayOID,
+	pgtype.TimeArrayOID,
+	pgtype.TimestamptzOID,
+	pgtype.TimestamptzArrayOID,
+	pgtype.IntervalOID,
+	pgtype.IntervalArrayOID,
+	pgtype.NumericArrayOID,
+	pgtype.TimetzOID,
+	pgtype.BitOID,
+	pgtype.BitArrayOID,
+	pgtype.VarbitOID,
+	pgtype.VarbitArrayOID,
+	pgtype.NumericOID,
+	pgtype.RecordOID,
+	pgtype.RecordArrayOID,
+	pgtype.UUIDOID,
+	pgtype.UUIDArrayOID,
+	pgtype.JSONBOID,
+	pgtype.JSONBArrayOID,
+	pgtype.DaterangeOID,
+	pgtype.DaterangeArrayOID,
+	pgtype.Int4rangeOID,
+	pgtype.Int4rangeArrayOID,
+	pgtype.NumrangeOID,
+	pgtype.NumrangeArrayOID,
+	pgtype.TsrangeOID,
+	pgtype.TsrangeArrayOID,
+	pgtype.TstzrangeOID,
+	pgtype.TstzrangeArrayOID,
+	pgtype.Int8rangeOID,
+	pgtype.Int8rangeArrayOID,
+	pgtype.JSONPathOID,
+	pgtype.JSONPathArrayOID,
+	pgtype.Int4multirangeOID,
+	pgtype.NummultirangeOID,
+	pgtype.TsmultirangeOID,
+	pgtype.TstzmultirangeOID,
+	pgtype.DatemultirangeOID,
+	pgtype.Int8multirangeOID,
+	pgtype.XID8OID,
+	// pgtype.TimetzArrayOID,          // not supported
+	// pgtype.Int4multirangeArrayOID,  // not supported
+	// pgtype.NummultirangeArrayOID,   // not supported
+	// pgtype.TsmultirangeArrayOID,    // not supported
+	// pgtype.TstzmultirangeArrayOID,  // not supported
+	// pgtype.DatemultirangeArrayOID,  // not supported
+	// pgtype.Int8multirangeArrayOID,  // not supported
+}
+
+// TODO: add test cases for:
+//  1. type modifiers != -1
+//  2. custom types
+//  3. nullable=true
+func TestPostgresOidToClickHouseType(t *testing.T) {
+	ctx := t.Context()
+	logger := internal.LoggerFromCtx(ctx)
+	newMap := pgtype.NewMap()
+	hushWarnOID := make(map[uint32]struct{})
+	customTypeMap := map[uint32]shared.CustomDataType{}
+	env := map[string]string{
+		"PEERDB_CLICKHOUSE_UNBOUNDED_NUMERIC_AS_STRING": "false",
+	}
+	nullable := false
+	nullableEnabled := false
+	for _, oid := range pgTypeInfo {
+		pgType, err := postgresOidToName(oid, customTypeMap, newMap)
+		require.NoError(t, err)
+
+		fd := &protos.FieldDescription{
+			Name:         strconv.FormatUint(uint64(oid), 10),
+			TypeModifier: -1,
+			Nullable:     nullable,
+			Type:         pgType,
+		}
+
+		kind := postgresOIDToQValueKind(oid, customTypeMap, newMap, hushWarnOID, logger)
+		derivedPgToChType, err := kind.ToDWHColumnType(
+			ctx,
+			env,
+			protos.DBType_CLICKHOUSE,
+			fd,
+			nullableEnabled,
+		)
+		require.NoError(t, err)
+		directPgToChType := clickhouse.PostgresToClickHouseType(oid, fd.TypeModifier, fd.Nullable)
+
+		require.Equal(t, derivedPgToChType, directPgToChType, "Mismatch type for oid=%d", oid)
+	}
+
+}

flow/connectors/postgres/schema.go

@@ -100,10 +100,11 @@ func (c *PostgresConnector) GetTablesInSchema(
 
 func (c *PostgresConnector) GetColumns(ctx context.Context, schema string, table string) (*protos.TableColumnsResponse, error) {
 	rows, err := c.conn.Query(ctx, `SELECT
-    DISTINCT attname AS column_name,
-    atttypid AS oid,
-    format_type(atttypid, atttypmod) AS data_type,
-    (pg_constraint.contype = 'p') AS is_primary_key
+	DISTINCT attname AS column_name,
+	atttypid AS oid,
+	atttypmod AS typmod,
+	format_type(atttypid, atttypmod) AS data_type,
+	(pg_constraint.contype = 'p') AS is_primary_key
 	FROM pg_attribute
 	JOIN pg_class ON pg_attribute.attrelid = pg_class.oid
 	JOIN pg_namespace ON pg_class.relnamespace = pg_namespace.oid
@@ -122,16 +123,23 @@ func (c *PostgresConnector) GetColumns(ctx context.Context, schema string, table
 	columns, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (*protos.ColumnsItem, error) {
 		var columnName pgtype.Text
 		var oid uint32
+		var typmod int32
 		var datatype pgtype.Text
 		var isPkey pgtype.Bool
-		if err := rows.Scan(&columnName, &oid, &datatype, &isPkey); err != nil {
+		if err := rows.Scan(&columnName, &oid, &typmod, &datatype, &isPkey); err != nil {
 			return nil, err
 		}
 		return &protos.ColumnsItem{
 			Name:  columnName.String,
 			Type:  datatype.String,
 			IsKey: isPkey.Bool,
 			Qkind: string(c.postgresOIDToQValueKind(oid, c.customTypeMapping)),
+			TypeInfo: &protos.ColumnsItem_PostgresTypeInfo{
+				PostgresTypeInfo: &protos.PostgresTypeInfo{
+					Oid:    oid,
+					Typmod: typmod,
+				},
+			},
 		}, nil
 	})
 	if err != nil {

flow/datatypes/numeric.go

@@ -1,13 +1,14 @@
 package datatypes
 
+import "github.com/PeerDB-io/peerdb/flow/shared"
+
 const (
 	// defaults
 	PeerDBBigQueryScale   = 20
 	PeerDBSnowflakeScale  = 20
 	PeerDBClickHouseScale = 38
 
 	PeerDBClickHouseMaxPrecision = 76
-	VARHDRSZ                     = 4
 )
 
 type WarehouseNumericCompatibility interface {
@@ -84,28 +85,15 @@ func MakeNumericTypmod(precision int32, scale int32) int32 {
 	if precision == 0 && scale == 0 {
 		return -1
 	}
-	return (precision << 16) | (scale & 0x7ff) + VARHDRSZ
-}
-
-// This is to reverse what make_numeric_typmod of Postgres does:
-// https://github.com/postgres/postgres/blob/21912e3c0262e2cfe64856e028799d6927862563/src/backend/utils/adt/numeric.c#L897
-func ParseNumericTypmod(typmod int32) (int16, int16) {
-	if typmod == -1 {
-		return 0, 0
-	}
-
-	offsetMod := typmod - VARHDRSZ
-	precision := int16((offsetMod >> 16) & 0x7FFF)
-	scale := int16(offsetMod & 0x7FFF)
-	return precision, scale
+	return (precision << 16) | (scale & 0x7ff) + shared.VARHDRSZ
 }
 
 func GetNumericTypeForWarehouse(typmod int32, warehouseNumeric WarehouseNumericCompatibility) (int16, int16) {
 	if typmod == -1 {
 		return warehouseNumeric.DefaultPrecisionAndScale()
 	}
 
-	precision, scale := ParseNumericTypmod(typmod)
+	precision, scale := shared.ParseNumericTypmod(typmod)
 	return GetNumericTypeForWarehousePrecisionScale(precision, scale, warehouseNumeric)
 }