fix(parquet/compress): avoid closing caller streams for uncompressed wrappers

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) {

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