This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new 3d25c98f perf(parquet/compress): reuse gzip writers (#1238)
3d25c98f is described below
commit 3d25c98fd2de516ae5c092fd904852e093c2f605
Author: Minh Vu <[email protected]>
AuthorDate: Fri Aug 28 17:04:46 2026 +0200
perf(parquet/compress): reuse gzip writers (#1238)
## Summary
- Pool gzip writers by compression level.
- Reset a writer before returning it to the pool.
- Avoid retaining the destination page buffer in the cached writer.
- Add a concurrent correctness test and page-size benchmarks.
## Benchmark
Apple M1 Pro. The output buffer is reused. Median of 3 runs.
| Input | upstream main | this PR | change |
| --- | ---: | ---: | ---: |
| Repeated, 64 KiB | 115 us, 1.08 MiB, 15 allocs | 25.7 us, 48 B, 1
alloc | 4.5x faster |
| Random, 64 KiB | 94.7 us, 1.08 MiB, 15 allocs | 15.6 us, 117 B, 1
alloc | 6.1x faster |
```text
go test ./parquet/compress -run '^$' -bench '^BenchmarkGzipEncodePages$'
-benchmem -benchtime=200ms -count=3
```
## Tests
- `PARQUET_TEST_DATA=/path/to/parquet-testing/data go test ./parquet/...
-count=1`
- `PARQUET_TEST_DATA=/path/to/parquet-testing/data go test -race
./parquet/compress ./parquet/file ./parquet/pqarrow -count=1`
- `go vet ./parquet/compress ./parquet/file`
---
parquet/compress/compress_test.go | 37 ++++++++++++++++++++
parquet/compress/gzip.go | 37 ++++++++++++++++++--
parquet/compress/gzip_benchmark_test.go | 61 +++++++++++++++++++++++++++++++++
3 files changed, 133 insertions(+), 2 deletions(-)
diff --git a/parquet/compress/compress_test.go
b/parquet/compress/compress_test.go
index 6483aa65..9f6e38d3 100644
--- a/parquet/compress/compress_test.go
+++ b/parquet/compress/compress_test.go
@@ -200,6 +200,43 @@ func TestGzipCompressBound(t *testing.T) {
}
}
+func TestGzipEncodeLevelConcurrent(t *testing.T) {
+ codec, err := compress.GetCodec(compress.Codecs.Gzip)
+ assert.NoError(t, err)
+
+ src := makeCompressibleData(4 * 1024)
+ levels := []int{
+ gzip.DefaultCompression,
+ gzip.NoCompression,
+ gzip.BestSpeed,
+ gzip.BestCompression,
+ gzip.StatelessCompression,
+ }
+
+ const workers = 16
+ var wg sync.WaitGroup
+ wg.Add(workers)
+ for i := 0; i < workers; i++ {
+ go func(i int) {
+ defer wg.Done()
+ for j := 0; j < len(levels); j++ {
+ level := levels[(i+j)%len(levels)]
+ compressed := codec.EncodeLevel(nil, src, level)
+ decoded, err := compress.Decode(codec, nil,
compressed)
+ if err != nil {
+ t.Errorf("level %d: decode failed: %v",
level, err)
+ return
+ }
+ if !bytes.Equal(src, decoded) {
+ t.Errorf("level %d: decoded data
differs", level)
+ return
+ }
+ }
+ }(i)
+ }
+ wg.Wait()
+}
+
func TestBrotliCompressBound(t *testing.T) {
codec, err := compress.GetCodec(compress.Codecs.Brotli)
assert.NoError(t, err)
diff --git a/parquet/compress/gzip.go b/parquet/compress/gzip.go
index bb6a63e3..13e917f2 100644
--- a/parquet/compress/gzip.go
+++ b/parquet/compress/gzip.go
@@ -20,12 +20,15 @@ import (
"bytes"
"fmt"
"io"
+ "sync"
"github.com/klauspost/compress/gzip"
)
type gzipCodec struct{}
+var gzipWriterPools [gzip.BestCompression - gzip.StatelessCompression +
1]sync.Pool
+
const (
gzipHeaderSize = 10
gzipTrailerSize = 8
@@ -84,18 +87,48 @@ func (g gzipCodec) EncodeLevel(dst, src []byte, level int)
[]byte {
dst = make([]byte, 0, maxlen)
}
buf := bytes.NewBuffer(dst[:0])
- w, err := gzip.NewWriterLevel(buf, level)
+ pool := gzipWriterPool(level)
+ var w *gzip.Writer
+ if pool != nil {
+ if cached := pool.Get(); cached != nil {
+ w = cached.(*gzip.Writer)
+ w.Reset(buf)
+ }
+ }
+ var err error
+ if w == nil {
+ w, err = gzip.NewWriterLevel(buf, level)
+ }
if err != nil {
panic(err)
}
_, err = w.Write(src)
if err != nil {
+ releaseGzipWriter(pool, w)
panic(err)
}
if err := w.Close(); err != nil {
+ releaseGzipWriter(pool, w)
panic(err)
}
- return buf.Bytes()
+ compressed := buf.Bytes()
+ releaseGzipWriter(pool, w)
+ return compressed
+}
+
+func gzipWriterPool(level int) *sync.Pool {
+ if level < gzip.StatelessCompression || level > gzip.BestCompression {
+ return nil
+ }
+ return &gzipWriterPools[level-gzip.StatelessCompression]
+}
+
+func releaseGzipWriter(pool *sync.Pool, w *gzip.Writer) {
+ if pool == nil {
+ return
+ }
+ w.Reset(nil)
+ pool.Put(w)
}
func (g gzipCodec) Encode(dst, src []byte) []byte {
diff --git a/parquet/compress/gzip_benchmark_test.go
b/parquet/compress/gzip_benchmark_test.go
new file mode 100644
index 00000000..8aa817b7
--- /dev/null
+++ b/parquet/compress/gzip_benchmark_test.go
@@ -0,0 +1,61 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package compress_test
+
+import (
+ "fmt"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/parquet/compress"
+ "github.com/klauspost/compress/gzip"
+)
+
+func BenchmarkGzipEncodePages(b *testing.B) {
+ codec, err := compress.GetCodec(compress.Codecs.Gzip)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ for _, tc := range []struct {
+ name string
+ data func(int) []byte
+ }{
+ {"repeated", makeCompressibleData},
+ {"random", makeRandomData},
+ } {
+ for _, pageSize := range []int{4 << 10, 64 << 10, 256 << 10} {
+ b.Run(tc.name+"/page="+formatBytes(pageSize), func(b
*testing.B) {
+ src := tc.data(pageSize)
+ dst := make([]byte, 0,
codec.CompressBound(int64(len(src))))
+
+ b.SetBytes(int64(len(src)))
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ dst = codec.EncodeLevel(dst[:0], src,
gzip.DefaultCompression)
+ }
+ })
+ }
+ }
+}
+
+func formatBytes(size int) string {
+ if size >= 1<<20 {
+ return fmt.Sprintf("%dMiB", size>>20)
+ }
+ return fmt.Sprintf("%dKiB", size>>10)
+}
