fix: Support list scalars from JSON (#1530)

Without this, lists from the test source (at least, could be more plugins/cases) get persisted as `[]`

I'm still not 100% sure how the lists end up getting encoded to JSON, probably we use Arrow's `.ValueStr` and/or `.GetOneForMarshal` somewhere.

Called from table resolver - source plugins need this fix.

Author: disq

scalar/list.go

@@ -1,6 +1,7 @@
 package scalar
 
 import (
+	"encoding/json"
 	"reflect"
 	"strings"
 
@@ -85,6 +86,49 @@ func (s *List) Set(val any) error {
 		return nil
 	}
 
+	switch value := val.(type) {
+	case string:
+		var x []any
+		if err := json.Unmarshal([]byte(value), &x); err != nil {
+			return err
+		}
+		length := len(x)
+		s.Value = make(Vector, length)
+		for i := 0; i < length; i++ {
+			s.Value[i] = NewScalar(s.Type.(*arrow.ListType).Elem())
+			if x[i] == nil {
+				continue
+			}
+			if err := s.Value[i].Set(x[i]); err != nil {
+				return err
+			}
+		}
+
+	case []byte:
+		var x []any
+		if err := json.Unmarshal(value, &x); err != nil {
+			return err
+		}
+		length := len(x)
+		s.Value = make(Vector, length)
+		for i := 0; i < length; i++ {
+			s.Value[i] = NewScalar(s.Type.(*arrow.ListType).Elem())
+			if x[i] == nil {
+				continue
+			}
+			if err := s.Value[i].Set(x[i]); err != nil {
+				return err
+			}
+		}
+
+	case *string:
+		if value == nil {
+			s.Valid = false
+			return nil
+		}
+		return s.Set(*value)
+	}
+
 	switch reflectedValue.Kind() {
 	case reflect.Array:
 		fallthrough

scalar/list_test.go

@@ -37,6 +37,15 @@ func TestListSet(t *testing.T) {
 			&Inet{Valid: false},
 			&Inet{Valid: false},
 		}, Valid: true, Type: arrow.ListOf(types.ExtensionTypes.Inet)}},
+		{source: `[1, 2]`, result: List{Value: []Scalar{
+			&Int{Value: 1, Valid: true},
+			&Int{Value: 2, Valid: true},
+		}, Valid: true, Type: arrow.ListOf(arrow.PrimitiveTypes.Int64)}},
+		{source: `[1, null, 2]`, result: List{Value: []Scalar{
+			&Int{Value: 1, Valid: true},
+			&Int{Valid: false},
+			&Int{Value: 2, Valid: true},
+		}, Valid: true, Type: arrow.ListOf(arrow.PrimitiveTypes.Int64)}},
 	}
 
 	for i, tt := range successfulTests {

scheduler/scheduler.go

@@ -11,11 +11,8 @@ import (
 
 	"github.com/apache/arrow/go/v15/arrow"
 
-	"github.com/apache/arrow/go/v15/arrow/array"
-	"github.com/apache/arrow/go/v15/arrow/memory"
 	"github.com/cloudquery/plugin-sdk/v4/caser"
 	"github.com/cloudquery/plugin-sdk/v4/message"
-	"github.com/cloudquery/plugin-sdk/v4/scalar"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 	"github.com/getsentry/sentry-go"
 	"github.com/rs/zerolog"
@@ -225,10 +222,7 @@ func (s *Scheduler) Sync(ctx context.Context, client schema.ClientMeta, tables s
 
 func resourceToRecord(resource *schema.Resource) arrow.Record {
 	vector := resource.GetValues()
-	bldr := array.NewRecordBuilder(memory.DefaultAllocator, resource.Table.ToArrowSchema())
-	scalar.AppendToRecordBuilder(bldr, vector)
-	rec := bldr.NewRecord()
-	return rec
+	return vector.ToArrowRecord(resource.Table.ToArrowSchema())
 }
 
 func (s *syncClient) logTablesMetrics(tables schema.Tables, client Client) {