fix(streams): added stream name normalizer

usedatabrew · Feb 13, 2024 · 63f02fa · 63f02fa
1 parent 1bf6f87
commit 63f02fa
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 4 deletions.
diff --git a/internal/helper/stream.go b/internal/helper/stream.go
@@ -0,0 +1,12 @@
+package helper
+
+import "strings"
+
+func NormalizeStreamName(stream string) string {
+	var normalizedName = stream
+	if len(strings.Split(stream, ".")) == 2 {
+		normalizedName = strings.Split(stream, ".")[1]
+	}
+
+	return normalizedName
+}
diff --git a/internal/processors/sql/plugin.go b/internal/processors/sql/plugin.go
@@ -9,6 +9,7 @@ import (
 	"github.com/barkimedes/go-deepcopy"
 	"github.com/blastrain/vitess-sqlparser/sqlparser"
 	"github.com/charmbracelet/log"
+	"github.com/usedatabrew/blink/internal/helper"
 	"github.com/usedatabrew/blink/internal/schema"
 	"github.com/usedatabrew/blink/internal/stream_context"
 	"github.com/usedatabrew/message"
@@ -83,7 +84,7 @@ func (p *Plugin) EvolveSchema(streamSchema *schema.StreamSchemaObj) error {
 
 	var streamToProcess *schema.StreamSchema
 	for _, stream := range streamSchema.GetLatestSchema() {
-		if stream.StreamName == streamName {
+		if helper.NormalizeStreamName(stream.StreamName) == streamName {
 			streamToProcess = &stream
 		}
 	}

diff --git a/internal/schema/schema_obj.go b/internal/schema/schema_obj.go
@@ -3,6 +3,7 @@ package schema
 import (
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/barkimedes/go-deepcopy"
+	"github.com/usedatabrew/blink/internal/helper"
 	"slices"
 )
 
@@ -27,7 +28,7 @@ func (s *StreamSchemaObj) GetLatestSchema() []StreamSchema {
 func (s *StreamSchemaObj) AddField(streamName, name string, fieldType arrow.DataType, driverType string) {
 	var streamSchemaCopy = s.getLastSchemaDeepCopy()
 	for idx, stream := range streamSchemaCopy {
-		if stream.StreamName == streamName {
+		if helper.NormalizeStreamName(stream.StreamName) == streamName {
 			arrowColumn := Column{
 				Name:                name,
 				DatabrewType:        fieldType.String(),
@@ -53,7 +54,7 @@ func (s *StreamSchemaObj) FakeEvolve() {
 func (s *StreamSchemaObj) RemoveField(streamName, columnName string) {
 	var streamSchemaCopy = s.getLastSchemaDeepCopy()
 	for streamIndex, stream := range streamSchemaCopy {
-		if stream.StreamName == streamName {
+		if helper.NormalizeStreamName(stream.StreamName) == streamName {
 			for colIdx, column := range stream.Columns {
 				if column.Name == columnName {
 					streamSchemaCopy[streamIndex].Columns = remove(stream.Columns, colIdx)

diff --git a/public/stream/config_loader.go b/public/stream/config_loader.go
@@ -1,6 +1,7 @@
 package stream
 
 import (
+	"fmt"
 	"github.com/usedatabrew/blink/config"
 	"os"
 	"regexp"
@@ -23,7 +24,7 @@ func ReadInitConfigFromYaml(configBytes []byte) (config.Configuration, error) {
 	for _, ss := range conf.Source.StreamSchema {
 		ss.SortColumnsAsc()
 	}
-
+	fmt.Println(string(configBytes))
 	config.ValidateConfigSchema(conf)
 	return conf, err
 }