Merge branch 'master' into refactor-storage-writer

clients/bigquery/converters/converters.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"strconv"
 
+	"github.com/artie-labs/transfer/lib/typing/ext"
+
 	"github.com/artie-labs/transfer/lib/typing/decimal"
 )
 
@@ -17,6 +19,8 @@ func (StringConverter) Convert(value any) (any, error) {
 		return castedValue.String(), nil
 	case bool:
 		return fmt.Sprint(castedValue), nil
+	case *ext.ExtendedTime:
+		return castedValue.String(""), nil
 	default:
 		return nil, fmt.Errorf("expected string/*decimal.Decimal/bool received %T with value %v", value, value)
 	}

clients/bigquery/storagewrite.go

@@ -151,14 +151,21 @@ func rowToMessage(row map[string]any, columns []columns.Column, messageDescripto
 				message.Set(field, protoreflect.ValueOfFloat64(float64(value)))
 			case int64:
 				message.Set(field, protoreflect.ValueOfFloat64(float64(value)))
+			case *decimal.Decimal:
+				float, err := value.Value().Float64()
+				if err != nil {
+					return nil, fmt.Errorf("failed to convert decimal to float64: %w", err)
+				}
+
+				message.Set(field, protoreflect.ValueOfFloat64(float))
 			case string:
 				floatValue, err := strconv.ParseFloat(value, 64)
 				if err != nil {
 					return nil, fmt.Errorf("failed to parse string to float64: %w", err)
 				}
 				message.Set(field, protoreflect.ValueOfFloat64(floatValue))
 			default:
-				return nil, fmt.Errorf("expected float32/float64/int32/int64/string received %T with value %v", value, value)
+				return nil, fmt.Errorf("expected float32/float64/int32/int64/*decimal.Decimal/string received %T with value %v", value, value)
 			}
 		case typing.EDecimal.Kind:
 			decimalValue, err := typing.AssertType[*decimal.Decimal](value)