From 87ef8b6bfa3f7b2eb9b148620e6b40d668f39693 Mon Sep 17 00:00:00 2001
From: Robin Tang <rtang.cs@gmail.com>
Date: Sun, 24 Nov 2024 12:22:38 -0800
Subject: [PATCH] TOASTED Arrays (#1066)

---
 clients/mssql/dialect/dialect.go      |  1 +
 lib/debezium/converters/basic.go      | 14 ++++++++++++++
 lib/debezium/converters/basic_test.go | 26 ++++++++++++++++++++++++++
 lib/debezium/schema.go                |  4 ++--
 4 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/clients/mssql/dialect/dialect.go b/clients/mssql/dialect/dialect.go
index 9359e95a4..5f120c053 100644
--- a/clients/mssql/dialect/dialect.go
+++ b/clients/mssql/dialect/dialect.go
@@ -49,6 +49,7 @@ func (md MSSQLDialect) BuildIsNotToastValueExpression(tableAlias constants.Table
 	if column.KindDetails == typing.Struct {
 		return fmt.Sprintf("COALESCE(%s, {}) != {'key': '%s'}", colName, constants.ToastUnavailableValuePlaceholder)
 	}
+
 	return fmt.Sprintf("COALESCE(%s, '') != '%s'", colName, constants.ToastUnavailableValuePlaceholder)
 }
 
diff --git a/lib/debezium/converters/basic.go b/lib/debezium/converters/basic.go
index 60e4ccb97..ff4a9de3f 100644
--- a/lib/debezium/converters/basic.go
+++ b/lib/debezium/converters/basic.go
@@ -87,3 +87,17 @@ func (Float64) Convert(value any) (any, error) {
 		return nil, fmt.Errorf("unexpected type %T", value)
 	}
 }
+
+type Array struct{}
+
+func (Array) ToKindDetails() typing.KindDetails {
+	return typing.Array
+}
+
+func (Array) Convert(value any) (any, error) {
+	if fmt.Sprint(value) == fmt.Sprintf("[%s]", constants.ToastUnavailableValuePlaceholder) {
+		return constants.ToastUnavailableValuePlaceholder, nil
+	}
+
+	return value, nil
+}
diff --git a/lib/debezium/converters/basic_test.go b/lib/debezium/converters/basic_test.go
index 95d5d130e..941befe0e 100644
--- a/lib/debezium/converters/basic_test.go
+++ b/lib/debezium/converters/basic_test.go
@@ -4,6 +4,8 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+
+	"github.com/artie-labs/transfer/lib/config/constants"
 )
 
 func TestJSON_Convert(t *testing.T) {
@@ -85,3 +87,27 @@ func TestFloat64_Convert(t *testing.T) {
 		}
 	}
 }
+
+func TestArray_Convert(t *testing.T) {
+	{
+		// Irrelevant data type
+		value, err := Array{}.Convert([]int{1, 2, 3, 4})
+		assert.NoError(t, err)
+		assert.Equal(t, []int{1, 2, 3, 4}, value)
+	}
+	{
+		// TOASTED data
+		{
+			// As []any
+			value, err := Array{}.Convert([]any{constants.ToastUnavailableValuePlaceholder})
+			assert.NoError(t, err)
+			assert.Equal(t, constants.ToastUnavailableValuePlaceholder, value)
+		}
+		{
+			// As []string
+			value, err := Array{}.Convert([]string{constants.ToastUnavailableValuePlaceholder})
+			assert.NoError(t, err)
+			assert.Equal(t, constants.ToastUnavailableValuePlaceholder, value)
+		}
+	}
+}
diff --git a/lib/debezium/schema.go b/lib/debezium/schema.go
index 5702cb753..539a5bf5b 100644
--- a/lib/debezium/schema.go
+++ b/lib/debezium/schema.go
@@ -137,6 +137,8 @@ func (f Field) ToValueConverter() (converters.ValueConverter, error) {
 		}
 
 		switch f.Type {
+		case Array:
+			return converters.Array{}, nil
 		case Double, Float:
 			return converters.Float64{}, nil
 		}
@@ -167,8 +169,6 @@ func (f Field) ToKindDetails() (typing.KindDetails, error) {
 		return typing.Struct, nil
 	case Boolean:
 		return typing.Boolean, nil
-	case Array:
-		return typing.Array, nil
 	default:
 		return typing.Invalid, fmt.Errorf("unhandled field type %q", f.Type)
 	}