[Table Config] Removing empty comment as a struct (#810)

clients/bigquery/bigquery.go

@@ -25,7 +25,6 @@ import (
 	"github.com/artie-labs/transfer/lib/kafkalib"
 	"github.com/artie-labs/transfer/lib/logger"
 	"github.com/artie-labs/transfer/lib/optimization"
-	"github.com/artie-labs/transfer/lib/ptr"
 	"github.com/artie-labs/transfer/lib/sql"
 )
 
@@ -118,7 +117,6 @@ func (s *Store) GetTableConfig(tableData *optimization.TableData) (*types.DwhTab
 		ColumnNameForName:     "column_name",
 		ColumnNameForDataType: "data_type",
 		ColumnNameForComment:  "description",
-		EmptyCommentValue:     ptr.ToString(""),
 		DropDeletedColumns:    tableData.TopicConfig().DropDeletedColumns,
 	}.GetTableConfig()
 }

clients/mssql/store.go

@@ -12,7 +12,6 @@ import (
 	"github.com/artie-labs/transfer/lib/destination/types"
 	"github.com/artie-labs/transfer/lib/kafkalib"
 	"github.com/artie-labs/transfer/lib/optimization"
-	"github.com/artie-labs/transfer/lib/ptr"
 	"github.com/artie-labs/transfer/lib/sql"
 )
 
@@ -87,7 +86,6 @@ func (s *Store) GetTableConfig(tableData *optimization.TableData) (*types.DwhTab
 		ColumnNameForName:     "column_name",
 		ColumnNameForDataType: "data_type",
 		ColumnNameForComment:  "description",
-		EmptyCommentValue:     ptr.ToString("<nil>"),
 		DropDeletedColumns:    tableData.TopicConfig().DropDeletedColumns,
 	}.GetTableConfig()
 }

clients/redshift/redshift.go

@@ -14,7 +14,6 @@ import (
 	"github.com/artie-labs/transfer/lib/destination/types"
 	"github.com/artie-labs/transfer/lib/kafkalib"
 	"github.com/artie-labs/transfer/lib/optimization"
-	"github.com/artie-labs/transfer/lib/ptr"
 	"github.com/artie-labs/transfer/lib/sql"
 )
 
@@ -71,7 +70,6 @@ func (s *Store) GetTableConfig(tableData *optimization.TableData) (*types.DwhTab
 		ColumnNameForName:     "column_name",
 		ColumnNameForDataType: "data_type",
 		ColumnNameForComment:  "description",
-		EmptyCommentValue:     ptr.ToString("<nil>"),
 		DropDeletedColumns:    tableData.TopicConfig().DropDeletedColumns,
 	}.GetTableConfig()
 }

clients/shared/table_config.go

@@ -27,20 +27,9 @@ type GetTableCfgArgs struct {
 	ColumnNameForDataType string
 	// Description of the column (used to annotate whether we need to backfill or not)
 	ColumnNameForComment string
-	EmptyCommentValue    *string
 	DropDeletedColumns   bool
 }
 
-func (g GetTableCfgArgs) ShouldParseComment(comment string) bool {
-	if g.EmptyCommentValue != nil && comment == *g.EmptyCommentValue {
-		return false
-	}
-
-	// Snowflake and Redshift both will return `<nil>` if the comment does not exist, this will check the value.
-	// BigQuery returns ""
-	return true
-}
-
 func (g GetTableCfgArgs) GetTableConfig() (*types.DwhTableConfig, error) {
 	if tableConfig := g.ConfigMap.TableConfigCache(g.TableID); tableConfig != nil {
 		return tableConfig, nil
@@ -97,28 +86,32 @@ func (g GetTableCfgArgs) GetTableConfig() (*types.DwhTableConfig, error) {
 				return nil, errors.New("invalid value")
 			}
 
-			row[columnNameList[idx]] = strings.ToLower(fmt.Sprint(*interfaceVal))
+			var value string
+			if *interfaceVal != nil {
+				value = strings.ToLower(fmt.Sprint(*interfaceVal))
+			}
+
+			row[columnNameList[idx]] = value
 		}
 
 		kindDetails, err := g.Dwh.Dialect().KindForDataType(row[g.ColumnNameForDataType], row[constants.StrPrecisionCol])
 		if err != nil {
 			return nil, fmt.Errorf("failed to get kind details: %w", err)
 		}
+
 		if kindDetails.Kind == typing.Invalid.Kind {
 			return nil, fmt.Errorf("failed to get kind details: unable to map type: %q to dwh type", row[g.ColumnNameForDataType])
 		}
 
 		col := columns.NewColumn(row[g.ColumnNameForName], kindDetails)
-		comment, isOk := row[g.ColumnNameForComment]
-		if isOk && g.ShouldParseComment(comment) {
-			// Try to parse the comment.
+		// We need to check to make sure the comment is not an empty string
+		if comment, isOk := row[g.ColumnNameForComment]; isOk && comment != "" {
 			var _colComment constants.ColComment
 			if err = json.Unmarshal([]byte(comment), &_colComment); err != nil {
-				return nil, fmt.Errorf("failed to unmarshal comment: %w", err)
+				return nil, fmt.Errorf("failed to unmarshal comment %q: %w", comment, err)
 			}
 
 			col.SetBackfilled(_colComment.Backfilled)
-
 		}
 
 		cols.AddColumn(col)

clients/shared/table_config_test.go

@@ -6,48 +6,11 @@ import (
 
 	"github.com/artie-labs/transfer/lib/destination/types"
 	"github.com/artie-labs/transfer/lib/mocks"
-	"github.com/artie-labs/transfer/lib/ptr"
 	"github.com/artie-labs/transfer/lib/typing"
 	"github.com/artie-labs/transfer/lib/typing/columns"
 	"github.com/stretchr/testify/assert"
 )
 
-func TestGetTableCfgArgs_ShouldParseComment(t *testing.T) {
-	type _testCase struct {
-		Name            string
-		EmptyCommentVal *string
-		Comment         string
-		ExpectedResult  bool
-	}
-
-	testCases := []_testCase{
-		{
-			Name:           "empty comment val = nil",
-			Comment:        "blah blah blah",
-			ExpectedResult: true,
-		},
-		{
-			Name:            "empty comment val = blah",
-			EmptyCommentVal: ptr.ToString("blah"),
-			Comment:         "blah",
-		},
-		{
-			Name:            "empty comment val = blah2",
-			EmptyCommentVal: ptr.ToString("blah2"),
-			Comment:         "blah",
-			ExpectedResult:  true,
-		},
-	}
-
-	for _, testCase := range testCases {
-		args := GetTableCfgArgs{
-			EmptyCommentValue: testCase.EmptyCommentVal,
-		}
-
-		assert.Equal(t, testCase.ExpectedResult, args.ShouldParseComment(testCase.Comment), testCase.Name)
-	}
-}
-
 func TestGetTableConfig(t *testing.T) {
 	// Return early because table is found in configMap.
 	cols := &columns.Columns{}

clients/snowflake/snowflake.go

@@ -41,7 +41,6 @@ func (s *Store) GetTableConfig(tableData *optimization.TableData) (*types.DwhTab
 		ColumnNameForName:     "name",
 		ColumnNameForDataType: "type",
 		ColumnNameForComment:  "comment",
-		EmptyCommentValue:     ptr.ToString("<nil>"),
 		DropDeletedColumns:    tableData.TopicConfig().DropDeletedColumns,
 	}.GetTableConfig()
 }