[arrow] branch master updated: ARROW-16163: [Go] IPC FileReader leaks memory when used with ZSTD compression

[ze...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zeroshade pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git


The following commit(s) were added to refs/heads/master by this push:
     new 667d427d8e ARROW-16163: [Go] IPC FileReader leaks memory when used with ZSTD compression
667d427d8e is described below

commit 667d427d8ef16b1535dc82ce6fe6b7384382231c
Author: Nicolas Moreau <ni...@botify.com>
AuthorDate: Wed Apr 13 13:15:37 2022 -0400

    ARROW-16163: [Go] IPC FileReader leaks memory when used with ZSTD compression
    
    When used in its stream version (calling .Read()), the ZSTD decoder makes use of a Goroutine with a channel to receive input. This channel is closed by the decoder's .Close() function, which ends the Goroutine and therefore the collection of its memory by the GC.
    
    Here, we add Close() to the exposed decompressor interface and call its close at the end of the function that uses it.
    
    Closes #12857 from cptjacky/ARROW-16163
    
    Authored-by: Nicolas Moreau <ni...@botify.com>
    Signed-off-by: Matthew Topol <mt...@factset.com>
---
 go/arrow/ipc/compression.go | 13 ++++++++++++-
 go/arrow/ipc/file_reader.go |  1 +
 2 files changed, 13 insertions(+), 1 deletion(-)

diff --git a/go/arrow/ipc/compression.go b/go/arrow/ipc/compression.go
index 352f8ed92c..3f1e00a123 100644
--- a/go/arrow/ipc/compression.go
+++ b/go/arrow/ipc/compression.go
@@ -82,6 +82,7 @@ func getCompressor(codec flatbuf.CompressionType) compressor {
 type decompressor interface {
 	io.Reader
 	Reset(io.Reader)
+	Close()
 }
 
 type zstdDecompressor struct {
@@ -94,10 +95,20 @@ func (z *zstdDecompressor) Reset(r io.Reader) {
 	}
 }
 
+func (z *zstdDecompressor) Close() {
+	z.Decoder.Close()
+}
+
+type lz4Decompressor struct {
+	*lz4.Reader
+}
+
+func (z *lz4Decompressor) Close() {}
+
 func getDecompressor(codec flatbuf.CompressionType) decompressor {
 	switch codec {
 	case flatbuf.CompressionTypeLZ4_FRAME:
-		return lz4.NewReader(nil)
+		return &lz4Decompressor{lz4.NewReader(nil)}
 	case flatbuf.CompressionTypeZSTD:
 		dec, err := zstd.NewReader(nil)
 		if err != nil {
diff --git a/go/arrow/ipc/file_reader.go b/go/arrow/ipc/file_reader.go
index 3734fd6781..98cf7ba997 100644
--- a/go/arrow/ipc/file_reader.go
+++ b/go/arrow/ipc/file_reader.go
@@ -326,6 +326,7 @@ func newRecord(schema *arrow.Schema, meta *memory.Buffer, body ReadAtSeeker, mem
 	bodyCompress := md.Compression(nil)
 	if bodyCompress != nil {
 		codec = getDecompressor(bodyCompress.Codec())
+		defer codec.Close()
 	}
 
 	ctx := &arrayLoaderContext{