Implement block encoder and decoder

Author: fpetkovski

alp/alp.go

@@ -2,6 +2,7 @@ package alp
 
 import (
 	"encoding/binary"
+	"errors"
 	"math"
 
 	"github.com/parquet-go/bitpack"
@@ -36,6 +37,8 @@ const (
 	EncodingUncompressed EncodingType = 3
 )
 
+var ErrInvalidEncoding = errors.New("invalid encoding")
+
 // CompressionMetadata contains metadata about the compressed data
 type CompressionMetadata struct {
 	EncodingType  EncodingType
@@ -100,7 +103,7 @@ func Encode(dst []byte, src []float64) []byte {
 		dst = make([]byte, packedSize+MetadataSize)
 	}
 	dst = dst[:packedSize]
-	bitpack.PackInt64(dst[MetadataSize:], forValues, uint(bitWidth))
+	bitpack.Pack(dst[MetadataSize:], forValues, uint(bitWidth))
 
 	// Create metadata
 	metadata := CompressionMetadata{
@@ -138,7 +141,7 @@ func Decode(dst []float64, data []byte) []float64 {
 	case EncodingALP:
 		result := dst[:metadata.Count]
 		ints := unsafecast.Slice[int64](result)
-		bitpack.UnpackInt64(ints, data[MetadataSize:], uint(metadata.BitWidth))
+		bitpack.Unpack(ints, data[MetadataSize:], uint(metadata.BitWidth))
 
 		minValue := metadata.FrameOfRef
 		numValues := metadata.Count
@@ -258,7 +261,7 @@ func DecompressValues(result []float64, src []byte, metadata CompressionMetadata
 		// Unpack src
 		packedData := src[MetadataSize:]
 		unpacked := unsafecast.Slice[int64](result)
-		bitpack.UnpackInt64(unpacked, packedData, uint(metadata.BitWidth))
+		bitpack.Unpack(unpacked, packedData, uint(metadata.BitWidth))
 
 		// Reverse frame-of-reference and convert back to float64 in one pass
 		minValue := metadata.FrameOfRef

alp/alp_stream.go

@@ -0,0 +1,159 @@
+package alp
+
+import (
+	"io"
+
+	"github.com/parquet-go/bitpack"
+	"github.com/parquet-go/bitpack/unsafecast"
+)
+
+// StreamEncode encodes float64 values using ALP with block-based packing for streaming decode
+func StreamEncode(dst []byte, src []float64, blockSize int) []byte {
+	if len(src) == 0 {
+		if cap(dst) < MetadataSize {
+			dst = make([]byte, MetadataSize)
+		}
+		dst = dst[:MetadataSize]
+
+		encodeMetadata(dst, CompressionMetadata{
+			EncodingType: EncodingNone,
+			Count:        0,
+		})
+		return dst
+	}
+
+	// Find global encoding parameters
+	exponent := findBestExponent(src)
+	factor := powersOf10[exponent+10]
+
+	// Convert all to integers with global exponent
+	forValues := encodeToIntegers(src, factor)
+
+	// Find global frame-of-reference
+	minValue := forValues[0]
+	for _, v := range forValues {
+		minValue = min(minValue, v)
+	}
+
+	// Apply frame-of-reference and find global bit-width
+	bitWidth := 0
+	for i, v := range forValues {
+		forValues[i] = v - minValue
+		bits := CalculateBitWidth(uint64(forValues[i]))
+		bitWidth = max(bitWidth, bits)
+	}
+
+	// Calculate total packed size.
+	blockSizeBytes := bitpack.ByteCount(uint(blockSize * bitWidth))
+	totalBlocks := (len(forValues) + blockSize - 1) / blockSize
+	packedSize := blockSizeBytes*totalBlocks + bitpack.PaddingInt64
+
+	// Create output buffer: metadata + packed blocks
+	totalSize := MetadataSize + packedSize
+	if cap(dst) < totalSize {
+		dst = make([]byte, totalSize)
+	}
+	dst = dst[:totalSize]
+
+	encodeMetadata(dst, CompressionMetadata{
+		EncodingType: EncodingALP,
+		Count:        int32(len(src)),
+		Exponent:     int8(exponent),
+		BitWidth:     uint8(bitWidth),
+		FrameOfRef:   minValue,
+	})
+
+	// Pack data in blocks continuously block after block.
+	offset := MetadataSize
+	for i := range totalBlocks {
+		var (
+			blockStart = i * blockSize
+			blockEnd   = min(blockStart+blockSize, len(forValues))
+			blockData  = forValues[blockStart:blockEnd]
+		)
+		// Pack this block.
+		bitpack.Pack(dst[offset:offset+blockSizeBytes], blockData, uint(bitWidth))
+		offset += blockSizeBytes
+	}
+
+	return dst
+}
+
+type StreamDecoder struct {
+	buf              []byte
+	metadata         CompressionMetadata
+	blockSize        int
+	decodedBuf       []float64 // Buffer for decoded block
+	decodedBufOffset int       // Current read position in decoded buffer
+	valuesRead       int32     // Total values read so far
+}
+
+func (d *StreamDecoder) Reset(buf []byte, blockSize int) {
+	d.buf = buf
+	d.blockSize = blockSize
+	if cap(d.decodedBuf) < blockSize {
+		d.decodedBuf = make([]float64, 0, blockSize)
+	}
+	d.decodedBuf = d.decodedBuf[:0]
+	d.decodedBufOffset = 0
+	d.valuesRead = 0
+
+	// Read global metadata
+	if len(buf) >= MetadataSize {
+		d.metadata = DecodeMetadata(buf)
+		d.buf = buf[MetadataSize:]
+	}
+}
+
+func (d *StreamDecoder) Decode(dst []float64) ([]float64, error) {
+	if d.valuesRead >= d.metadata.Count {
+		return dst[:0], io.EOF
+	}
+
+	// If we've consumed all values from current block, decode next block
+	if d.decodedBufOffset >= len(d.decodedBuf) {
+		// Determine block size
+		remaining := d.metadata.Count - d.valuesRead
+		blockSize := min(int32(d.blockSize), remaining)
+
+		// Calculate size of packed data for this block (no per-block padding)
+		packedSize := bitpack.ByteCount(uint(int(blockSize) * int(d.metadata.BitWidth)))
+
+		// Allocate buffer for decoded block
+		if cap(d.decodedBuf) < int(blockSize) {
+			d.decodedBuf = make([]float64, blockSize)
+		}
+		d.decodedBuf = d.decodedBuf[:blockSize]
+
+		// Unpack entire block
+		ints := unsafecast.Slice[int64](d.decodedBuf)
+		bitpack.Unpack(ints, d.buf, uint(d.metadata.BitWidth))
+		d.buf = d.buf[packedSize:]
+
+		// Convert to float64
+		minValue := d.metadata.FrameOfRef
+		invFactor := powersOf10[(10-d.metadata.Exponent+21)%21]
+
+		for i := range d.decodedBuf {
+			d.decodedBuf[i] = float64(ints[i]+minValue) * invFactor
+		}
+
+		d.decodedBufOffset = 0
+	}
+
+	// Return a chunk from the decoded buffer
+	remaining := len(d.decodedBuf) - d.decodedBufOffset
+	n := min(len(dst), remaining)
+
+	copy(dst[:n], d.decodedBuf[d.decodedBufOffset:d.decodedBufOffset+n])
+	d.decodedBufOffset += n
+	d.valuesRead += int32(n)
+
+	// Check if we're done
+	var err error
+	if d.valuesRead >= d.metadata.Count {
+		err = io.EOF
+	}
+
+	return dst[:n], err
+}

alp/alp_test.go

@@ -1,6 +1,7 @@
 package alp
 
 import (
+	"io"
 	"math"
 	"testing"
 
@@ -238,17 +239,169 @@ func TestFindMaxBitWidth(t *testing.T) {
 // PackInt64Array packs an array of uint64 values with the same bit width
 func PackInt64Array(values []int64, bitWidth uint) []byte {
 	dst := make([]byte, BitPackedSize(uint32(len(values)), bitWidth))
-	bitpack.PackInt64(dst, values, bitWidth)
+	bitpack.Pack(dst, values, bitWidth)
 	return dst
 }
 
 // UnpackInt64Array unpacks an array of int64 values
 func UnpackInt64Array(data []byte, count int, bitWidth uint) []int64 {
 	dst := make([]int64, count)
-	bitpack.UnpackInt64(dst, data, bitWidth)
+	bitpack.Unpack(dst, data, bitWidth)
 	return dst
 }
 
 func BitPackedSize(numValues uint32, bitWidth uint) int {
 	return bitpack.ByteCount(uint(numValues)*bitWidth) + bitpack.PaddingInt64
 }
+
+func TestDecodeRange(t *testing.T) {
+	tests := []struct {
+		name      string
+		data      []float64
+		blockSize int
+		bufSize   int
+	}{
+		{
+			name:      "smaller read buffer than values",
+			data:      []float64{6, 2, 3, 4, 5, 6},
+			blockSize: 120,
+			bufSize:   3,
+		},
+		{
+			name:      "smaller read buffer than values",
+			data:      []float64{1, 2, 3, 4, 5, 6},
+			blockSize: 120,
+			bufSize:   3,
+		},
+		{
+			name:      "smaller read buffer than values",
+			data:      []float64{1.1, 2.2, 3.2, 4.4, 5.5, 6.6, 7.7, 8.8, 9.9, 10.0},
+			blockSize: 120,
+			bufSize:   3,
+		},
+		{
+			name:      "read buffer multiple of values",
+			data:      []float64{1.1, 2.2, 3.3, 4.4, 5.5, 6.6, 7.7, 8.8},
+			blockSize: 120,
+			bufSize:   4,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Encode with StreamEncode
+			compressed := StreamEncode(nil, tt.data, tt.blockSize)
+
+			// Decode with StreamDecoder
+			decoder := StreamDecoder{}
+			decoder.Reset(compressed, tt.blockSize)
+
+			fullDecoded := make([]float64, 0, len(tt.data))
+			readBuf := make([]float64, tt.bufSize)
+
+			for {
+				readBuf, err := decoder.Decode(readBuf)
+				if err != io.EOF {
+					if err != nil {
+						t.Fatalf("unexpected error %v", err)
+					}
+				}
+				fullDecoded = append(fullDecoded, readBuf...)
+				if err == io.EOF {
+					break
+				}
+			}
+			// Verify length
+			if len(fullDecoded) != len(tt.data) {
+				t.Fatalf("length mismatch: got %d, want %d", len(fullDecoded), len(tt.data))
+			}
+
+			// Verify values with tolerance
+			for i := range tt.data {
+				if math.Abs(fullDecoded[i]-tt.data[i]) > 1e-10 {
+					t.Errorf("value mismatch at index %d: got %f, want %f", i, fullDecoded[i], tt.data[i])
+				}
+			}
+		})
+	}
+}
+
+func TestStreamEncoderDecoder(t *testing.T) {
+	tests := []struct {
+		name      string
+		data      []float64
+		blockSize int
+	}{
+		{
+			name:      "small dataset",
+			data:      []float64{1, 2, 3, 4, 5, 6},
+			blockSize: 3,
+		},
+		{
+			name:      "small dataset uneven block size",
+			data:      []float64{1, 2, 3, 4, 5},
+			blockSize: 3,
+		},
+		{
+			name:      "single block",
+			data:      []float64{1.1, 2.2, 3.3, 4.4, 5.5, 6.6, 7.7, 8.8},
+			blockSize: 120,
+		},
+		{
+			name:      "multiple blocks different ranges",
+			data:      []float64{1, 2, 3, 100, 101, 102, 1000, 1001, 1002},
+			blockSize: 3,
+		},
+		{
+			name: "large dataset",
+			data: func() []float64 {
+				data := make([]float64, 1000)
+				for i := range data {
+					data[i] = float64(i) * 0.1
+				}
+				return data
+			}(),
+			blockSize: 120,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Encode
+			compressed := StreamEncode(nil, tt.data, tt.blockSize)
+
+			t.Logf("Original size: %d bytes, Compressed size: %d bytes, Ratio: %.2f%%",
+				len(tt.data)*8, len(compressed), float64(len(compressed))/float64(len(tt.data)*8)*100)
+
+			// Decode
+			decoder := StreamDecoder{}
+			decoder.Reset(compressed, tt.blockSize)
+
+			decoded := make([]float64, 0, len(tt.data))
+			readBuf := make([]float64, tt.blockSize)
+
+			for {
+				result, err := decoder.Decode(readBuf)
+				decoded = append(decoded, result...)
+				if err == io.EOF {
+					break
+				}
+				if err != nil {
+					t.Fatalf("unexpected error: %v", err)
+				}
+			}
+
+			// Verify length
+			if len(decoded) != len(tt.data) {
+				t.Fatalf("length mismatch: got %d, want %d", len(decoded), len(tt.data))
+			}
+
+			// Verify values
+			for i := range tt.data {
+				if math.Abs(decoded[i]-tt.data[i]) > 1e-10 {
+					t.Errorf("value mismatch at index %d: got %f, want %f", i, decoded[i], tt.data[i])
+				}
+			}
+		})
+	}
+}