From 820e9408f91eb054e9e86415640567c9e51386ab Mon Sep 17 00:00:00 2001 From: erezrokah Date: Fri, 2 Oct 2026 16:46:47 +0300 Subject: [PATCH] feat: Compare generated Parquet schemas between table versions --- go.mod | 14 ++-- go.sum | 28 +++---- parquet/assess.go | 98 +++++++++++++++++++++++ parquet/assess_test.go | 172 +++++++++++++++++++++++++++++++++++++++++ parquet/write.go | 10 ++- 5 files changed, 297 insertions(+), 25 deletions(-) create mode 100644 parquet/assess.go create mode 100644 parquet/assess_test.go diff --git a/go.mod b/go.mod index 3f285386..8eb46ac4 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/apache/arrow-go/v18 v18.8.0 github.com/bradleyjkemp/cupaloy/v2 v2.8.0 github.com/cloudquery/codegen v0.4.1 - github.com/cloudquery/plugin-sdk/v4 v4.96.6 + github.com/cloudquery/plugin-sdk/v4 v4.96.7-0.20261002134302-e33b75272f56 github.com/goccy/go-json v0.10.6 github.com/invopop/jsonschema v0.14.0 github.com/pb33f/ordered-map/v2 v2.3.1 @@ -40,12 +40,12 @@ require ( github.com/zeebo/xxh3 v1.1.0 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect go.yaml.in/yaml/v4 v4.0.0-rc.2 // indirect - golang.org/x/exp v0.0.0-20260824195058-e88cd73687aa // indirect - golang.org/x/net v0.58.0 // indirect - golang.org/x/sync v0.22.0 // indirect - golang.org/x/sys v0.47.0 // indirect - golang.org/x/text v0.41.0 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d // indirect + golang.org/x/exp v0.0.0-20260908205506-85c1c2202aba // indirect + golang.org/x/net v0.59.0 // indirect + golang.org/x/sync v0.23.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/text v0.42.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260819154853-08b0e4226688 // indirect google.golang.org/grpc v1.84.0 // indirect google.golang.org/protobuf v1.36.12 // indirect ) diff --git a/go.sum b/go.sum index 3c8fb7a9..6d6af7ba 100644 --- a/go.sum +++ b/go.sum @@ -22,8 +22,8 @@ github.com/cloudquery/codegen v0.4.1 h1:c9D18N925tUvnDeGHIl3JWKj37TyII9daHufkf8h github.com/cloudquery/codegen v0.4.1/go.mod h1:QWIOD6R1aCa+YM+th+9Qt9lZw+ztdJR9JDEMLWyazwM= github.com/cloudquery/jsonschema v0.0.0-20260703174721-45e7e20e0ed8 h1:s7B+c57yTtVL8Zhcebae5poFInJwTftuMakj625CRbw= github.com/cloudquery/jsonschema v0.0.0-20260703174721-45e7e20e0ed8/go.mod h1:KMcD1TlufeD5r5DmbYvYWs+cyULzIVuKkd7jpoKWVjk= -github.com/cloudquery/plugin-sdk/v4 v4.96.6 h1:dGSsyUlLYIjUOV+btlEPF6J4u6aCKqvQuDyonmEVgww= -github.com/cloudquery/plugin-sdk/v4 v4.96.6/go.mod h1:YxiA5AIDIA/RK0rGeBlfUCcabzYzMvThx+c882pB03A= +github.com/cloudquery/plugin-sdk/v4 v4.96.7-0.20261002134302-e33b75272f56 h1:cDJbRcSiJmGFF7HHAeaTKOI4FTNK55+1skEO1SKlGDs= +github.com/cloudquery/plugin-sdk/v4 v4.96.7-0.20261002134302-e33b75272f56/go.mod h1:v0eP8ku/QWkCU5NqvwulISYyJqV7ORJRh6eeNDMlm0g= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= @@ -96,20 +96,20 @@ go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= go.yaml.in/yaml/v4 v4.0.0-rc.2 h1:/FrI8D64VSr4HtGIlUtlFMGsm7H7pWTbj6vOLVZcA6s= go.yaml.in/yaml/v4 v4.0.0-rc.2/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0= -golang.org/x/exp v0.0.0-20260824195058-e88cd73687aa h1:QSyA8ishJCyT21kER9KwNt0b7BM3iRK4x9QXhjN5Fdk= -golang.org/x/exp v0.0.0-20260824195058-e88cd73687aa/go.mod h1:zeBbvyFKDaLwa7CH/zI8KXt7gTl14SF7sO08Pl5jBCM= -golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= -golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= -golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= -golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= -golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= -golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= -golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= +golang.org/x/exp v0.0.0-20260908205506-85c1c2202aba h1:Ck8QetSgk912qxWLMCKxd0in+aiyBQyDSMae6e/xmpU= +golang.org/x/exp v0.0.0-20260908205506-85c1c2202aba/go.mod h1:50RgIsmK7OwqzTTeqcSXQW8SswW0o8fRcDxmqGluJ8E= +golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues= +golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d h1:IL4hdHzcUv2l/gcg98/Rj3FbtE6axwqslOW8SW0C+S0= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260819154853-08b0e4226688 h1:cYNAzI2sUwhmCcoj9TxvihSrqsxt6uIkj3rDRhSDmW4= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260819154853-08b0e4226688/go.mod h1:DjtHYE8FKJLivXcBEjGwndXfIC23G0VpXiXKqG179uA= google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0= google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2QN2C0= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= diff --git a/parquet/assess.go b/parquet/assess.go new file mode 100644 index 00000000..efcf8880 --- /dev/null +++ b/parquet/assess.go @@ -0,0 +1,98 @@ +package parquet + +import ( + "strconv" + "strings" + + "github.com/apache/arrow-go/v18/parquet" + "github.com/apache/arrow-go/v18/parquet/pqarrow" + pqschema "github.com/apache/arrow-go/v18/parquet/schema" + "github.com/cloudquery/plugin-sdk/v4/plugin" + "github.com/cloudquery/plugin-sdk/v4/schema" +) + +// ParquetSchema returns the Parquet schema WriteHeader writes for the table, without writing anything. +func (c *Client) ParquetSchema(t *schema.Table) (*pqschema.Schema, error) { + return pqarrow.ToParquet(convertSchema(t.ToArrowSchema()), c.writerProps(), pqarrow.DefaultWriterProps()) +} + +// AssessTable compares the Parquet schemas generated for the old and new table. +func (c *Client) AssessTable(pair plugin.TablePair) (plugin.TableFinding, error) { + finding := plugin.TableFinding{TableName: pair.TableName(), Category: plugin.AssessCategoryNoChange} + if pair.New == nil { + finding.Category = plugin.AssessCategoryTableRemoved + return finding, nil + } + oldNames, oldTypes, err := c.fieldTypes(pair.Old) + if err != nil { + return plugin.TableFinding{}, err + } + newNames, newTypes, err := c.fieldTypes(pair.New) + if err != nil { + return plugin.TableFinding{}, err + } + for _, name := range oldNames { + if oldTypes[name] != newTypes[name] { + finding.Columns = append(finding.Columns, fileSchemaChange(name, oldTypes[name], newTypes[name])) + } + } + for _, name := range newNames { + if _, ok := oldTypes[name]; !ok { + finding.Columns = append(finding.Columns, fileSchemaChange(name, "", newTypes[name])) + } + } + if len(finding.Columns) > 0 { + finding.Category = plugin.AssessCategoryFileSchemaChanged + } + return finding, nil +} + +func fileSchemaChange(name, oldType, newType string) plugin.ColumnFinding { + return plugin.ColumnFinding{ + ColumnName: name, + Category: plugin.AssessCategoryFileSchemaChanged, + OldType: oldType, + NewType: newType, + } +} + +func (c *Client) fieldTypes(t *schema.Table) ([]string, map[string]string, error) { + typesByName := make(map[string]string) + if t == nil { + return nil, typesByName, nil + } + sc, err := c.ParquetSchema(t) + if err != nil { + return nil, nil, err + } + root := sc.Root() + names := make([]string, root.NumFields()) + for i := range names { + field := root.Field(i) + names[i] = field.Name() + typesByName[field.Name()] = parquetType(field) + } + return names, typesByName, nil +} + +func parquetType(node pqschema.Node) string { + repetition := strings.ToLower(node.RepetitionType().String()) + logicalType := "" + if lt := node.LogicalType(); lt != nil && !lt.IsNone() { + logicalType = " (" + lt.String() + ")" + } + group, ok := node.(*pqschema.GroupNode) + if !ok { + primitive := node.(*pqschema.PrimitiveNode) + physicalType := strings.ToLower(primitive.PhysicalType().String()) + if primitive.PhysicalType() == parquet.Types.FixedLenByteArray { + physicalType += "(" + strconv.Itoa(primitive.TypeLength()) + ")" + } + return repetition + " " + physicalType + logicalType + } + children := make([]string, group.NumFields()) + for i := range children { + children[i] = group.Field(i).Name() + ": " + parquetType(group.Field(i)) + } + return repetition + " group" + logicalType + " {" + strings.Join(children, "; ") + "}" +} diff --git a/parquet/assess_test.go b/parquet/assess_test.go new file mode 100644 index 00000000..8ba5e6eb --- /dev/null +++ b/parquet/assess_test.go @@ -0,0 +1,172 @@ +package parquet + +import ( + "bytes" + "testing" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/parquet/file" + "github.com/cloudquery/plugin-sdk/v4/plugin" + "github.com/cloudquery/plugin-sdk/v4/schema" + "github.com/cloudquery/plugin-sdk/v4/types" + "github.com/stretchr/testify/require" +) + +const ( + parquetString = "optional byte_array (String)" + parquetJSON = parquetString + parquetBool = "optional boolean" +) + +func cloudflareCertificatePacksV11() *schema.Table { + return &schema.Table{Name: "cloudflare_certificate_packs", Columns: schema.ColumnList{ + schema.CqIDColumn, + schema.CqParentIDColumn, + {Name: "account_id", Type: arrow.BinaryTypes.String}, + {Name: "zone_id", Type: arrow.BinaryTypes.String}, + {Name: "id", Type: arrow.BinaryTypes.String, PrimaryKey: true}, + {Name: "primary_certificate", Type: arrow.BinaryTypes.String}, + {Name: "certificate_authority", Type: arrow.BinaryTypes.String}, + {Name: "type", Type: arrow.BinaryTypes.String}, + {Name: "hosts", Type: arrow.ListOf(arrow.BinaryTypes.String)}, + {Name: "status", Type: arrow.BinaryTypes.String}, + {Name: "certificates", Type: types.ExtensionTypes.JSON}, + {Name: "created_on", Type: arrow.BinaryTypes.String}, + {Name: "validity_days", Type: arrow.PrimitiveTypes.Float64}, + {Name: "validation_method", Type: arrow.BinaryTypes.String}, + }} +} + +func cloudflareCertificatePacksV12() *schema.Table { + return &schema.Table{Name: "cloudflare_certificate_packs", Columns: schema.ColumnList{ + schema.CqIDColumn, + schema.CqParentIDColumn, + {Name: "account_id", Type: arrow.BinaryTypes.String}, + {Name: "zone_id", Type: arrow.BinaryTypes.String}, + {Name: "id", Type: arrow.BinaryTypes.String, PrimaryKey: true}, + {Name: "validity_days", Type: arrow.PrimitiveTypes.Float64}, + {Name: "certificates", Type: types.ExtensionTypes.JSON}, + {Name: "hosts", Type: arrow.ListOf(arrow.BinaryTypes.String)}, + {Name: "status", Type: arrow.BinaryTypes.String}, + {Name: "type", Type: arrow.BinaryTypes.String}, + {Name: "certificate_authority", Type: arrow.BinaryTypes.String}, + {Name: "cloudflare_branding", Type: arrow.FixedWidthTypes.Boolean}, + {Name: "primary_certificate", Type: arrow.BinaryTypes.String}, + {Name: "validation_errors", Type: types.ExtensionTypes.JSON}, + {Name: "validation_method", Type: arrow.BinaryTypes.String}, + {Name: "validation_records", Type: types.ExtensionTypes.JSON}, + }} +} + +func TestParquetSchemaMatchesWrittenFile(t *testing.T) { + table := schema.TestTable("test", schema.TestSourceOptions{}) + cl, err := NewClient() + require.NoError(t, err) + + var b bytes.Buffer + h, err := cl.WriteHeader(&b, table) + require.NoError(t, err) + require.NoError(t, h.WriteFooter()) + + reader, err := file.NewParquetReader(bytes.NewReader(b.Bytes())) + require.NoError(t, err) + defer reader.Close() + + generated, err := cl.ParquetSchema(table) + require.NoError(t, err) + require.True(t, reader.MetaData().Schema.Equals(generated)) +} + +func TestAssessTable(t *testing.T) { + stringList := arrow.ListOf(arrow.BinaryTypes.String) + tagsTable := func(dataType arrow.DataType) *schema.Table { + return &schema.Table{Name: "datadog_monitors", Columns: schema.ColumnList{ + {Name: "id", Type: arrow.PrimitiveTypes.Int64, PrimaryKey: true, NotNull: true}, + {Name: "tags", Type: dataType}, + }} + } + tests := []struct { + name string + pair plugin.TablePair + want plugin.TableFinding + }{ + { + name: "cloudflare certificate packs v11.4.0 to v12.0.0", + pair: plugin.TablePair{Old: cloudflareCertificatePacksV11(), New: cloudflareCertificatePacksV12()}, + want: plugin.TableFinding{ + TableName: "cloudflare_certificate_packs", + Category: plugin.AssessCategoryFileSchemaChanged, + Columns: []plugin.ColumnFinding{ + {ColumnName: "created_on", Category: plugin.AssessCategoryFileSchemaChanged, OldType: parquetString}, + {ColumnName: "cloudflare_branding", Category: plugin.AssessCategoryFileSchemaChanged, NewType: parquetBool}, + {ColumnName: "validation_errors", Category: plugin.AssessCategoryFileSchemaChanged, NewType: parquetJSON}, + {ColumnName: "validation_records", Category: plugin.AssessCategoryFileSchemaChanged, NewType: parquetJSON}, + }, + }, + }, + { + name: "list of strings to json", + pair: plugin.TablePair{Old: tagsTable(stringList), New: tagsTable(types.ExtensionTypes.JSON)}, + want: plugin.TableFinding{ + TableName: "datadog_monitors", + Category: plugin.AssessCategoryFileSchemaChanged, + Columns: []plugin.ColumnFinding{{ + ColumnName: "tags", + Category: plugin.AssessCategoryFileSchemaChanged, + OldType: "optional group (List) {list: repeated group {element: optional byte_array (String)}}", + NewType: parquetJSON, + }}, + }, + }, + { + name: "types that write the same parquet type", + pair: plugin.TablePair{Old: tagsTable(arrow.BinaryTypes.LargeString), New: tagsTable(types.ExtensionTypes.UUID)}, + want: plugin.TableFinding{TableName: "datadog_monitors", Category: plugin.AssessCategoryNoChange}, + }, + { + name: "timestamp precision change", + pair: plugin.TablePair{Old: tagsTable(arrow.FixedWidthTypes.Timestamp_ms), New: tagsTable(arrow.FixedWidthTypes.Timestamp_us)}, + want: plugin.TableFinding{ + TableName: "datadog_monitors", + Category: plugin.AssessCategoryFileSchemaChanged, + Columns: []plugin.ColumnFinding{{ + ColumnName: "tags", + Category: plugin.AssessCategoryFileSchemaChanged, + OldType: "optional int64 (Timestamp(isAdjustedToUTC=true, timeUnit=milliseconds, is_from_converted_type=false, force_set_converted_type=true))", + NewType: "optional int64 (Timestamp(isAdjustedToUTC=true, timeUnit=microseconds, is_from_converted_type=false, force_set_converted_type=true))", + }}, + }, + }, + { + name: "same table", + pair: plugin.TablePair{Old: cloudflareCertificatePacksV12(), New: cloudflareCertificatePacksV12()}, + want: plugin.TableFinding{TableName: "cloudflare_certificate_packs", Category: plugin.AssessCategoryNoChange}, + }, + { + name: "added table", + pair: plugin.TablePair{New: tagsTable(stringList)}, + want: plugin.TableFinding{ + TableName: "datadog_monitors", + Category: plugin.AssessCategoryFileSchemaChanged, + Columns: []plugin.ColumnFinding{ + {ColumnName: "id", Category: plugin.AssessCategoryFileSchemaChanged, NewType: "required int64 (Int(bitWidth=64, isSigned=true))"}, + {ColumnName: "tags", Category: plugin.AssessCategoryFileSchemaChanged, NewType: "optional group (List) {list: repeated group {element: optional byte_array (String)}}"}, + }, + }, + }, + { + name: "removed table", + pair: plugin.TablePair{Old: tagsTable(stringList)}, + want: plugin.TableFinding{TableName: "datadog_monitors", Category: plugin.AssessCategoryTableRemoved}, + }, + } + cl, err := NewClient() + require.NoError(t, err) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got, err := cl.AssessTable(tc.pair) + require.NoError(t, err) + require.Equal(t, tc.want, got) + }) + } +} diff --git a/parquet/write.go b/parquet/write.go index 974a33e2..e6cc9330 100644 --- a/parquet/write.go +++ b/parquet/write.go @@ -21,16 +21,18 @@ type Handle struct { var _ ftypes.Handle = (*Handle)(nil) -func (c *Client) WriteHeader(w io.Writer, t *schema.Table) (ftypes.Handle, error) { - props := parquet.NewWriterProperties( +func (c *Client) writerProps() *parquet.WriterProperties { + return parquet.NewWriterProperties( parquet.WithMaxRowGroupLength(c.spec.GetMaxRowGroupLength()), parquet.WithCompression(compress.Codecs.Snappy), parquet.WithVersion(c.spec.GetVersion()), parquet.WithRootRepetition(c.spec.GetRootRepetition()), ) - arrprops := pqarrow.DefaultWriterProps() +} + +func (c *Client) WriteHeader(w io.Writer, t *schema.Table) (ftypes.Handle, error) { newSchema := convertSchema(t.ToArrowSchema()) - fw, err := pqarrow.NewFileWriter(newSchema, &nopCloseWriter{Writer: w}, props, arrprops) + fw, err := pqarrow.NewFileWriter(newSchema, &nopCloseWriter{Writer: w}, c.writerProps(), pqarrow.DefaultWriterProps()) if err != nil { return nil, err }