implement backend for fetching type conversions

flow/cmd/peer_data.go

@@ -13,6 +13,7 @@ import (
 	"google.golang.org/protobuf/types/descriptorpb"
 
 	"github.com/PeerDB-io/peerdb/flow/connectors"
+	connclickhouse "github.com/PeerDB-io/peerdb/flow/connectors/clickhouse"
 	connpostgres "github.com/PeerDB-io/peerdb/flow/connectors/postgres"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
@@ -176,6 +177,13 @@ func (h *FlowRequestHandler) GetColumns(
 	return conn.GetColumns(ctx, req.SchemaName, req.TableName)
 }
 
+func (h *FlowRequestHandler) GetColumnsTypeConversion(
+	ctx context.Context,
+	req *protos.ColumnsTypeConversionRequest,
+) (*protos.ColumnsTypeConversionResponse, error) {
+	return connclickhouse.GetColumnsTypeConversion()
+}
+
 func (h *FlowRequestHandler) GetSlotInfo(
 	ctx context.Context,
 	req *protos.PostgresPeerActivityInfoRequest,

flow/connectors/clickhouse/qrep_avro_sync.go

@@ -364,47 +364,3 @@ func (s *ClickHouseAvroSyncMethod) writeToAvroFile(
 
 	return avroFile, nil
 }
-
-// add more supported type conversions as needed
-var supportedDestinationTypes = map[string][]qvalue.TypeConversion{
-	"String": {qvalue.NewTypeConversion(
-		qvalue.NumericToStringSchemaConversion,
-		qvalue.NumericToStringValueConversion,
-	)},
-}
-
-func findTypeConversions(schema qvalue.QRecordSchema, columns []*protos.ColumnSetting) map[string]qvalue.TypeConversion {
-	typeConversions := make(map[string]qvalue.TypeConversion)
-
-	colNameToType := make(map[string]qvalue.QValueKind, len(schema.Fields))
-	for _, field := range schema.Fields {
-		colNameToType[field.Name] = field.Type
-	}
-
-	for _, col := range columns {
-		colType, exist := colNameToType[col.SourceName]
-		if !exist {
-			continue
-		}
-		conversions, exist := supportedDestinationTypes[col.DestinationType]
-		if !exist {
-			continue
-		}
-		for _, conversion := range conversions {
-			if conversion.FromKind() == colType {
-				typeConversions[col.SourceName] = conversion
-			}
-		}
-	}
-
-	return typeConversions
-}
-
-func applyTypeConversions(schema qvalue.QRecordSchema, typeConversions map[string]qvalue.TypeConversion) qvalue.QRecordSchema {
-	for i, field := range schema.Fields {
-		if conversion, exist := typeConversions[field.Name]; exist {
-			schema.Fields[i] = conversion.SchemaConversion(field)
-		}
-	}
-	return schema
-}

flow/connectors/clickhouse/type_conversion.go

@@ -0,0 +1,94 @@
+package connclickhouse
+
+import (
+	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/model/qvalue"
+)
+
+/*
+This file handles the mapping for ClickHouse destination types and
+their corresponding TypeConversion implementations. A TypeConversion
+object contains two functions: one for schema conversion (QField) and
+one for value conversion (QValue). This allows the avro writer to
+stage the schema/data in the converted type format, and therefore
+successfully uploaded to the desired destination type in ClickHouse.
+
+To add a type conversion:
+	(1) In flow/model/qvalue/type_converter.go:
+	- implement a SchemaConversionFn interface to convert the QField type
+	- implement a ValueConversionFn interface to convert the QValue data
+
+	(2) Add the new conversion to the `supportedDestinationTypes` map here
+		(if destination type doesn't exist, create a new map entry for it).
+
+The GetColumnsTypeConversion function returns the full list of supported
+type conversions. Note that the source types are QValueKind, this allows
+the implementation to be source-connector agnostic.
+*/
+
+var supportedDestinationTypes = map[string][]qvalue.TypeConversion{
+	"String": {qvalue.NewTypeConversion(
+		qvalue.NumericToStringSchemaConversion,
+		qvalue.NumericToStringValueConversion,
+	)},
+}
+
+func GetColumnsTypeConversion() (*protos.ColumnsTypeConversionResponse, error) {
+	res := make([]*protos.ColumnsTypeConversion, 0)
+	for qkind, destTypes := range listSupportedTypeConversions() {
+		res = append(res, &protos.ColumnsTypeConversion{
+			Qkind:            string(qkind),
+			DestinationTypes: destTypes,
+		})
+	}
+	return &protos.ColumnsTypeConversionResponse{
+		Conversions: res,
+	}, nil
+}
+
+func listSupportedTypeConversions() map[qvalue.QValueKind][]string {
+	typeConversions := make(map[qvalue.QValueKind][]string)
+
+	for dstType, l := range supportedDestinationTypes {
+		for _, conversion := range l {
+			typeConversions[conversion.FromKind()] = append(typeConversions[conversion.FromKind()], dstType)
+		}
+	}
+	return typeConversions
+}
+
+func findTypeConversions(schema qvalue.QRecordSchema, columns []*protos.ColumnSetting) map[string]qvalue.TypeConversion {
+	typeConversions := make(map[string]qvalue.TypeConversion)
+
+	colNameToType := make(map[string]qvalue.QValueKind, len(schema.Fields))
+	for _, field := range schema.Fields {
+		colNameToType[field.Name] = field.Type
+	}
+
+	for _, col := range columns {
+		colType, exist := colNameToType[col.SourceName]
+		if !exist {
+			continue
+		}
+		conversions, exist := supportedDestinationTypes[col.DestinationType]
+		if !exist {
+			continue
+		}
+		for _, conversion := range conversions {
+			if conversion.FromKind() == colType {
+				typeConversions[col.SourceName] = conversion
+			}
+		}
+	}
+
+	return typeConversions
+}
+
+func applyTypeConversions(schema qvalue.QRecordSchema, typeConversions map[string]qvalue.TypeConversion) qvalue.QRecordSchema {
+	for i, field := range schema.Fields {
+		if conversion, exist := typeConversions[field.Name]; exist {
+			schema.Fields[i] = conversion.SchemaConversion(field)
+		}
+	}
+	return schema
+}

flow/connectors/mysql/schema.go

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

flow/connectors/postgres/schema.go

@@ -101,6 +101,7 @@ 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
 	FROM pg_attribute
@@ -120,15 +121,17 @@ 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 datatype pgtype.Text
 		var isPkey pgtype.Bool
-		if err := rows.Scan(&columnName, &datatype, &isPkey); err != nil {
+		if err := rows.Scan(&columnName, &oid, &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)),
 		}, nil
 	})
 	if err != nil {

protos/route.proto

@@ -185,11 +185,26 @@ message ColumnsItem {
   string name = 1;
   string type = 2;
   bool is_key = 3;
+  string qkind = 4;
 }
+
 message TableColumnsResponse {
   repeated ColumnsItem columns = 1;
 }
 
+message ColumnsTypeConversionRequest {
+  string destination_peer_type = 1;
+}
+
+message ColumnsTypeConversion {
+  string qkind = 1;
+  repeated string destination_types = 2;
+}
+
+message ColumnsTypeConversionResponse {
+  repeated ColumnsTypeConversion conversions = 1;
+}
+
 message PostgresPeerActivityInfoRequest { string peer_name = 1; }
 
 message PeerInfoRequest { string peer_name = 1; }
@@ -560,6 +575,12 @@ service FlowService {
     };
   }
 
+  rpc GetColumnsTypeConversion(ColumnsTypeConversionRequest) returns (ColumnsTypeConversionResponse) {
+    option (google.api.http) = {
+      get : "/v1/peers/columns/all_type_conversions",
+    };
+  }
+
   rpc GetSlotInfo(PostgresPeerActivityInfoRequest) returns (PeerSlotResponse) {
     option (google.api.http) = {
       get : "/v1/peers/slots/{peer_name}"