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 edc4921b fix(parquet/compress): avoid closing caller streams for
uncompressed wrappers (#1003)
edc4921b is described below
commit edc4921bf8cd908150d6a80aa28cb3255c87430b
Author: Minh Vu <[email protected]>
AuthorDate: Mon Jul 27 22:01:22 2026 +0200
fix(parquet/compress): avoid closing caller streams for uncompressed
wrappers (#1003)
The compressed streaming codecs close their own wrappers without taking
ownership of the caller stream. The uncompressed codec passed through
existing closers directly, so closing the wrapper also closed the
underlying reader or writer.
This always wraps uncompressed streams in no-op closers and adds
regression tests for both read and write wrappers.
Tests: `go test ./parquet/compress`
---
parquet/compress/compress.go | 12 ++----------
parquet/compress/compress_test.go | 24 ++++++++++++++++++++++++
2 files changed, 26 insertions(+), 10 deletions(-)
diff --git a/parquet/compress/compress.go b/parquet/compress/compress.go
index 4743c986..835f94e4 100644
--- a/parquet/compress/compress.go
+++ b/parquet/compress/compress.go
@@ -157,11 +157,7 @@ func RegisterCodec(compression Compression, codec Codec) {
type nocodec struct{}
func (nocodec) NewReader(r io.Reader) io.ReadCloser {
- ret, ok := r.(io.ReadCloser)
- if !ok {
- return io.NopCloser(r)
- }
- return ret
+ return io.NopCloser(r)
}
func (nocodec) Decode(dst, src []byte) []byte {
@@ -194,11 +190,7 @@ func (nocodec) EncodeLevel(dst, src []byte, _ int) []byte {
}
func (nocodec) NewWriter(w io.Writer) io.WriteCloser {
- ret, ok := w.(io.WriteCloser)
- if !ok {
- return writerNopCloser{w}
- }
- return ret
+ return writerNopCloser{w}
}
func (n nocodec) NewWriterLevel(w io.Writer, _ int) (io.WriteCloser, error) {
diff --git a/parquet/compress/compress_test.go
b/parquet/compress/compress_test.go
index eedf0f5f..9410b2ed 100644
--- a/parquet/compress/compress_test.go
+++ b/parquet/compress/compress_test.go
@@ -34,6 +34,16 @@ type panickingCodec struct{ compress.Codec }
func (panickingCodec) Decode([]byte, []byte) []byte {
panic(errors.New("invalid block")) }
+type closeTrackingBuffer struct {
+ bytes.Buffer
+ closed bool
+}
+
+func (b *closeTrackingBuffer) Close() error {
+ b.closed = true
+ return nil
+}
+
const (
RandomDataSize = 3 * 1024 * 1024
CompressibleDataSize = 8 * 1024 * 1024
@@ -234,6 +244,20 @@ func TestCompressReaderWriter(t *testing.T) {
}
}
+func TestUncompressedStreamCloseDoesNotCloseUnderlyingStream(t *testing.T) {
+ codec, err := compress.GetCodec(compress.Codecs.Uncompressed)
+ assert.NoError(t, err)
+ streamingCodec := codec.(compress.StreamingCodec)
+
+ source := &closeTrackingBuffer{}
+ assert.NoError(t, streamingCodec.NewReader(source).Close())
+ assert.False(t, source.closed)
+
+ sink := &closeTrackingBuffer{}
+ assert.NoError(t, streamingCodec.NewWriter(sink).Close())
+ assert.False(t, sink.closed)
+}
+
var marshalTests = []struct {
text string
codec compress.Compression
