Move shared validations into their own module (#3547)

This allows to import just the validation code with minimal dependencies
and avoid tracking unrelated dependency updates in PeerDB.

Connector-specific code remaining in shared/ was kind of small so moved
it to other places. CH type conversion was supposed to be shared with
other services in the future but it pulls in QValue so we can decide to
move all of it to shared-light if we want to expose QValue or to move
just type conversions and give it its own small data contracts.

Author: ilidemi

flow/connectors/clickhouse/avro_sync.go

@@ -16,8 +16,8 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/model/qvalue"
+	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/pkg/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared"
-	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/shared/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )

flow/connectors/clickhouse/cdc.go

@@ -14,8 +14,8 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/model/qvalue"
+	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/pkg/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared"
-	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/shared/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
@@ -118,7 +118,7 @@ func (c *ClickHouseConnector) syncRecordsViaAvro(
 		req.Records.GetRecords(), tableNameRowsMapping, syncBatchID, unboundedNumericAsString,
 		protos.DBType_CLICKHOUSE,
 	)
-	numericTruncator := model.NewStreamNumericTruncator(req.TableMappings, peerdb_clickhouse.NumericDestinationTypes)
+	numericTruncator := model.NewStreamNumericTruncator(req.TableMappings, NumericDestinationTypes)
 	stream, err := utils.RecordsToRawTableStream(streamReq, numericTruncator)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert records to raw table stream: %w", err)

flow/connectors/clickhouse/clickhouse.go

@@ -22,8 +22,8 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
+	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/pkg/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared"
-	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/shared/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 

flow/connectors/clickhouse/normalize.go

@@ -19,8 +19,8 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/model/qvalue"
+	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/pkg/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared"
-	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/shared/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 

flow/connectors/clickhouse/normalize_query.go

@@ -10,7 +10,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/model/qvalue"
-	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/shared/clickhouse"
+	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/pkg/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 

flow/connectors/clickhouse/type_conversion.go

@@ -2,13 +2,69 @@ package connclickhouse
 
 import (
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
-	"github.com/PeerDB-io/peerdb/flow/shared/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
+/*
+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/shared/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).
+*/
+var SupportedDestinationTypes = map[string][]types.TypeConversion{
+	"String": {
+		types.NewTypeConversion(
+			types.NumericToStringSchemaConversion,
+			types.NumericToStringValueConversion,
+		),
+	},
+	"Int256": {
+		types.NewTypeConversion(
+			types.NumericToInt256SchemaConversion,
+			types.NumericToInt256ValueConversion,
+		),
+	},
+	"UInt256": {
+		types.NewTypeConversion(
+			types.NumericToUInt256SchemaConversion,
+			types.NumericToUInt256ValueConversion,
+		),
+	},
+}
+
+var NumericDestinationTypes = map[string]struct{}{
+	"String":  {},
+	"Int256":  {},
+	"UInt256": {},
+}
+
+// returns the full list of supported type conversions. The keys are
+// QValueKind to allows the implementation to be source-connector agnostic.
+func ListSupportedTypeConversions() map[types.QValueKind][]string {
+	typeConversions := make(map[types.QValueKind][]string)
+
+	for dstType, l := range SupportedDestinationTypes {
+		for _, conversion := range l {
+			typeConversions[conversion.FromKind()] = append(typeConversions[conversion.FromKind()], dstType)
+		}
+	}
+	return typeConversions
+}
+
 func GetColumnsTypeConversion() (*protos.ColumnsTypeConversionResponse, error) {
 	res := make([]*protos.ColumnsTypeConversion, 0)
-	for qkind, destTypes := range clickhouse.ListSupportedTypeConversions() {
+	for qkind, destTypes := range ListSupportedTypeConversions() {
 		res = append(res, &protos.ColumnsTypeConversion{
 			Qkind:            string(qkind),
 			DestinationTypes: destTypes,
@@ -32,7 +88,7 @@ func findTypeConversions(schema types.QRecordSchema, columns []*protos.ColumnSet
 		if !exist {
 			continue
 		}
-		conversions, exist := clickhouse.SupportedDestinationTypes[col.DestinationType]
+		conversions, exist := SupportedDestinationTypes[col.DestinationType]
 		if !exist {
 			continue
 		}

flow/connectors/clickhouse/validate.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
-	chvalidate "github.com/PeerDB-io/peerdb/flow/shared/clickhouse"
+	chvalidate "github.com/PeerDB-io/peerdb/flow/pkg/clickhouse"
 )
 
 func (c *ClickHouseConnector) ValidateMirrorDestination(

flow/connectors/mongo/mongo.go

@@ -21,8 +21,8 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
+	peerdb_mongo "github.com/PeerDB-io/peerdb/flow/pkg/mongo"
 	"github.com/PeerDB-io/peerdb/flow/shared"
-	peerdb_mongo "github.com/PeerDB-io/peerdb/flow/shared/mongo"
 )
 
 const (

flow/connectors/mongo/qrep.go

@@ -17,7 +17,6 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/otel_metrics"
 	"github.com/PeerDB-io/peerdb/flow/shared"
-	shared_mongo "github.com/PeerDB-io/peerdb/flow/shared/mongo"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
@@ -28,7 +27,7 @@ func (c *MongoConnector) GetQRepPartitions(
 ) ([]*protos.QRepPartition, error) {
 	fullTablePartition := []*protos.QRepPartition{
 		{
-			PartitionId:        shared_mongo.MongoFullTablePartitionId,
+			PartitionId:        utils.FullTablePartitionID,
 			Range:              nil,
 			FullTablePartition: true,
 		},

flow/connectors/mongo/schema.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
-	shared_mongo "github.com/PeerDB-io/peerdb/flow/shared/mongo"
+	shared_mongo "github.com/PeerDB-io/peerdb/flow/pkg/mongo"
 )
 
 func (c *MongoConnector) GetAllTables(ctx context.Context) (*protos.AllTablesResponse, error) {