From f6ffdf50561501ec485674ae3154554418c3d2f7 Mon Sep 17 00:00:00 2001 From: Greg Furman Date: Sun, 24 May 2026 16:23:40 +0200 Subject: [PATCH] feat(parquet): Support nested 2D lists --- CHANGELOG.md | 6 +++ go.mod | 9 ++-- go.sum | 24 ++++++++--- internal/impl/parquet/convert.go | 2 +- .../impl/parquet/processor_encode_test.go | 5 +-- internal/impl/parquet/schema.go | 43 +++++++++++++++++-- 6 files changed, 72 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e2d672f71..7101330e5d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,12 @@ Changelog All notable changes to this project will be documented in this file. +## 1.18.0 - TBD + +### Added + +- nested list-of-list support to `parquet` processors @gregfurman + ## 1.17.0 - 2026-04-15 ### Added diff --git a/go.mod b/go.mod index 2c1eee6e23..b8c510668b 100644 --- a/go.mod +++ b/go.mod @@ -94,7 +94,7 @@ require ( github.com/jackc/pgx/v5 v5.9.0 github.com/jhump/protoreflect v1.15.6 github.com/jmespath/go-jmespath v0.4.0 - github.com/klauspost/compress v1.18.4 + github.com/klauspost/compress v1.18.6 github.com/klauspost/pgzip v1.2.6 github.com/knights-analytics/hugot v0.6.4 github.com/lib/pq v1.10.9 @@ -113,7 +113,7 @@ require ( github.com/opensearch-project/opensearch-go/v4 v4.5.0 github.com/ory/dockertest/v3 v3.10.0 github.com/oschwald/geoip2-golang v1.9.0 - github.com/parquet-go/parquet-go v0.25.1 + github.com/parquet-go/parquet-go v0.28.0 github.com/pebbe/zmq4 v1.2.11 github.com/pierrec/lz4/v4 v4.1.26 github.com/pkg/sftp v1.13.7 @@ -228,11 +228,14 @@ require ( github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect + github.com/parquet-go/bitpack v1.0.0 // indirect + github.com/parquet-go/jsonlite v1.5.2 // indirect github.com/patrickmn/go-cache v2.1.0+incompatible // indirect github.com/spf13/cast v1.7.0 // indirect github.com/spf13/pflag v1.0.6 // indirect github.com/spiffe/go-spiffe/v2 v2.6.0 // indirect github.com/theparanoids/crypki v1.20.9 // indirect + github.com/twpayne/go-geom v1.6.1 // indirect github.com/yalue/onnxruntime_go v1.26.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.43.0 // indirect @@ -286,7 +289,7 @@ require ( github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.54.0 // indirect github.com/Microsoft/go-winio v0.6.2 // indirect github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 // indirect - github.com/andybalholm/brotli v1.2.0 + github.com/andybalholm/brotli v1.2.1 github.com/apache/arrow/go/v15 v15.0.2 // indirect github.com/ardielle/ardielle-go v1.5.2 // indirect github.com/armon/go-metrics v0.3.4 // indirect diff --git a/go.sum b/go.sum index 684f263c80..d5e2ced70b 100644 --- a/go.sum +++ b/go.sum @@ -690,6 +690,8 @@ github.com/ClickHouse/ch-go v0.71.0 h1:bUdZ/EZj/LcVHsMqaRUP2holqygrPWQKeMjc6nZoy github.com/ClickHouse/ch-go v0.71.0/go.mod h1:NwbNc+7jaqfY58dmdDUbG4Jl22vThgx1cYjBw0vtgXw= github.com/ClickHouse/clickhouse-go/v2 v2.43.0 h1:fUR05TrF1GyvLDa/mAQjkx7KbgwdLRffs2n9O3WobtE= github.com/ClickHouse/clickhouse-go/v2 v2.43.0/go.mod h1:o6jf7JM/zveWC/PP277BLxjHy5KjnGX/jfljhM4s34g= +github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU= +github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU= github.com/DataDog/datadog-api-client-go/v2 v2.56.0 h1:HKcfvAODmJCUw7nfbDKKqkEUgcu7CfxUPA9EFRJrHEI= github.com/DataDog/datadog-api-client-go/v2 v2.56.0/go.mod h1:d3tOEgUd2kfsr9uuHQdY+nXrWp4uikgTgVCPdKNK30U= github.com/DataDog/datadog-go v2.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ= @@ -746,13 +748,17 @@ github.com/ajstarks/deck v0.0.0-20200831202436-30c9fc6549a9/go.mod h1:JynElWSGnm github.com/ajstarks/deck/generate v0.0.0-20210309230005-c3f852c02e19/go.mod h1:T13YZdzov6OU0A1+RfKZiZN9ca6VeKdBdyDV+BY97Tk= github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw= github.com/ajstarks/svgo v0.0.0-20211024235047-1546f124cd8b/go.mod h1:1KcenG0jGWcpt8ov532z81sp/kMMUG485J2InIOyADM= +github.com/alecthomas/assert/v2 v2.10.0 h1:jjRCHsj6hBJhkmhznrCzoNpbA3zqy0fYiUcYZP/GkPY= +github.com/alecthomas/assert/v2 v2.10.0/go.mod h1:Bze95FyfUr7x34QZrjL+XP+0qgp/zg8yS+TtBj1WA3k= +github.com/alecthomas/repr v0.4.0 h1:GhI2A8MACjfegCPVq9f1FLvIBS+DrQ2KQBFZP1iFzXc= +github.com/alecthomas/repr v0.4.0/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/andybalholm/brotli v1.0.4/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig= -github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ= -github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= +github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro= +github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY= github.com/antithesishq/antithesis-sdk-go v0.6.0-default-no-op h1:kpBdlEPbRvff0mDD1gk7o9BhI16b9p5yYAXRlidpqJE= github.com/antithesishq/antithesis-sdk-go v0.6.0-default-no-op/go.mod h1:IUpT2DPAKh6i/YhSbt6Gl3v2yvUZjmKncl7U91fup7E= @@ -1461,8 +1467,8 @@ github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= github.com/klauspost/compress v1.14.4/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU= -github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c= -github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= +github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= +github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= @@ -1642,8 +1648,12 @@ github.com/oschwald/geoip2-golang v1.9.0 h1:uvD3O6fXAXs+usU+UGExshpdP13GAqp4GBrz github.com/oschwald/geoip2-golang v1.9.0/go.mod h1:BHK6TvDyATVQhKNbQBdrj9eAvuwOMi2zSFXizL3K81Y= github.com/oschwald/maxminddb-golang v1.11.0 h1:aSXMqYR/EPNjGE8epgqwDay+P30hCBZIveY0WZbAWh0= github.com/oschwald/maxminddb-golang v1.11.0/go.mod h1:YmVI+H0zh3ySFR3w+oz8PCfglAFj3PuCmui13+P9zDg= -github.com/parquet-go/parquet-go v0.25.1 h1:l7jJwNM0xrk0cnIIptWMtnSnuxRkwq53S+Po3KG8Xgo= -github.com/parquet-go/parquet-go v0.25.1/go.mod h1:AXBuotO1XiBtcqJb/FKFyjBG4aqa3aQAAWF3ZPzCanY= +github.com/parquet-go/bitpack v1.0.0 h1:AUqzlKzPPXf2bCdjfj4sTeacrUwsT7NlcYDMUQxPcQA= +github.com/parquet-go/bitpack v1.0.0/go.mod h1:XnVk9TH+O40eOOmvpAVZ7K2ocQFrQwysLMnc6M/8lgs= +github.com/parquet-go/jsonlite v1.5.2 h1:8TZzYknFOHUpgjTLf80qbzc+8GdeT/3a3fdXSzhMylE= +github.com/parquet-go/jsonlite v1.5.2/go.mod h1:nDjpkpL4EOtqs6NQugUsi0Rleq9sW/OtC1NnZEnxzF0= +github.com/parquet-go/parquet-go v0.28.0 h1:ECyksyv8T2pOrlLsN7aWJIoQakyk/HtxQ2lchgS4els= +github.com/parquet-go/parquet-go v0.28.0/go.mod h1:navtkAYr2LGoJVp141oXPlO/sxLvaOe3la2JEoD8+rg= github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc= github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ= @@ -1849,6 +1859,8 @@ github.com/twmb/franz-go/pkg/kadm v1.16.0 h1:STMs1t5lYR5mR974PSiwNzE5TvsosByTp+r github.com/twmb/franz-go/pkg/kadm v1.16.0/go.mod h1:MUdcUtnf9ph4SFBLLA/XxE29rvLhWYLM9Ygb8dfSCvw= github.com/twmb/franz-go/pkg/kmsg v1.9.0 h1:JojYUph2TKAau6SBtErXpXGC7E3gg4vGZMv9xFU/B6M= github.com/twmb/franz-go/pkg/kmsg v1.9.0/go.mod h1:CMbfazviCyY6HM0SXuG5t9vOwYDHRCSrJJyBAe5paqg= +github.com/twpayne/go-geom v1.6.1 h1:iLE+Opv0Ihm/ABIcvQFGIiFBXd76oBIar9drAwHFhR4= +github.com/twpayne/go-geom v1.6.1/go.mod h1:Kr+Nly6BswFsKM5sd31YaoWS5PeDDH2NftJTK7Gd028= github.com/urfave/cli/v2 v2.27.7 h1:bH59vdhbjLv3LAvIu6gd0usJHgoTTPhCFib8qqOwXYU= github.com/urfave/cli/v2 v2.27.7/go.mod h1:CyNAG/xg+iAOg0N4MPGZqVmv2rCoP267496AOXUZjA4= github.com/viant/afs v1.30.0 h1:dbgVVSCPwGHUgpgkWJ5gdjKBqssT7OV7Z2M81CjwZEY= diff --git a/internal/impl/parquet/convert.go b/internal/impl/parquet/convert.go index 098ae45001..b68f914d9d 100644 --- a/internal/impl/parquet/convert.go +++ b/internal/impl/parquet/convert.go @@ -372,7 +372,7 @@ func transformList(data any) any { if slice, ok := data.([]any); ok { wrapped := make([]any, len(slice)) for i, item := range slice { - wrapped[i] = map[string]any{"element": item} + wrapped[i] = map[string]any{"element": transformList(item)} } return map[string]any{"list": wrapped} } diff --git a/internal/impl/parquet/processor_encode_test.go b/internal/impl/parquet/processor_encode_test.go index f05d94d830..874994bee7 100644 --- a/internal/impl/parquet/processor_encode_test.go +++ b/internal/impl/parquet/processor_encode_test.go @@ -502,10 +502,7 @@ use_parquet_list_format: false } func TestParquetDecodeListFormatEdgeCases(t *testing.T) { - // FIXME: Close https://github.com/warpstreamlabs/bento/issues/360 when 2D array support added to parquet-go. - t.Skip("2D slices are not currently supported by parquet encoder.") - - tctx := context.Background() + tctx := t.Context() encodeConf, err := parquetEncodeProcessorConfig().ParseYAML(` schema: diff --git a/internal/impl/parquet/schema.go b/internal/impl/parquet/schema.go index cb9a2d49a5..06115acf0c 100644 --- a/internal/impl/parquet/schema.go +++ b/internal/impl/parquet/schema.go @@ -28,6 +28,18 @@ func GenerateStructType( return generateStructTypeFromFields(fields, schemaOpts) } +type tags []struct { + name, value string +} + +func (t tags) String() string { + res := make([]string, len(t)) + for i, tg := range t { + res[i] = fmt.Sprintf(`%s:%q`, tg.name, tg.value) + } + return strings.Join(res, " ") +} + func generateStructTypeFromFields( fields []*service.ParsedConfig, schemaOpts schemaOpts, @@ -100,13 +112,23 @@ func generateStructTypeFromFields( parquetTag := strings.Join(components, ",") + tt := tags{ + {"parquet", parquetTag}, + {"json", name}, + } + + if isListOfLists(field) { + // includes a 'parquet-element' tag, to associate a list of elements + // See https://pkg.go.dev/github.com/parquet-go/parquet-go#SchemaOf + tt = append(tt, struct{ name, value string }{"parquet-element", ",list"}) + } + structField := reflect.StructField{ Name: exportedName, Type: fieldType, - Tag: reflect.StructTag(fmt.Sprintf( - `parquet:"%s" json:"%s"`, - parquetTag, name)), + Tag: reflect.StructTag(tt.String()), } + structFields = append(structFields, structField) } @@ -295,6 +317,21 @@ func wrapType( return baseType, nil } +func isListOfLists(field *service.ParsedConfig) bool { + typeStr, _ := field.FieldString("type") + if typeStr != "LIST" { + return false + } + + subfields, _ := field.FieldAnyList("fields") + if len(subfields) != 1 { + return false + } + + typeStr, _ = subfields[0].FieldString("type") + return typeStr == "LIST" +} + func isDeltaLengthByteArrayEncodable(typeStr string) bool { switch typeStr { case "BYTE_ARRAY", "UTF8":