|
| 1 | +package connclickhouse |
| 2 | + |
| 3 | +import ( |
| 4 | + "github.com/PeerDB-io/peerdb/flow/generated/protos" |
| 5 | + "github.com/PeerDB-io/peerdb/flow/model/qvalue" |
| 6 | +) |
| 7 | + |
| 8 | +/* |
| 9 | +This file handles the mapping for ClickHouse destination types and |
| 10 | +their corresponding TypeConversion implementations. A TypeConversion |
| 11 | +object contains two functions: one for schema conversion (QField) and |
| 12 | +one for value conversion (QValue). This allows the avro writer to |
| 13 | +stage the schema/data in the converted type format, and therefore |
| 14 | +successfully uploaded to the desired destination type in ClickHouse. |
| 15 | +
|
| 16 | +To add a type conversion: |
| 17 | + (1) In flow/model/qvalue/type_converter.go: |
| 18 | + - implement a SchemaConversionFn interface to convert the QField type |
| 19 | + - implement a ValueConversionFn interface to convert the QValue data |
| 20 | +
|
| 21 | + (2) Add the new conversion to the `supportedDestinationTypes` map here |
| 22 | + (if destination type doesn't exist, create a new map entry for it). |
| 23 | +
|
| 24 | +The GetColumnsTypeConversion function returns the full list of supported |
| 25 | +type conversions. Note that the source types are QValueKind, this allows |
| 26 | +the implementation to be source-connector agnostic. |
| 27 | +*/ |
| 28 | + |
| 29 | +var supportedDestinationTypes = map[string][]qvalue.TypeConversion{ |
| 30 | + "String": {qvalue.NewTypeConversion( |
| 31 | + qvalue.NumericToStringSchemaConversion, |
| 32 | + qvalue.NumericToStringValueConversion, |
| 33 | + )}, |
| 34 | +} |
| 35 | + |
| 36 | +func GetColumnsTypeConversion() (*protos.ColumnsTypeConversionResponse, error) { |
| 37 | + res := make([]*protos.ColumnsTypeConversion, 0) |
| 38 | + for qkind, destTypes := range listSupportedTypeConversions() { |
| 39 | + res = append(res, &protos.ColumnsTypeConversion{ |
| 40 | + Qkind: string(qkind), |
| 41 | + DestinationTypes: destTypes, |
| 42 | + }) |
| 43 | + } |
| 44 | + return &protos.ColumnsTypeConversionResponse{ |
| 45 | + Conversions: res, |
| 46 | + }, nil |
| 47 | +} |
| 48 | + |
| 49 | +func listSupportedTypeConversions() map[qvalue.QValueKind][]string { |
| 50 | + typeConversions := make(map[qvalue.QValueKind][]string) |
| 51 | + |
| 52 | + for dstType, l := range supportedDestinationTypes { |
| 53 | + for _, conversion := range l { |
| 54 | + typeConversions[conversion.FromKind()] = append(typeConversions[conversion.FromKind()], dstType) |
| 55 | + } |
| 56 | + } |
| 57 | + return typeConversions |
| 58 | +} |
| 59 | + |
| 60 | +func findTypeConversions(schema qvalue.QRecordSchema, columns []*protos.ColumnSetting) map[string]qvalue.TypeConversion { |
| 61 | + typeConversions := make(map[string]qvalue.TypeConversion) |
| 62 | + |
| 63 | + colNameToType := make(map[string]qvalue.QValueKind, len(schema.Fields)) |
| 64 | + for _, field := range schema.Fields { |
| 65 | + colNameToType[field.Name] = field.Type |
| 66 | + } |
| 67 | + |
| 68 | + for _, col := range columns { |
| 69 | + colType, exist := colNameToType[col.SourceName] |
| 70 | + if !exist { |
| 71 | + continue |
| 72 | + } |
| 73 | + conversions, exist := supportedDestinationTypes[col.DestinationType] |
| 74 | + if !exist { |
| 75 | + continue |
| 76 | + } |
| 77 | + for _, conversion := range conversions { |
| 78 | + if conversion.FromKind() == colType { |
| 79 | + typeConversions[col.SourceName] = conversion |
| 80 | + } |
| 81 | + } |
| 82 | + } |
| 83 | + |
| 84 | + return typeConversions |
| 85 | +} |
| 86 | + |
| 87 | +func applyTypeConversions(schema qvalue.QRecordSchema, typeConversions map[string]qvalue.TypeConversion) qvalue.QRecordSchema { |
| 88 | + for i, field := range schema.Fields { |
| 89 | + if conversion, exist := typeConversions[field.Name]; exist { |
| 90 | + schema.Fields[i] = conversion.SchemaConversion(field) |
| 91 | + } |
| 92 | + } |
| 93 | + return schema |
| 94 | +} |
0 commit comments