[bigquery] Move encodeStructToJSONString function (#742)

clients/bigquery/dialect/dialect.go

@@ -14,8 +14,6 @@ import (
 )
 
 const (
-	BQStreamingTimeFormat = "15:04:05"
-
 	bqLayout = "2006-01-02 15:04:05 MST"
 )
 

clients/bigquery/storagewrite.go

@@ -1,13 +1,17 @@
 package bigquery
 
 import (
+	"encoding/json"
 	"fmt"
+	"log/slog"
 	"strconv"
+	"strings"
 	"time"
 
 	"cloud.google.com/go/bigquery/storage/apiv1/storagepb"
 	"cloud.google.com/go/bigquery/storage/managedwriter/adapt"
 	"github.com/artie-labs/transfer/lib/array"
+	"github.com/artie-labs/transfer/lib/config/constants"
 	"github.com/artie-labs/transfer/lib/typing"
 	"github.com/artie-labs/transfer/lib/typing/columns"
 	"github.com/artie-labs/transfer/lib/typing/decimal"
@@ -188,7 +192,7 @@ func rowToMessage(row map[string]any, columns []columns.Column, messageDescripto
 				return nil, fmt.Errorf("unsupported extended time details: %q", column.KindDetails.ExtendedTimeDetails.Type)
 			}
 		case typing.Struct.Kind:
-			stringValue, err := EncodeStructToJSONString(value)
+			stringValue, err := encodeStructToJSONString(value)
 			if err != nil {
 				return nil, err
 			} else if stringValue == "" {
@@ -211,3 +215,29 @@ func rowToMessage(row map[string]any, columns []columns.Column, messageDescripto
 	}
 	return message, nil
 }
+
+// encodeStructToJSONString takes a struct as either a string or Go object and encodes it into a JSON string.
+// Structs from relational and Mongo are different.
+// MongoDB will return the native objects back such as `map[string]any{"hello": "world"}`
+// Relational will return a string representation of the struct such as `{"hello": "world"}`
+func encodeStructToJSONString(value any) (string, error) {
+	if stringValue, isOk := value.(string); isOk {
+		if strings.Contains(stringValue, constants.ToastUnavailableValuePlaceholder) {
+			return fmt.Sprintf(`{"key":"%s"}`, constants.ToastUnavailableValuePlaceholder), nil
+		}
+		return stringValue, nil
+	}
+
+	bytes, err := json.Marshal(value)
+	if err != nil {
+		return "", fmt.Errorf("failed to marshal value: %w", err)
+	}
+
+	stringValue := string(bytes)
+	if strings.Contains(stringValue, constants.ToastUnavailableValuePlaceholder) {
+		// TODO: Remove this if we don't see it in the logs.
+		slog.Error("encoded JSON value contains the toast unavailable value placeholder")
+		return fmt.Sprintf(`{"key":"%s"}`, constants.ToastUnavailableValuePlaceholder), nil
+	}
+	return stringValue, nil
+}

clients/bigquery/storagewrite_test.go

@@ -171,3 +171,30 @@ func TestRowToMessage(t *testing.T) {
 		"cArray":         []any{"foo", "bar"},
 	}, result)
 }
+
+func TestEncodeStructToJSONString(t *testing.T) {
+	{
+		// Empty string:
+		result, err := encodeStructToJSONString("")
+		assert.NoError(t, err)
+		assert.Equal(t, "", result)
+	}
+	{
+		// Toasted string:
+		result, err := encodeStructToJSONString("__debezium_unavailable_value")
+		assert.NoError(t, err)
+		assert.Equal(t, `{"key":"__debezium_unavailable_value"}`, result)
+	}
+	{
+		// Map:
+		result, err := encodeStructToJSONString(map[string]any{"foo": "bar", "baz": 1234})
+		assert.NoError(t, err)
+		assert.Equal(t, `{"baz":1234,"foo":"bar"}`, result)
+	}
+	{
+		// Toasted map (should not happen):
+		result, err := encodeStructToJSONString(map[string]any{"__debezium_unavailable_value": "bar", "baz": 1234})
+		assert.NoError(t, err)
+		assert.Equal(t, `{"key":"__debezium_unavailable_value"}`, result)
+	}
+}

lib/typing/ext/variables.go

@@ -3,7 +3,6 @@ package ext
 import "time"
 
 const (
-	BigQueryDateTimeFormat = "2006-01-02 15:04:05.999999"
 	ISO8601                = "2006-01-02T15:04:05.999-07:00"
 	PostgresDateFormat     = "2006-01-02"
 	PostgresTimeFormat     = "15:04:05.999999-07" // microsecond precision