diff --git a/parquet/compress/compress.go b/parquet/compress/compress.go index 4743c9867..835f94e42 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 eedf0f5f0..9410b2ede 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