Merge pull request #44 from Recidiviz/dan/struct-field-test

Bump to `v0.18.0-recidiviz.16.2`

Author: ohaibbq

go.mod

@@ -118,4 +118,4 @@ require (
 	modernc.org/sqlite v1.37.0
 )
 
-replace github.com/goccy/go-zetasqlite => github.com/Recidiviz/go-zetasqlite v0.18.0-recidiviz.16.1
+replace github.com/goccy/go-zetasqlite => github.com/Recidiviz/go-zetasqlite v0.18.0-recidiviz.16.2

go.sum

@@ -42,8 +42,8 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.51.0/go.mod h1:SZiPHWGOOk3bl8tkevxkoiwPgsIl6CwrWcbwjfHZpdM=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.51.0 h1:6/0iUd0xrnX7qt+mLNRwg5c0PGv8wpE8K90ryANQwMI=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.51.0/go.mod h1:otE2jQekW/PqXk1Awf5lmfokJx4uwuqcj1ab5SpGeW0=
-github.com/Recidiviz/go-zetasqlite v0.18.0-recidiviz.16.1 h1:OhF/RKqkgZ2hVdhdzlSs+a2EIZPRmeutfUfAeC5a8uM=
-github.com/Recidiviz/go-zetasqlite v0.18.0-recidiviz.16.1/go.mod h1:xtUAGxrJMK0vqv5Yj/AYvrcP3g338Tbh9oTyYk1VML8=
+github.com/Recidiviz/go-zetasqlite v0.18.0-recidiviz.16.2 h1:GoVmgEOtY7H0UpqIySDkOUm/bXWuapnA8qZ1CanWLZE=
+github.com/Recidiviz/go-zetasqlite v0.18.0-recidiviz.16.2/go.mod h1:xtUAGxrJMK0vqv5Yj/AYvrcP3g338Tbh9oTyYk1VML8=
 github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ=
 github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
 github.com/apache/arrow-go/v18 v18.4.1 h1:q/jVkBWCJOB9reDgaIZIdruLQUb1kbkvOnOFezVH1C4=

internal/contentdata/repository.go

@@ -251,7 +251,7 @@ func (r *Repository) Query(ctx context.Context, tx *connection.Tx, projectID, da
 				// GoogleSQL for BigQuery translates a NULL array into an empty array in the query result
 				v = []interface{}{}
 			}
-			cell, err := r.convertValueToCell(v)
+			cell, err := r.convertValueToCell(v, fields[idx])
 			if err != nil {
 				return nil, fmt.Errorf("failed to convert value to cell: %w", err)
 			}
@@ -307,48 +307,82 @@ func (r *Repository) queryParameterValueToGoValue(value *bigqueryv2.QueryParamet
 	return value.Value, nil
 }
 
-// zetasqlite returns []map[string]interface{} value as struct value, also returns []interface{} value as array value.
+// zetasqlite returns map[string]interface{} value as struct value, also returns []interface{} value as array value.
 // we need to convert them to specifically TableRow and TableCell type.
-func (r *Repository) convertValueToCell(value interface{}) (*internaltypes.TableCell, error) {
+// schema provides the field ordering for struct types to ensure deterministic field order.
+func (r *Repository) convertValueToCell(value interface{}, schema *bigqueryv2.TableFieldSchema) (*internaltypes.TableCell, error) {
 	if value == nil {
 		return &internaltypes.TableCell{V: nil}, nil
 	}
 	rv := reflect.ValueOf(value)
 	kind := rv.Type().Kind()
-	if kind != reflect.Slice && kind != reflect.Array {
-		v := fmt.Sprint(value)
-		return &internaltypes.TableCell{V: v, Bytes: int64(len(v))}, nil
-	}
-	elemType := rv.Type().Elem()
-	if elemType.Kind() == reflect.Map {
+	if kind == reflect.Map {
 		// value is struct type
 		var (
 			cells      []*internaltypes.TableCell
 			totalBytes int64
 		)
-		for i := 0; i < rv.Len(); i++ {
-			fieldV := rv.Index(i)
-			keys := fieldV.MapKeys()
-			if len(keys) != 1 {
-				return nil, fmt.Errorf("unexpected key number of field map value. expected 1 but got %d", len(keys))
+
+		// Build a map of field values for quick lookup
+		fieldValues := make(map[string]reflect.Value)
+		keys := rv.MapKeys()
+		for _, key := range keys {
+			fieldValues[key.Interface().(string)] = rv.MapIndex(key)
+		}
+
+		// Process fields in schema order to ensure deterministic ordering
+		// (Go map iteration order is randomized)
+		if schema != nil && schema.Fields != nil {
+			for _, fieldSchema := range schema.Fields {
+				fieldValue, exists := fieldValues[fieldSchema.Name]
+				if !exists {
+					// Field not present in data, skip it
+					continue
+				}
+				cell, err := r.convertValueToCell(fieldValue.Interface(), fieldSchema)
+				if err != nil {
+					return nil, err
+				}
+				cell.Name = fieldSchema.Name
+				totalBytes += cell.Bytes
+				cells = append(cells, cell)
 			}
-			cell, err := r.convertValueToCell(fieldV.MapIndex(keys[0]).Interface())
-			if err != nil {
-				return nil, err
+		} else {
+			// Fallback: no schema available, process in arbitrary order
+			for _, key := range keys {
+				cell, err := r.convertValueToCell(rv.MapIndex(key).Interface(), nil)
+				if err != nil {
+					return nil, err
+				}
+				cell.Name = key.Interface().(string)
+				totalBytes += cell.Bytes
+				cells = append(cells, cell)
 			}
-			cell.Name = keys[0].Interface().(string)
-			totalBytes += cell.Bytes
-			cells = append(cells, cell)
 		}
 		return &internaltypes.TableCell{V: internaltypes.TableRow{F: cells}, Bytes: totalBytes}, nil
 	}
+	if kind != reflect.Slice && kind != reflect.Array {
+		v := fmt.Sprint(value)
+		return &internaltypes.TableCell{V: v, Bytes: int64(len(v))}, nil
+	}
 	// array type
 	var (
 		cells            = []*internaltypes.TableCell{}
 		totalBytes int64 = 0
 	)
+	// For array elements, schema.Type will be the element type (e.g., STRUCT for array of structs)
+	// and schema.Fields will contain the struct fields
+	var elementSchema *bigqueryv2.TableFieldSchema
+	if schema != nil {
+		elementSchema = &bigqueryv2.TableFieldSchema{
+			Name:   schema.Name,
+			Type:   schema.Type,
+			Mode:   "NULLABLE", // Array elements can be nullable
+			Fields: schema.Fields,
+		}
+	}
 	for i := 0; i < rv.Len(); i++ {
-		cell, err := r.convertValueToCell(rv.Index(i).Interface())
+		cell, err := r.convertValueToCell(rv.Index(i).Interface(), elementSchema)
 		if err != nil {
 			return nil, err
 		}

internal/types/types.go

@@ -71,9 +71,20 @@ func (r *TableRow) Data() (map[string]interface{}, error) {
 }
 
 func (r *TableRow) AVROValue(fields []*types.AVROFieldSchema) (map[string]interface{}, error) {
+	// Create a map of field names to schemas for efficient lookup
+	// This is necessary because r.F may have fields in a different order than the schema
+	fieldSchemaMap := make(map[string]*types.AVROFieldSchema, len(fields))
+	for _, field := range fields {
+		fieldSchemaMap[field.Name] = field
+	}
+
 	rowMap := map[string]interface{}{}
-	for idx, cell := range r.F {
-		v, err := cell.AVROValue(fields[idx])
+	for _, cell := range r.F {
+		fieldSchema, ok := fieldSchemaMap[cell.Name]
+		if !ok {
+			return nil, fmt.Errorf("field %q not found in schema", cell.Name)
+		}
+		v, err := cell.AVROValue(fieldSchema)
 		if err != nil {
 			return nil, err
 		}

server/struct_field_test.go

@@ -2,6 +2,7 @@ package server_test
 
 import (
 	"context"
+	"strings"
 	"testing"
 
 	"cloud.google.com/go/bigquery"
@@ -387,3 +388,118 @@ func TestStructFieldNamesWithBackticks(t *testing.T) {
 		}
 	})
 }
+
+// TestEmptyStructInsertion tests that loading JSON data with empty STRUCT values
+// works correctly when IgnoreUnknownValues is set to true. This corresponds to
+// the Python BigQuery client test that loads {"conductor": {}} into a table
+// with a STRUCT field containing nested fields.
+func TestEmptyStructInsertion(t *testing.T) {
+	ctx := context.Background()
+
+	bqServer, err := server.New(server.TempStorage)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	const (
+		projectID = "test"
+		datasetID = "ds1"
+		tableID   = "t1"
+	)
+
+	// Create a table with a STRUCT field "conductor" containing a FLOAT64 field "length"
+	project := types.NewProject(
+		projectID,
+		types.NewDataset(
+			datasetID,
+			types.NewTable(
+				tableID,
+				[]*types.Column{
+					types.NewColumn(
+						"conductor",
+						types.STRUCT,
+						types.ColumnFields(
+							types.NewColumn("length", types.FLOAT64),
+						),
+					),
+				},
+				nil,
+			),
+		),
+	)
+
+	if err := bqServer.Load(server.StructSource(project)); err != nil {
+		t.Fatal(err)
+	}
+
+	testServer := bqServer.TestServer()
+	defer func() {
+		testServer.Close()
+		bqServer.Close()
+	}()
+
+	client, err := bigquery.NewClient(
+		ctx,
+		projectID,
+		option.WithEndpoint(testServer.URL),
+		option.WithoutAuthentication(),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	// Test loading JSON with an empty struct {"conductor": {}}
+	t.Run("load empty struct", func(t *testing.T) {
+		jsonData := `{"conductor": {}}`
+
+		source := bigquery.NewReaderSource(strings.NewReader(jsonData))
+		source.SourceFormat = bigquery.JSON
+
+		loader := client.Dataset(datasetID).Table(tableID).LoaderFrom(source)
+
+		job, err := loader.Run(ctx)
+		if err != nil {
+			t.Fatalf("failed to run load job: %v", err)
+		}
+
+		status, err := job.Wait(ctx)
+		if err != nil {
+			t.Fatalf("failed to wait for load job: %v", err)
+		}
+
+		if err := status.Err(); err != nil {
+			t.Fatalf("load job failed: %v", err)
+		}
+
+		t.Log("Successfully loaded empty struct")
+	})
+
+	// Test querying the loaded data
+	t.Run("query conductor.length from loaded data", func(t *testing.T) {
+		query := client.Query("SELECT conductor.length FROM `" + datasetID + "." + tableID + "`")
+
+		it, err := query.Read(ctx)
+		if err != nil {
+			t.Fatalf("failed to execute query: %v", err)
+		}
+
+		var row struct {
+			Length bigquery.NullFloat64 `bigquery:"length"`
+		}
+
+		if err := it.Next(&row); err != nil {
+			if err == iterator.Done {
+				t.Fatal("expected at least one row")
+			}
+			t.Fatalf("failed to read row: %v", err)
+		}
+
+		// The empty struct should result in NULL for the length field
+		if row.Length.Valid {
+			t.Errorf("expected NULL for conductor.length, got %v", row.Length.Float64)
+		}
+
+		t.Logf("Query succeeded: conductor.length is NULL as expected for empty struct")
+	})
+}