20230220399

Committed 15 Dec 2025 11:15AM UTC coverage: 94.629% (-0.1%) from 94.736%

Build # 20230220399

Build Type

Pull #584

github

Committed by

web-flow

Commit Message

Merge f25cd9f0f into 9af4aa479

Pull Request Pull Request #584: Fix zstd potential memory leak

Coverage Stats

11 of 19 new or added lines in 2 files covered. (57.89%)

6272 of 6628 relevant lines covered (94.63%)

31077.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.78

/ocf/codec.go

package ocf

import (
        "bytes"
        "compress/flate"
        "encoding/binary"
        "errors"
        "fmt"
        "hash/crc32"
        "io"

        "github.com/golang/snappy"
        "github.com/klauspost/compress/zstd"
)

// CodecName represents a compression codec name.
type CodecName string

// Supported compression codecs.
const (
        Null      CodecName = "null"
        Deflate   CodecName = "deflate"
        Snappy    CodecName = "snappy"
        ZStandard CodecName = "zstandard"
)

type codecOptions struct {
        DeflateCompressionLevel int
        ZStandardOptions        zstdOptions
}

type zstdOptions struct {
        EOptions []zstd.EOption
        DOptions []zstd.DOption
}

func resolveCodec(name CodecName, codecOpts codecOptions) (Codec, error) {
        switch name {
        case Null, "":
                return &NullCodec{}, nil

        case Deflate:
                return &DeflateCodec{compLvl: codecOpts.DeflateCompressionLevel}, nil

        case Snappy:
                return &SnappyCodec{}, nil

        case ZStandard:
                return newZStandardCodec(codecOpts.ZStandardOptions), nil

        default:
                return nil, fmt.Errorf("unknown codec %s", name)
        }
}

// Codec represents a compression codec.
type Codec interface {
        // Decode decodes the given bytes.
        Decode([]byte) ([]byte, error)
        // Encode encodes the given bytes.
        Encode([]byte) []byte
}

// NullCodec is a no op codec.
type NullCodec struct{}

// Decode decodes the given bytes.
func (*NullCodec) Decode(b []byte) ([]byte, error) {
        return b, nil
}

// Encode encodes the given bytes.
func (*NullCodec) Encode(b []byte) []byte {
        return b
}

// DeflateCodec is a flate compression codec.
type DeflateCodec struct {
        compLvl int
}

// Decode decodes the given bytes.
func (c *DeflateCodec) Decode(b []byte) ([]byte, error) {
        r := flate.NewReader(bytes.NewBuffer(b))
        data, err := io.ReadAll(r)
        if err != nil {
                _ = r.Close()
                return nil, err
        }
        _ = r.Close()

        return data, nil
}

// Encode encodes the given bytes.
func (c *DeflateCodec) Encode(b []byte) []byte {
        data := bytes.NewBuffer(make([]byte, 0, len(b)))

        w, _ := flate.NewWriter(data, c.compLvl)
        _, _ = w.Write(b)
        _ = w.Close()

        return data.Bytes()
}

// SnappyCodec is a snappy compression codec.
type SnappyCodec struct{}

// Decode decodes the given bytes.
func (*SnappyCodec) Decode(b []byte) ([]byte, error) {
        l := len(b)
        if l < 5 {
                return nil, errors.New("block does not contain snappy checksum")
        }

        dst, err := snappy.Decode(nil, b[:l-4])
        if err != nil {
                return nil, err
        }

        crc := binary.BigEndian.Uint32(b[l-4:])
        if crc32.ChecksumIEEE(dst) != crc {
                return nil, errors.New("snappy checksum mismatch")
        }

        return dst, nil
}

// Encode encodes the given bytes.
func (*SnappyCodec) Encode(b []byte) []byte {
        dst := snappy.Encode(nil, b)

        dst = append(dst, 0, 0, 0, 0)
        binary.BigEndian.PutUint32(dst[len(dst)-4:], crc32.ChecksumIEEE(b))

        return dst
}

// ZStandardCodec is a zstandard compression codec.
type ZStandardCodec struct {
        decoder *zstd.Decoder
        encoder *zstd.Encoder
}

func newZStandardCodec(opts zstdOptions) *ZStandardCodec {
        decoder, _ := zstd.NewReader(nil, opts.DOptions...)
        encoder, _ := zstd.NewWriter(nil, opts.EOptions...)
        return &ZStandardCodec{
                decoder: decoder,
                encoder: encoder,
        }
}

// Decode decodes the given bytes.
func (zstdCodec *ZStandardCodec) Decode(b []byte) ([]byte, error) {
        defer func() { _ = zstdCodec.decoder.Reset(nil) }()
        return zstdCodec.decoder.DecodeAll(b, nil)
}

// Encode encodes the given bytes.
func (zstdCodec *ZStandardCodec) Encode(b []byte) []byte {
        defer zstdCodec.encoder.Reset(nil)
        return zstdCodec.encoder.EncodeAll(b, nil)
}

// Close closes the zstandard encoder and decoder, releasing resources.
func (zstdCodec *ZStandardCodec) Close() error {
        if zstdCodec.decoder != nil {
                zstdCodec.decoder.Close()
        }
        if zstdCodec.encoder != nil {
                return zstdCodec.encoder.Close()
        }
        return nil
}

1	package ocf
2
3	import (
4	"bytes"
5	"compress/flate"
6	"encoding/binary"
7	"errors"
8	"fmt"
9	"hash/crc32"
10	"io"
11
12	"github.com/golang/snappy"
13	"github.com/klauspost/compress/zstd"
14	)
15
16	// CodecName represents a compression codec name.
17	type CodecName string
18
19	// Supported compression codecs.
20	const (
21	Null CodecName = "null"
22	Deflate CodecName = "deflate"
23	Snappy CodecName = "snappy"
24	ZStandard CodecName = "zstandard"
25	)
26
27	type codecOptions struct {
28	DeflateCompressionLevel int
29	ZStandardOptions zstdOptions
30	}
31
32	type zstdOptions struct {
33	EOptions []zstd.EOption
34	DOptions []zstd.DOption
35	}
36
37	func resolveCodec(name CodecName, codecOpts codecOptions) (Codec, error) {	100✔
38	switch name {	100✔
39	case Null, "":	60✔
40	return &NullCodec{}, nil	60✔
41
42	case Deflate:	8✔
43	return &DeflateCodec{compLvl: codecOpts.DeflateCompressionLevel}, nil	8✔
44
45	case Snappy:	10✔
46	return &SnappyCodec{}, nil	10✔
47
48	case ZStandard:	18✔
49	return newZStandardCodec(codecOpts.ZStandardOptions), nil	18✔
50
51	default:	4✔
52	return nil, fmt.Errorf("unknown codec %s", name)	4✔
53	}
54	}
55
56	// Codec represents a compression codec.
57	type Codec interface {
58	// Decode decodes the given bytes.
59	Decode([]byte) ([]byte, error)
60	// Encode encodes the given bytes.
61	Encode([]byte) []byte
62	}
63
64	// NullCodec is a no op codec.
65	type NullCodec struct{}
66
67	// Decode decodes the given bytes.
68	func (*NullCodec) Decode(b []byte) ([]byte, error) {	16✔
69	return b, nil	16✔
70	}	16✔
71
72	// Encode encodes the given bytes.
73	func (*NullCodec) Encode(b []byte) []byte {	30✔
74	return b	30✔
75	}	30✔
76
77	// DeflateCodec is a flate compression codec.
78	type DeflateCodec struct {
79	compLvl int
80	}
81
82	// Decode decodes the given bytes.
83	func (c *DeflateCodec) Decode(b []byte) ([]byte, error) {	4✔
84	r := flate.NewReader(bytes.NewBuffer(b))	4✔
85	data, err := io.ReadAll(r)	4✔
86	if err != nil {	6✔
87	_ = r.Close()	2✔
88	return nil, err	2✔
89	}	2✔
90	_ = r.Close()	2✔
91		2✔
92	return data, nil	2✔
93	}
94
95	// Encode encodes the given bytes.
96	func (c *DeflateCodec) Encode(b []byte) []byte {	4✔
97	data := bytes.NewBuffer(make([]byte, 0, len(b)))	4✔
98		4✔
99	w, _ := flate.NewWriter(data, c.compLvl)	4✔
100	_, _ = w.Write(b)	4✔
101	_ = w.Close()	4✔
102		4✔
103	return data.Bytes()	4✔
104	}	4✔
105
106	// SnappyCodec is a snappy compression codec.
107	type SnappyCodec struct{}
108
109	// Decode decodes the given bytes.
110	func (*SnappyCodec) Decode(b []byte) ([]byte, error) {	8✔
111	l := len(b)	8✔
112	if l < 5 {	10✔
113	return nil, errors.New("block does not contain snappy checksum")	2✔
114	}	2✔
115
116	dst, err := snappy.Decode(nil, b[:l-4])	6✔
117	if err != nil {	8✔
118	return nil, err	2✔
119	}	2✔
120
121	crc := binary.BigEndian.Uint32(b[l-4:])	4✔
122	if crc32.ChecksumIEEE(dst) != crc {	6✔
123	return nil, errors.New("snappy checksum mismatch")	2✔
124	}	2✔
125
126	return dst, nil	2✔
127	}
128
129	// Encode encodes the given bytes.
130	func (*SnappyCodec) Encode(b []byte) []byte {	2✔
131	dst := snappy.Encode(nil, b)	2✔
132		2✔
133	dst = append(dst, 0, 0, 0, 0)	2✔
134	binary.BigEndian.PutUint32(dst[len(dst)-4:], crc32.ChecksumIEEE(b))	2✔
135		2✔
136	return dst	2✔
137	}	2✔
138
139	// ZStandardCodec is a zstandard compression codec.
140	type ZStandardCodec struct {
141	decoder *zstd.Decoder
142	encoder *zstd.Encoder
143	}
144
145	func newZStandardCodec(opts zstdOptions) *ZStandardCodec {	18✔
146	decoder, _ := zstd.NewReader(nil, opts.DOptions...)	18✔
147	encoder, _ := zstd.NewWriter(nil, opts.EOptions...)	18✔
148	return &ZStandardCodec{	18✔
149	decoder: decoder,	18✔
150	encoder: encoder,	18✔
151	}	18✔
152	}	18✔
153
154	// Decode decodes the given bytes.
155	func (zstdCodec *ZStandardCodec) Decode(b []byte) ([]byte, error) {	16✔
156	defer func() { _ = zstdCodec.decoder.Reset(nil) }()	32✔
157	return zstdCodec.decoder.DecodeAll(b, nil)	16✔
158	}
159
160	// Encode encodes the given bytes.
161	func (zstdCodec *ZStandardCodec) Encode(b []byte) []byte {	12✔
162	defer zstdCodec.encoder.Reset(nil)	12✔
163	return zstdCodec.encoder.EncodeAll(b, nil)	12✔
164	}	12✔
165
166	// Close closes the zstandard encoder and decoder, releasing resources.
167	func (zstdCodec *ZStandardCodec) Close() error {	4✔
168	if zstdCodec.decoder != nil {	8✔
169	zstdCodec.decoder.Close()	4✔
170	}	4✔
171	if zstdCodec.encoder != nil {	8✔
172	return zstdCodec.encoder.Close()	4✔
173	}	4✔
NEW 174	return nil	×
175	}

hamba / avro / 20230220399

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous