feat: warehouse transformer

go.mod

@@ -41,6 +41,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.3.1
+ github.com/dlclark/regexp2 v1.11.4
  github.com/docker/docker v27.3.1+incompatible
  github.com/go-chi/chi/v5 v5.1.0
  github.com/go-redis/redis v6.15.9+incompatible
@@ -187,7 +188,6 @@ require (
  github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
  github.com/dgraph-io/ristretto v1.0.0 // 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/transformer/transformer.go

@@ -140,13 +140,25 @@ func WithClient(client *http.Client) 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
+}
+
 // handle is the handle for this class
 type handle struct {
  sentStat stats.Measurement
@@ -526,7 +538,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

warehouse/transformer/alias.go

@@ -0,0 +1,62 @@
+package transformer
+
+import (
+ "fmt"
+
+ "github.com/samber/lo"
+
+ "github.com/rudderlabs/rudder-server/warehouse/transformer/internal/rules"
+)
+
+func (t *transformer) handleAliasEvent(pi *processingInfo) ([]map[string]any, error) {
+ event := make(map[string]any)
+ columnTypes := make(map[string]string)
+
+ if err := t.setDataAndColumnTypeFromInput(pi, pi.event.Message["traits"], event, columnTypes, &prefixInfo{
+ completePrefix: "alias_traits_",
+ completeLevel: 2,
+ }); err != nil {
+ return nil, fmt.Errorf("alias: setting data and column types from input: %w", err)
+ }
+ if err := t.setDataAndColumnTypeFromInput(pi, pi.event.Message["context"], event, columnTypes, &prefixInfo{
+ completePrefix: "alias_context_",
+ completeLevel: 2,
+ prefix: "context_",
+ }); err != nil {
+ return nil, fmt.Errorf("alias: setting data and column types from input: %w", err)
+ }
+ if err := t.setDataAndColumnTypeFromRules(pi, event, columnTypes,
+ lo.Assign(rules.DefaultRules, rules.AliasRules), rules.DefaultFunctionalRules,
+ ); err != nil {
+ return nil, fmt.Errorf("alias: setting data and column types from rules: %w", err)
+ }
+
+ if err := storeRudderEvent(pi, event, columnTypes); err != nil {
+ return nil, fmt.Errorf("alias: storing rudder event: %w", err)
+ }
+
+ tableName, err := SafeTableName(pi.event.Metadata.DestinationType, pi.itrOpts, "aliases")
+ if err != nil {
+ return nil, fmt.Errorf("alias: safe table name: %w", err)
+ }
+ columns, err := t.getColumns(pi.event.Metadata.DestinationType, event, columnTypes)
+ if err != nil {
+ return nil, fmt.Errorf("alias: getting columns: %w", err)
+ }
+
+ mergeEvents, err := t.handleMergeEvent(pi)
+ if err != nil {
+ return nil, fmt.Errorf("merge event: %w", err)
+ }
+
+ aliasOutput := map[string]any{
+ "data": event,
+ "metadata": map[string]any{
+ "table": tableName,
+ "columns": columns,
+ "receivedAt": pi.event.Metadata.ReceivedAt,
+ },
+ "userId": "",
+ }
+ return append([]map[string]any{aliasOutput}, mergeEvents...), nil
+}