You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2025/07/16 06:01:40 UTC
(skywalking-banyandb) branch main updated: Support dictionary encoding for low cardinality columns (#695)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 1ab7ebdf Support dictionary encoding for low cardinality columns (#695)
1ab7ebdf is described below
commit 1ab7ebdf539a3764185e018d3444a36a6e5a5fe8
Author: Huang Youliang <bu...@gmail.com>
AuthorDate: Wed Jul 16 14:01:35 2025 +0800
Support dictionary encoding for low cardinality columns (#695)
* Support dictionary encoding for low cardinality columns
---------
Co-authored-by: Copilot <17...@users.noreply.github.com>
Co-authored-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
banyand/measure/column.go | 35 +++++++++++++++++++++++++++++++++--
banyand/stream/tag.go | 35 +++++++++++++++++++++++++++++++++--
pkg/encoding/dictionary.go | 27 ++++++++++++++++++---------
pkg/encoding/dictionary_test.go | 26 +++++++++++++-------------
pkg/encoding/encoding.go | 1 +
6 files changed, 99 insertions(+), 26 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 5cc0071b..1a401b96 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -20,6 +20,7 @@ Release Notes.
- Add Load Balancer Feature to Liaison.
- Implement fadvise for large files to prevent page cache pollution.
- Data Model: Introduce the `Trace` data model to store the trace/span data.
+- Support dictionary encoding for low cardinality columns.
- Push down aggregation for topN query.
### Bug Fixes
diff --git a/banyand/measure/column.go b/banyand/measure/column.go
index 2e8961ce..1c17929b 100644
--- a/banyand/measure/column.go
+++ b/banyand/measure/column.go
@@ -65,9 +65,23 @@ func releaseFloat64Slice(float64Slice *[]float64) {
float64SlicePool.Put(float64Slice)
}
+func generateDictionary() *encoding.Dictionary {
+ v := dictionaryPool.Get()
+ if v == nil {
+ return &encoding.Dictionary{}
+ }
+ return v
+}
+
+func releaseDictionary(d *encoding.Dictionary) {
+ d.Reset()
+ dictionaryPool.Put(d)
+}
+
var (
int64SlicePool = pool.Register[*[]int64]("measure-int64Slice")
float64SlicePool = pool.Register[*[]float64]("measure-float64Slice")
+ dictionaryPool = pool.Register[*encoding.Dictionary]("measure-dictionary")
)
type column struct {
@@ -206,7 +220,17 @@ func (c *column) encodeFloat64Column(bb *bytes.Buffer) {
}
func (c *column) encodeDefault(bb *bytes.Buffer) {
- bb.Buf = encoding.EncodeBytesBlock(bb.Buf[:0], c.values)
+ dict := generateDictionary()
+ defer releaseDictionary(dict)
+ for _, v := range c.values {
+ if !dict.Add(v) {
+ bb.Buf = encoding.EncodeBytesBlock(bb.Buf[:0], c.values)
+ bb.Buf = append([]byte{byte(encoding.EncodeTypePlain)}, bb.Buf...)
+ return
+ }
+ }
+ bb.Buf = dict.Encode(bb.Buf[:0])
+ bb.Buf = append([]byte{byte(encoding.EncodeTypeDictionary)}, bb.Buf...)
}
func (c *column) mustReadValues(decoder *encoding.BytesBlockDecoder, reader fs.Reader, cm columnMetadata, count uint64) {
@@ -340,8 +364,15 @@ func (c *column) decodeFloat64Column(decoder *encoding.BytesBlockDecoder, path s
}
func (c *column) decodeDefault(decoder *encoding.BytesBlockDecoder, bb *bytes.Buffer, count uint64, path string) {
+ encodeType := encoding.EncodeType(bb.Buf[0])
var err error
- c.values, err = decoder.Decode(c.values[:0], bb.Buf, count)
+ if encodeType == encoding.EncodeTypeDictionary {
+ dict := generateDictionary()
+ defer releaseDictionary(dict)
+ c.values, err = dict.Decode(c.values[:0], bb.Buf[1:], count)
+ } else {
+ c.values, err = decoder.Decode(c.values[:0], bb.Buf, count)
+ }
if err != nil {
logger.Panicf("%s: cannot decode values: %v", path, err)
}
diff --git a/banyand/stream/tag.go b/banyand/stream/tag.go
index cf0e0b11..aeefef1f 100644
--- a/banyand/stream/tag.go
+++ b/banyand/stream/tag.go
@@ -65,9 +65,23 @@ func releaseFloat64Slice(float64Slice *[]float64) {
float64SlicePool.Put(float64Slice)
}
+func generateDictionary() *encoding.Dictionary {
+ v := dictionaryPool.Get()
+ if v == nil {
+ return &encoding.Dictionary{}
+ }
+ return v
+}
+
+func releaseDictionary(d *encoding.Dictionary) {
+ d.Reset()
+ dictionaryPool.Put(d)
+}
+
var (
int64SlicePool = pool.Register[*[]int64]("stream-int64Slice")
float64SlicePool = pool.Register[*[]float64]("stream-float64Slice")
+ dictionaryPool = pool.Register[*encoding.Dictionary]("stream-dictionary")
)
type tag struct {
@@ -221,7 +235,17 @@ func (t *tag) encodeFloat64Tag(bb *bytes.Buffer) {
}
func (t *tag) encodeDefault(bb *bytes.Buffer) {
- bb.Buf = encoding.EncodeBytesBlock(bb.Buf[:0], t.values)
+ dict := generateDictionary()
+ defer releaseDictionary(dict)
+ for _, v := range t.values {
+ if !dict.Add(v) {
+ bb.Buf = encoding.EncodeBytesBlock(bb.Buf[:0], t.values)
+ bb.Buf = append([]byte{byte(encoding.EncodeTypePlain)}, bb.Buf...)
+ return
+ }
+ }
+ bb.Buf = dict.Encode(bb.Buf[:0])
+ bb.Buf = append([]byte{byte(encoding.EncodeTypeDictionary)}, bb.Buf...)
}
func (t *tag) mustReadValues(decoder *encoding.BytesBlockDecoder, reader fs.Reader, cm tagMetadata, count uint64) {
@@ -355,8 +379,15 @@ func (t *tag) decodeFloat64Tag(decoder *encoding.BytesBlockDecoder, path string,
}
func (t *tag) decodeDefault(decoder *encoding.BytesBlockDecoder, bb *bytes.Buffer, count uint64, path string) {
+ encodeType := encoding.EncodeType(bb.Buf[0])
var err error
- t.values, err = decoder.Decode(t.values[:0], bb.Buf, count)
+ if encodeType == encoding.EncodeTypeDictionary {
+ dict := generateDictionary()
+ defer releaseDictionary(dict)
+ t.values, err = dict.Decode(t.values[:0], bb.Buf[1:], count)
+ } else {
+ t.values, err = decoder.Decode(t.values[:0], bb.Buf, count)
+ }
if err != nil {
logger.Panicf("%s: cannot decode values: %v", path, err)
}
diff --git a/pkg/encoding/dictionary.go b/pkg/encoding/dictionary.go
index a878bfa2..ed9b4a61 100644
--- a/pkg/encoding/dictionary.go
+++ b/pkg/encoding/dictionary.go
@@ -29,6 +29,7 @@ const maxUniqueValues = 256
type Dictionary struct {
values [][]byte
indices []uint32
+ tmp []uint32
}
// NewDictionary creates a dictionary.
@@ -36,6 +37,7 @@ func NewDictionary() *Dictionary {
return &Dictionary{
values: make([][]byte, 0),
indices: make([]uint32, 0),
+ tmp: make([]uint32, 0),
}
}
@@ -43,6 +45,7 @@ func NewDictionary() *Dictionary {
func (d *Dictionary) Reset() {
d.values = d.values[:0]
d.indices = d.indices[:0]
+ d.tmp = d.tmp[:0]
}
// Add adds a value to the dictionary.
@@ -63,34 +66,40 @@ func (d *Dictionary) Add(value []byte) bool {
}
// Encode encodes the dictionary.
-func (d *Dictionary) Encode(dst []byte, tmp []uint32) []byte {
+func (d *Dictionary) Encode(dst []byte) []byte {
dst = VarUint64ToBytes(dst, uint64(len(d.values)))
dst = EncodeBytesBlock(dst, d.values)
- re := encodeRLE(tmp, d.indices)
+ re := encodeRLE(d.tmp, d.indices)
be := encodeBitPacking(re)
dst = append(dst, be...)
return dst
}
// Decode decodes the dictionary.
-func (d *Dictionary) Decode(src []byte, tmp []uint32) error {
+func (d *Dictionary) Decode(dst [][]byte, src []byte, itemsCount uint64) ([][]byte, error) {
src, count := BytesToVarUint64(src)
if count == 0 {
- return nil
+ return dst, nil
}
values, src, err := d.decodeBytesBlockWithTail(src, count)
if err != nil {
- return err
+ return nil, err
}
d.values = values
- tmp, err = decodeBitPacking(tmp, src)
+ d.tmp, err = decodeBitPacking(d.tmp, src)
if err != nil {
- return err
+ return nil, err
}
- d.indices = decodeRLE(d.indices[:0], tmp)
- return nil
+ d.indices = decodeRLE(d.indices, d.tmp)
+ if uint64(len(d.indices)) != itemsCount {
+ return nil, fmt.Errorf("unexpected item counts; got %d; want %d", len(d.indices), itemsCount)
+ }
+ for _, index := range d.indices {
+ dst = append(dst, d.values[index])
+ }
+ return dst, nil
}
func (d *Dictionary) decodeBytesBlockWithTail(src []byte, itemsCount uint64) ([][]byte, []byte, error) {
diff --git a/pkg/encoding/dictionary_test.go b/pkg/encoding/dictionary_test.go
index d2b0e9f8..5445e03b 100644
--- a/pkg/encoding/dictionary_test.go
+++ b/pkg/encoding/dictionary_test.go
@@ -56,16 +56,17 @@ func TestEncodeAndDecodeDictionary(t *testing.T) {
dict.Add(value)
}
- tmp := make([]uint32, 0)
- encoded := dict.Encode(nil, tmp)
- decoded := NewDictionary()
- err := decoded.Decode(encoded, tmp[:0])
+ encoded := dict.Encode(nil)
+ decoded := make([][]byte, 0)
+ dict.Reset()
+ decoded, err := dict.Decode(decoded, encoded, 5)
require.NoError(t, err)
expectedValues := [][]byte{[]byte("skywalking"), []byte("banyandb"), []byte("hello"), []byte("world")}
- require.Equal(t, expectedValues, decoded.values)
+ require.Equal(t, expectedValues, dict.values)
expectedIndices := []uint32{0, 1, 2, 3, 2}
- require.Equal(t, expectedIndices, decoded.indices)
+ require.Equal(t, expectedIndices, dict.indices)
+ require.Equal(t, values, decoded)
}
type parameter struct {
@@ -90,15 +91,14 @@ func BenchmarkEncodeDictionary(b *testing.B) {
}
encoded := make([]byte, 0)
- tmp := make([]uint32, 0)
- encoded = dict.Encode(encoded, tmp)
+ encoded = dict.Encode(encoded)
compressedSize := len(encoded)
compressRatio := float64(rawSize) / float64(compressedSize)
b.ResetTimer()
b.ReportMetric(compressRatio, "compress_ratio")
for i := 0; i < b.N; i++ {
- dict.Encode(encoded[:0], tmp[:0])
+ dict.Encode(encoded[:0])
}
})
}
@@ -114,13 +114,13 @@ func BenchmarkDecodeDictionary(b *testing.B) {
dict.Add(value)
}
encoded := make([]byte, 0)
- tmp := make([]uint32, 0)
- encoded = dict.Encode(encoded, tmp)
+ encoded = dict.Encode(encoded)
b.ResetTimer()
+ decoded := NewDictionary()
for i := 0; i < b.N; i++ {
- decoded := NewDictionary()
- err := decoded.Decode(encoded, tmp[:0])
+ decoded.Reset()
+ _, err := decoded.Decode(values[:0], encoded, uint64(p.count))
if err != nil {
b.Fatal(err)
}
diff --git a/pkg/encoding/encoding.go b/pkg/encoding/encoding.go
index 2683d0f8..a6a24494 100644
--- a/pkg/encoding/encoding.go
+++ b/pkg/encoding/encoding.go
@@ -94,6 +94,7 @@ const (
EncodeTypeDeltaWithVersion
EncodeTypeDeltaOfDeltaWithVersion
EncodeTypePlain
+ EncodeTypeDictionary
)
// GetVersionType returns the version type of the given encoding type.