feat: warehouse transformer (#5205)

Author: achettyiitr

go.mod

@@ -42,6 +42,7 @@ require (
 	github.com/databricks/databricks-sql-go v1.6.1
 	github.com/denisenkom/go-mssqldb v0.12.3
 	github.com/dgraph-io/badger/v4 v4.5.0
+	github.com/dlclark/regexp2 v1.11.4
 	github.com/docker/docker v27.5.0+incompatible
 	github.com/go-chi/chi/v5 v5.2.0
 	github.com/go-redis/redis v6.15.9+incompatible
@@ -193,7 +194,6 @@ require (
 	github.com/danieljoos/wincred v1.2.2 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
-	github.com/dlclark/regexp2 v1.11.4 // indirect
 	github.com/dnephin/pflag v1.0.7 // indirect
 	github.com/docker/cli v27.2.1+incompatible // indirect
 	github.com/docker/cli-docs-tool v0.8.0 // indirect

processor/processor.go

@@ -15,6 +15,7 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/rudderlabs/rudder-server/enterprise/trackedusers"
+	whutils "github.com/rudderlabs/rudder-server/warehouse/utils"
 
 	"golang.org/x/sync/errgroup"
 
@@ -57,6 +58,7 @@ import (
 	. "github.com/rudderlabs/rudder-server/utils/tx" //nolint:staticcheck
 	"github.com/rudderlabs/rudder-server/utils/types"
 	"github.com/rudderlabs/rudder-server/utils/workerpool"
+	wtrans "github.com/rudderlabs/rudder-server/warehouse/transformer"
 )
 
 const (
@@ -84,12 +86,18 @@ type trackedUsersReporter interface {
 	GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceIdFilter map[string]bool) []*trackedusers.UsersReport
 }
 
+type warehouseTransformer interface {
+	transformer.DestinationTransformer
+	CompareAndLog(events []transformer.TransformerEvent, pResponse, wResponse transformer.Response, metadata *transformer.Metadata, eventsByMessageID map[string]types.SingularEventWithReceivedAt)
+}
+
 // Handle is a handle to the processor module
 type Handle struct {
-	conf          *config.Config
-	tracer        stats.Tracer
-	backendConfig backendconfig.BackendConfig
-	transformer   transformer.Transformer
+	conf                 *config.Config
+	tracer               stats.Tracer
+	backendConfig        backendconfig.BackendConfig
+	transformer          transformer.Transformer
+	warehouseTransformer warehouseTransformer
 
 	gatewayDB                  jobsdb.JobsDB
 	routerDB                   jobsdb.JobsDB
@@ -158,6 +166,7 @@ type Handle struct {
 		eventAuditEnabled               map[string]bool
 		credentialsMap                  map[string][]transformer.Credential
 		nonEventStreamSources           map[string]bool
+		enableWarehouseTransformations  config.ValueLoader[bool]
 	}
 
 	drainConfig struct {
@@ -617,6 +626,9 @@ func (proc *Handle) Setup(
 			"partition": partition,
 		})
 	}
+
+	proc.warehouseTransformer = wtrans.New(proc.conf, proc.logger, proc.statsFactory)
+
 	if proc.config.enableDedup {
 		var err error
 		proc.dedup, err = dedup.New(proc.conf, proc.statsFactory)
@@ -817,6 +829,7 @@ func (proc *Handle) loadReloadableConfig(defaultPayloadLimit int64, defaultMaxEv
 	proc.config.archivalEnabled = config.GetReloadableBoolVar(true, "archival.Enabled")
 	// Capture event name as a tag in event level stats
 	proc.config.captureEventNameStats = config.GetReloadableBoolVar(false, "Processor.Stats.captureEventName")
+	proc.config.enableWarehouseTransformations = config.GetReloadableBoolVar(false, "Processor.enableWarehouseTransformations")
 }
 
 type connection struct {
@@ -2902,6 +2915,7 @@ func (proc *Handle) transformSrcDest(
 			proc.logger.Debug("Dest Transform input size", len(eventsToTransform))
 			s := time.Now()
 			response = proc.transformer.Transform(ctx, eventsToTransform, proc.config.transformBatchSize.Load())
+			proc.handleWarehouseTransformations(ctx, eventsToTransform, response, commonMetaData, eventsByMessageID)
 
 			destTransformationStat := proc.newDestinationTransformationStat(sourceID, workspaceID, transformAt, destination)
 			destTransformationStat.transformTime.Since(s)
@@ -3060,6 +3074,27 @@ func (proc *Handle) transformSrcDest(
 	}
 }
 
+func (proc *Handle) handleWarehouseTransformations(
+	ctx context.Context,
+	eventsToTransform []transformer.TransformerEvent,
+	pResponse transformer.Response,
+	commonMetaData *transformer.Metadata,
+	eventsByMessageID map[string]types.SingularEventWithReceivedAt,
+) {
+	if len(eventsToTransform) == 0 {
+		return
+	}
+	if _, ok := whutils.WarehouseDestinationMap[commonMetaData.DestinationType]; !ok {
+		return
+	}
+	if !proc.config.enableWarehouseTransformations.Load() {
+		return
+	}
+
+	wResponse := proc.warehouseTransformer.Transform(ctx, eventsToTransform, proc.config.transformBatchSize.Load())
+	proc.warehouseTransformer.CompareAndLog(eventsToTransform, pResponse, wResponse, commonMetaData, eventsByMessageID)
+}
+
 func (proc *Handle) saveDroppedJobs(ctx context.Context, droppedJobs []*jobsdb.JobT, tx *Tx) error {
 	if len(droppedJobs) > 0 {
 		for i := range droppedJobs { // each dropped job should have a unique jobID in the scope of the batch

processor/transformer/transformer.go

@@ -164,13 +164,25 @@ func WithClient(client HTTPDoer) Opt {
 	}
 }
 
-// Transformer provides methods to transform events
-type Transformer interface {
-	Transform(ctx context.Context, clientEvents []TransformerEvent, batchSize int) Response
+type UserTransformer interface {
 	UserTransform(ctx context.Context, clientEvents []TransformerEvent, batchSize int) Response
+}
+
+type DestinationTransformer interface {
+	Transform(ctx context.Context, clientEvents []TransformerEvent, batchSize int) Response
+}
+
+type TrackingPlanValidator interface {
 	Validate(ctx context.Context, clientEvents []TransformerEvent, batchSize int) Response
 }
 
+// Transformer provides methods to transform events
+type Transformer interface {
+	UserTransformer
+	DestinationTransformer
+	TrackingPlanValidator
+}
+
 type HTTPDoer interface {
 	Do(req *http.Request) (*http.Response, error)
 }
@@ -591,7 +603,7 @@ func (trans *handle) destTransformURL(destType string) string {
 	destinationEndPoint := fmt.Sprintf("%s/v0/destinations/%s", trans.config.destTransformationURL, strings.ToLower(destType))
 
 	if _, ok := warehouseutils.WarehouseDestinationMap[destType]; ok {
-		whSchemaVersionQueryParam := fmt.Sprintf("whSchemaVersion=%s&whIDResolve=%v", trans.conf.GetString("Warehouse.schemaVersion", "v1"), warehouseutils.IDResolutionEnabled())
+		whSchemaVersionQueryParam := fmt.Sprintf("whIDResolve=%t", trans.conf.GetBool("Warehouse.enableIDResolution", false))
 		switch destType {
 		case warehouseutils.RS:
 			return destinationEndPoint + "?" + whSchemaVersionQueryParam
@@ -603,7 +615,7 @@ func (trans *handle) destTransformURL(destType string) string {
 		}
 	}
 	if destType == warehouseutils.SnowpipeStreaming {
-		return fmt.Sprintf("%s?whSchemaVersion=%s&whIDResolve=%t", destinationEndPoint, trans.conf.GetString("Warehouse.schemaVersion", "v1"), warehouseutils.IDResolutionEnabled())
+		return fmt.Sprintf("%s?whIDResolve=%t", destinationEndPoint, trans.conf.GetBool("Warehouse.enableIDResolution", false))
 	}
 	return destinationEndPoint
 }

runner/buckets.go

@@ -145,5 +145,18 @@ var (
 			// 1ms, 5ms, 10ms, 25ms, 50ms, 100ms, 250ms, 500ms, 1s
 			0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1,
 		},
+
+		"warehouse_dest_transform_mismatched_events": {
+			0.1, 0.25, 0.5, 1, 2.5, 5, 7.5, 10, 12.5, 15, 30, 60, 120, 300, 600, 900, 1800, 3600,
+		},
+		"warehouse_dest_transform_input_events": {
+			1, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 12500, 15000, 30000, 50000, 75000, 100000,
+		},
+		"warehouse_dest_transform_output_events": {
+			1, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 12500, 15000, 30000, 50000, 75000, 100000,
+		},
+		"warehouse_dest_transform_output_failed_events": {
+			1, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 12500, 15000, 30000, 50000, 75000, 100000,
+		},
 	}
 )

warehouse/internal/model/schema.go

@@ -17,7 +17,7 @@ const (
 	JSONDataType           SchemaType = "json"
 	TextDataType           SchemaType = "text"
 	DateTimeDataType       SchemaType = "datetime"
-	ArrayOfBooleanDatatype SchemaType = "array(boolean)"
+	ArrayOfBooleanDataType SchemaType = "array(boolean)"
 )
 
 type WHSchema struct {

warehouse/slave/worker.go

@@ -319,7 +319,7 @@ func (w *worker) processStagingFile(ctx context.Context, job payload) ([]uploadR
 					}
 
 					columnVal = newColumnVal
-				case model.ArrayOfBooleanDatatype:
+				case model.ArrayOfBooleanDataType:
 					if boolValue, ok := columnVal.([]interface{}); ok {
 						newColumnVal := make([]interface{}, len(boolValue))
 

warehouse/transformer/datatype.go

@@ -0,0 +1,94 @@
+package transformer
+
+import (
+	"reflect"
+
+	"github.com/rudderlabs/rudder-server/warehouse/internal/model"
+	"github.com/rudderlabs/rudder-server/warehouse/transformer/internal/utils"
+	whutils "github.com/rudderlabs/rudder-server/warehouse/utils"
+)
+
+func dataTypeFor(destType, key string, val any, isJSONKey bool) string {
+	if typeName := primitiveType(val); typeName != "" {
+		return typeName
+	}
+	if strVal, ok := val.(string); ok && utils.ValidTimestamp(strVal) {
+		return model.DateTimeDataType
+	}
+	if override := dataTypeOverride(destType, key, val, isJSONKey); override != "" {
+		return override
+	}
+	return model.StringDataType
+}
+
+func primitiveType(val any) string {
+	switch v := val.(type) {
+	case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64:
+		return model.IntDataType
+	case float64:
+		return getFloatType(v)
+	case float32:
+		return getFloatType(float64(v))
+	case bool:
+		return model.BooleanDataType
+	default:
+		return ""
+	}
+}
+
+func getFloatType(v float64) string {
+	// JSON unmarshalling treats all numbers as float64 by default, even if they are whole numbers
+	// So, we need to check if the float is actually an integer
+	if v == float64(int64(v)) {
+		return model.IntDataType
+	}
+	return model.FloatDataType
+}
+
+func dataTypeOverride(destType, key string, val any, isJSONKey bool) string {
+	switch destType {
+	case whutils.POSTGRES, whutils.SNOWFLAKE, whutils.SnowpipeStreaming:
+		return overrideForPostgresSnowflake(key, isJSONKey)
+	case whutils.RS:
+		return overrideForRedshift(val, isJSONKey)
+	default:
+		return ""
+	}
+}
+
+func overrideForPostgresSnowflake(key string, isJSONKey bool) string {
+	if isJSONKey || key == violationErrors {
+		return model.JSONDataType
+	}
+	return model.StringDataType
+}
+
+func overrideForRedshift(val any, isJSONKey bool) string {
+	if isJSONKey {
+		return model.JSONDataType
+	}
+	if val == nil {
+		return model.StringDataType
+	}
+	switch reflect.TypeOf(val).Kind() {
+	case reflect.Slice, reflect.Array:
+		if jsonVal, _ := json.Marshal(val); len(jsonVal) > redshiftStringLimit {
+			return model.TextDataType
+		}
+		return model.StringDataType
+	case reflect.String:
+		if len(val.(string)) > redshiftStringLimit {
+			return model.TextDataType
+		}
+		return model.StringDataType
+	default:
+		return model.StringDataType
+	}
+}
+
+func convertValIfDateTime(val any, colType string) any {
+	if colType == model.DateTimeDataType {
+		return utils.ToTimestamp(val)
+	}
+	return val
+}