From e33b75272f56f9b2bc56cb498b2c498bd27cb2ee Mon Sep 17 00:00:00 2001 From: erezrokah Date: Fri, 2 Oct 2026 16:43:02 +0300 Subject: [PATCH] feat: Add shared synthetic-record helpers for destination assessment --- schema/synthetic.go | 280 +++++++++++++++++++++++++++++++++++++++ schema/synthetic_test.go | 200 ++++++++++++++++++++++++++++ 2 files changed, 480 insertions(+) create mode 100644 schema/synthetic.go create mode 100644 schema/synthetic_test.go diff --git a/schema/synthetic.go b/schema/synthetic.go new file mode 100644 index 0000000000..a78ad58245 --- /dev/null +++ b/schema/synthetic.go @@ -0,0 +1,280 @@ +package schema + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "reflect" + "strings" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/cloudquery/plugin-sdk/v4/types" +) + +var ErrUnableToCompare = errors.New("unable to compare") + +type SyntheticCase string + +const ( + SyntheticCasePopulated SyntheticCase = "populated" + SyntheticCaseNull SyntheticCase = "null" + SyntheticCaseEmptyCollection SyntheticCase = "empty collection" + SyntheticCaseNested SyntheticCase = "nested" + SyntheticCaseEscaping SyntheticCase = "escaping" +) + +var syntheticCases = []SyntheticCase{ + SyntheticCasePopulated, + SyntheticCaseNull, + SyntheticCaseEmptyCollection, + SyntheticCaseNested, + SyntheticCaseEscaping, +} + +const syntheticEscapingString = `"comma, \"quote\", back\\slash\nnew line\ttab é"` + +// SyntheticPair is a value that means the same under the old and the new field. +// Value is its Arrow JSON form, which both fields accept and read back as equal values. +type SyntheticPair struct { + Case SyntheticCase + Value string +} + +// SyntheticPairs returns equivalent values for each case both fields support. +// It returns ErrUnableToCompare when no populated value means the same under both fields. +func SyntheticPairs(oldField, newField arrow.Field) ([]SyntheticPair, error) { + var pairs []SyntheticPair + for _, c := range syntheticCases { + if c == SyntheticCaseNull { + if oldField.Nullable && newField.Nullable { + pairs = append(pairs, SyntheticPair{Case: c, Value: "null"}) + } + continue + } + candidates := append(syntheticCandidates(oldField.Type, c), syntheticCandidates(newField.Type, c)...) + for _, candidate := range candidates { + if isEquivalent(oldField.Type, newField.Type, candidate) { + pairs = append(pairs, SyntheticPair{Case: c, Value: candidate}) + break + } + } + } + if len(pairs) == 0 || pairs[0].Case != SyntheticCasePopulated { + return nil, fmt.Errorf("%w: no equivalent value for %s and %s", ErrUnableToCompare, oldField.Type, newField.Type) + } + return pairs, nil +} + +// SyntheticRecords builds one-column records for the old and the new field, with one row per pair. +func SyntheticRecords(oldField, newField arrow.Field, pairs []SyntheticPair) (oldRecord, newRecord arrow.RecordBatch, err error) { + values := make([]string, len(pairs)) + for i, pair := range pairs { + values[i] = pair.Value + } + rows := "[" + strings.Join(values, ",") + "]" + oldRecord, err = syntheticRecord(oldField, rows) + if err != nil { + return nil, nil, err + } + newRecord, err = syntheticRecord(newField, rows) + if err != nil { + oldRecord.Release() + return nil, nil, err + } + return oldRecord, newRecord, nil +} + +func syntheticRecord(field arrow.Field, rows string) (arrow.RecordBatch, error) { + bldr := array.NewRecordBuilder(memory.DefaultAllocator, arrow.NewSchema([]arrow.Field{field}, nil)) + defer bldr.Release() + if err := bldr.Field(0).UnmarshalJSON([]byte(rows)); err != nil { + return nil, fmt.Errorf("failed to build synthetic values for field %s: %w", field.Name, err) + } + return bldr.NewRecordBatch(), nil +} + +func isEquivalent(oldType, newType arrow.DataType, value string) bool { + oldValue, ok := readBack(oldType, value) + if !ok { + return false + } + newValue, ok := readBack(newType, value) + return ok && reflect.DeepEqual(oldValue, newValue) +} + +func readBack(dataType arrow.DataType, value string) (any, bool) { + bldr := array.NewBuilder(memory.DefaultAllocator, dataType) + defer bldr.Release() + if err := bldr.UnmarshalJSON([]byte("[" + value + "]")); err != nil || bldr.Len() != 1 { + return nil, false + } + arr := bldr.NewArray() + defer arr.Release() + marshaled, err := json.Marshal(arr.GetOneForMarshal(0)) + if err != nil { + return nil, false + } + var decoded any + if err := json.NewDecoder(bytes.NewReader(marshaled)).Decode(&decoded); err != nil { + return nil, false + } + return decoded, true +} + +func syntheticCandidates(dataType arrow.DataType, c SyntheticCase) []string { + switch dt := dataType.(type) { + case *arrow.ListType: + return syntheticListCandidates(dt.Elem(), c) + case *arrow.LargeListType: + return syntheticListCandidates(dt.Elem(), c) + case *arrow.MapType: + if c == SyntheticCaseNested && !isCollection(dt.ItemType()) { + return nil + } + return syntheticListCandidates(dt.Elem(), c) + case *arrow.StructType: + return syntheticStructCandidates(dt, c) + case *types.JSONType: + return syntheticJSONCandidates(c) + } + if c == SyntheticCasePopulated { + return syntheticScalarCandidates(dataType) + } + if c == SyntheticCaseEscaping && isStringLike(dataType) { + return []string{syntheticEscapingString} + } + return nil +} + +func syntheticListCandidates(elem arrow.DataType, c SyntheticCase) []string { + switch c { + case SyntheticCaseEmptyCollection: + return []string{"[]"} + case SyntheticCaseNested: + if !isCollection(elem) { + return nil + } + return wrapInList(syntheticCandidates(elem, SyntheticCasePopulated)) + default: + return wrapInList(syntheticCandidates(elem, c)) + } +} + +func wrapInList(values []string) []string { + wrapped := make([]string, len(values)) + for i, v := range values { + wrapped[i] = "[" + v + "]" + } + return wrapped +} + +func syntheticStructCandidates(dt *arrow.StructType, c SyntheticCase) []string { + if c == SyntheticCaseEmptyCollection { + return nil + } + if c == SyntheticCaseNested && !hasCollectionField(dt) { + return nil + } + hasEscaping := false + fields := make([]string, dt.NumFields()) + for i, field := range dt.Fields() { + value := firstCandidate(field.Type, SyntheticCasePopulated) + if c == SyntheticCaseEscaping { + if escaping := firstCandidate(field.Type, SyntheticCaseEscaping); escaping != "" { + value = escaping + hasEscaping = true + } + } + if value == "" { + return nil + } + name, _ := json.Marshal(field.Name) + fields[i] = string(name) + ":" + value + } + if c == SyntheticCaseEscaping && !hasEscaping { + return nil + } + return []string{"{" + strings.Join(fields, ",") + "}"} +} + +func firstCandidate(dataType arrow.DataType, c SyntheticCase) string { + candidates := syntheticCandidates(dataType, c) + if len(candidates) == 0 { + return "" + } + return candidates[0] +} + +func hasCollectionField(dt *arrow.StructType) bool { + for _, field := range dt.Fields() { + if isCollection(field.Type) { + return true + } + } + return false +} + +func isCollection(dataType arrow.DataType) bool { + switch dataType.(type) { + case *arrow.ListType, *arrow.LargeListType, *arrow.MapType, *arrow.StructType, *types.JSONType: + return true + } + return false +} + +func syntheticJSONCandidates(c SyntheticCase) []string { + switch c { + case SyntheticCasePopulated: + return []string{`{"env":"prod"}`} + case SyntheticCaseEmptyCollection: + return []string{`{}`, `[]`} + case SyntheticCaseNested: + return []string{`{"env":{"tags":["prod"]}}`} + case SyntheticCaseEscaping: + return []string{`{"note":` + syntheticEscapingString + `}`} + } + return nil +} + +func isStringLike(dataType arrow.DataType) bool { + return arrow.TypeEqual(dataType, arrow.BinaryTypes.String) || arrow.TypeEqual(dataType, arrow.BinaryTypes.LargeString) +} + +func syntheticScalarCandidates(dataType arrow.DataType) []string { + switch { + case isStringLike(dataType): + return []string{`"env:prod"`} + case arrow.TypeEqual(dataType, types.ExtensionTypes.UUID): + return []string{`"6ba7b810-9dad-11d1-80b4-00c04fd430c8"`} + case arrow.TypeEqual(dataType, types.ExtensionTypes.Inet): + return []string{`"192.0.2.1/24"`} + case arrow.TypeEqual(dataType, types.ExtensionTypes.MAC): + return []string{`"00:00:5e:00:53:01"`} + case arrow.IsInteger(dataType.ID()), arrow.IsFloating(dataType.ID()), dataType.ID() == arrow.DURATION: + return []string{`42`} + case arrow.IsDecimal(dataType.ID()): + return []string{`"12.5"`} + } + switch dataType.ID() { + case arrow.BOOL: + return []string{`true`} + case arrow.BINARY, arrow.LARGE_BINARY: + return []string{`"ZW52OnByb2Q="`} + case arrow.TIMESTAMP: + return []string{`"2024-01-02T03:04:05Z"`} + case arrow.DATE32, arrow.DATE64: + return []string{`"2024-01-02"`} + case arrow.TIME32, arrow.TIME64: + return []string{`"03:04:05"`} + case arrow.INTERVAL_MONTHS: + return []string{`{"months":1}`} + case arrow.INTERVAL_DAY_TIME: + return []string{`{"days":1,"milliseconds":1}`} + case arrow.INTERVAL_MONTH_DAY_NANO: + return []string{`{"months":1,"days":1,"nanoseconds":1}`} + } + return nil +} diff --git a/schema/synthetic_test.go b/schema/synthetic_test.go new file mode 100644 index 0000000000..a293e0f8df --- /dev/null +++ b/schema/synthetic_test.go @@ -0,0 +1,200 @@ +package schema + +import ( + "encoding/json" + "testing" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/cloudquery/plugin-sdk/v4/types" + "github.com/stretchr/testify/require" +) + +func nullableField(dataType arrow.DataType) arrow.Field { + return arrow.Field{Name: "col", Type: dataType, Nullable: true} +} + +func TestSyntheticPairs(t *testing.T) { + stringList := arrow.ListOf(arrow.BinaryTypes.String) + tests := []struct { + name string + oldField arrow.Field + newField arrow.Field + want []SyntheticPair + }{ + { + name: "list of strings to json", + oldField: nullableField(stringList), + newField: nullableField(types.ExtensionTypes.JSON), + want: []SyntheticPair{ + {Case: SyntheticCasePopulated, Value: `["env:prod"]`}, + {Case: SyntheticCaseNull, Value: "null"}, + {Case: SyntheticCaseEmptyCollection, Value: `[]`}, + {Case: SyntheticCaseEscaping, Value: `[` + syntheticEscapingString + `]`}, + }, + }, + { + name: "json to list of strings", + oldField: nullableField(types.ExtensionTypes.JSON), + newField: nullableField(stringList), + want: []SyntheticPair{ + {Case: SyntheticCasePopulated, Value: `["env:prod"]`}, + {Case: SyntheticCaseNull, Value: "null"}, + {Case: SyntheticCaseEmptyCollection, Value: `[]`}, + {Case: SyntheticCaseEscaping, Value: `[` + syntheticEscapingString + `]`}, + }, + }, + { + name: "nested list to json", + oldField: nullableField(arrow.ListOf(stringList)), + newField: nullableField(types.ExtensionTypes.JSON), + want: []SyntheticPair{ + {Case: SyntheticCasePopulated, Value: `[["env:prod"]]`}, + {Case: SyntheticCaseNull, Value: "null"}, + {Case: SyntheticCaseEmptyCollection, Value: `[]`}, + {Case: SyntheticCaseNested, Value: `[["env:prod"]]`}, + {Case: SyntheticCaseEscaping, Value: `[[` + syntheticEscapingString + `]]`}, + }, + }, + { + name: "struct to json", + oldField: nullableField(arrow.StructOf(arrow.Field{Name: "tags", Type: stringList, Nullable: true})), + newField: nullableField(types.ExtensionTypes.JSON), + want: []SyntheticPair{ + {Case: SyntheticCasePopulated, Value: `{"tags":["env:prod"]}`}, + {Case: SyntheticCaseNull, Value: "null"}, + {Case: SyntheticCaseNested, Value: `{"tags":["env:prod"]}`}, + {Case: SyntheticCaseEscaping, Value: `{"tags":[` + syntheticEscapingString + `]}`}, + }, + }, + { + name: "string to uuid", + oldField: nullableField(arrow.BinaryTypes.String), + newField: nullableField(types.ExtensionTypes.UUID), + want: []SyntheticPair{ + {Case: SyntheticCasePopulated, Value: `"6ba7b810-9dad-11d1-80b4-00c04fd430c8"`}, + {Case: SyntheticCaseNull, Value: "null"}, + }, + }, + { + name: "int32 to int64", + oldField: nullableField(arrow.PrimitiveTypes.Int32), + newField: nullableField(arrow.PrimitiveTypes.Int64), + want: []SyntheticPair{ + {Case: SyntheticCasePopulated, Value: `42`}, + {Case: SyntheticCaseNull, Value: "null"}, + }, + }, + { + name: "timestamp precision change", + oldField: nullableField(arrow.FixedWidthTypes.Timestamp_s), + newField: nullableField(arrow.FixedWidthTypes.Timestamp_us), + want: []SyntheticPair{ + {Case: SyntheticCasePopulated, Value: `"2024-01-02T03:04:05Z"`}, + {Case: SyntheticCaseNull, Value: "null"}, + }, + }, + { + name: "map to map with escaping", + oldField: nullableField(arrow.MapOf(arrow.BinaryTypes.String, arrow.BinaryTypes.String)), + newField: nullableField(arrow.MapOf(arrow.BinaryTypes.String, arrow.BinaryTypes.LargeString)), + want: []SyntheticPair{ + {Case: SyntheticCasePopulated, Value: `[{"key":"env:prod","value":"env:prod"}]`}, + {Case: SyntheticCaseNull, Value: "null"}, + {Case: SyntheticCaseEmptyCollection, Value: `[]`}, + {Case: SyntheticCaseEscaping, Value: `[{"key":` + syntheticEscapingString + `,"value":` + syntheticEscapingString + `}]`}, + }, + }, + { + name: "map of lists is nested", + oldField: nullableField(arrow.MapOf(arrow.BinaryTypes.String, stringList)), + newField: nullableField(arrow.MapOf(arrow.BinaryTypes.String, stringList)), + want: []SyntheticPair{ + {Case: SyntheticCasePopulated, Value: `[{"key":"env:prod","value":["env:prod"]}]`}, + {Case: SyntheticCaseNull, Value: "null"}, + {Case: SyntheticCaseEmptyCollection, Value: `[]`}, + {Case: SyntheticCaseNested, Value: `[{"key":"env:prod","value":["env:prod"]}]`}, + {Case: SyntheticCaseEscaping, Value: `[{"key":` + syntheticEscapingString + `,"value":[` + syntheticEscapingString + `]}]`}, + }, + }, + { + name: "not null field has no null pair", + oldField: arrow.Field{Name: "col", Type: arrow.BinaryTypes.String}, + newField: nullableField(arrow.BinaryTypes.String), + want: []SyntheticPair{ + {Case: SyntheticCasePopulated, Value: `"env:prod"`}, + {Case: SyntheticCaseEscaping, Value: syntheticEscapingString}, + }, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got, err := SyntheticPairs(tc.oldField, tc.newField) + require.NoError(t, err) + require.Equal(t, tc.want, got) + }) + } +} + +func TestSyntheticPairs_UnableToCompare(t *testing.T) { + tests := []struct { + name string + oldField arrow.Field + newField arrow.Field + }{ + {name: "int to string", oldField: nullableField(arrow.PrimitiveTypes.Int64), newField: nullableField(arrow.BinaryTypes.String)}, + {name: "string to int", oldField: nullableField(arrow.BinaryTypes.String), newField: nullableField(arrow.PrimitiveTypes.Int64)}, + {name: "duration unit change", oldField: nullableField(arrow.FixedWidthTypes.Duration_s), newField: nullableField(arrow.FixedWidthTypes.Duration_ms)}, + {name: "string to list", oldField: nullableField(arrow.BinaryTypes.String), newField: nullableField(arrow.ListOf(arrow.BinaryTypes.String))}, + {name: "bool to int", oldField: nullableField(arrow.FixedWidthTypes.Boolean), newField: nullableField(arrow.PrimitiveTypes.Int64)}, + {name: "fixed size list", oldField: nullableField(arrow.FixedSizeListOf(2, arrow.BinaryTypes.String)), newField: nullableField(arrow.FixedSizeListOf(2, arrow.BinaryTypes.String))}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + _, err := SyntheticPairs(tc.oldField, tc.newField) + require.ErrorIs(t, err, ErrUnableToCompare) + }) + } +} + +func TestSyntheticRecords(t *testing.T) { + oldField := arrow.Field{Name: "tags", Type: arrow.ListOf(arrow.BinaryTypes.String), Nullable: true} + newField := arrow.Field{Name: "tags", Type: types.ExtensionTypes.JSON, Nullable: true} + pairs, err := SyntheticPairs(oldField, newField) + require.NoError(t, err) + + oldRecord, newRecord, err := SyntheticRecords(oldField, newField, pairs) + require.NoError(t, err) + defer oldRecord.Release() + defer newRecord.Release() + + require.True(t, oldRecord.Schema().Equal(arrow.NewSchema([]arrow.Field{oldField}, nil))) + require.True(t, newRecord.Schema().Equal(arrow.NewSchema([]arrow.Field{newField}, nil))) + require.EqualValues(t, len(pairs), oldRecord.NumRows()) + require.EqualValues(t, len(pairs), newRecord.NumRows()) + + for i, pair := range pairs { + oldValue, err := json.Marshal(oldRecord.Column(0).GetOneForMarshal(i)) + require.NoError(t, err) + newValue, err := json.Marshal(newRecord.Column(0).GetOneForMarshal(i)) + require.NoError(t, err) + require.JSONEq(t, pair.Value, string(oldValue), "old value for case %s", pair.Case) + require.JSONEq(t, pair.Value, string(newValue), "new value for case %s", pair.Case) + } + populated, err := json.Marshal(newRecord.Column(0).GetOneForMarshal(0)) + require.NoError(t, err) + require.Equal(t, `["env:prod"]`, string(populated)) +} + +func TestSyntheticPairs_SameTypeIsComparable(t *testing.T) { + for _, column := range TestTable("test", TestSourceOptions{}).Columns { + t.Run(column.Name, func(t *testing.T) { + field := column.ToArrowField() + pairs, err := SyntheticPairs(field, field) + require.NoError(t, err) + oldRecord, newRecord, err := SyntheticRecords(field, field, pairs) + require.NoError(t, err) + oldRecord.Release() + newRecord.Release() + }) + } +}