From ae634e9920772a7d9162a9d658892ae77457a672 Mon Sep 17 00:00:00 2001 From: Baptiste Canton Date: Sat, 22 Aug 2026 18:47:32 +0200 Subject: [PATCH] feat(mutators): add shuffle/unshuffle byte-transposition filter Groups fixed-width elements by byte position, so a following compressor sees runs of similar bytes instead of the interleaving that defeats LZ77. This is the classic filter shipped by Blosc, HDF5 and Parquet. It compresses nothing by itself; it is meant to be chained: ccat data.bin -m shuffle,zstd ccat data.sz -m unzstd,unshuffle On 160KB of float64 samples, where general-purpose compressors are weak or counterproductive: zstd xz shuffle,zstd shuffle,xz random doubles 160019 143208 141909 129828 smooth series 160019 124732 126483 107812 (the input is 160000 bytes: zstd alone grows it) Element size is the standard X:8 argument, so it also covers float32 and int32 (shuffle:4), RGB pixels (shuffle:3) and so on. Blocks of 64KiB are transposed at a time and both directions derive the same block size from the element size, so the stream needs no header and stays streamable. Trailing bytes that do not fill an element pass through untouched. Co-Authored-By: Claude Fable 5 --- pkg/mutators/single/shuffle.go | 97 +++++++++++++++++++++++++++++ pkg/mutators/single/shuffle_test.go | 85 +++++++++++++++++++++++++ 2 files changed, 182 insertions(+) create mode 100644 pkg/mutators/single/shuffle.go create mode 100644 pkg/mutators/single/shuffle_test.go diff --git a/pkg/mutators/single/shuffle.go b/pkg/mutators/single/shuffle.go new file mode 100644 index 000000000..ca1bd774c --- /dev/null +++ b/pkg/mutators/single/shuffle.go @@ -0,0 +1,97 @@ +package mutators + +import ( + "fmt" + "io" +) + +// shuffleTargetBlock is the amount of data transposed at once, in bytes. It is +// rounded down to a whole number of elements, and both shuffle and unshuffle +// derive the same value from the element size, so no header is needed to +// round-trip: each side simply reads one block at a time. +const shuffleTargetBlock = 64 * 1024 + +func init() { + singleRegister("shuffle", shuffle, + withDescription("group the X:8-byte elements by byte position, so a following compressor sees runs instead of interleaved bytes"), + withCategory("filter"), + withConfigBuilder(stdConfigUint64WithDefault(8)), + ) + singleRegister("unshuffle", unshuffle, + withDescription("reverse shuffle (X:8 must match the shuffle element size)"), + withCategory("filter"), + withConfigBuilder(stdConfigUint64WithDefault(8)), + ) +} + +func shuffle(w io.WriteCloser, r io.ReadCloser, config any) (int64, error) { + return transposeBlocks(w, r, config, transpose) +} + +func unshuffle(w io.WriteCloser, r io.ReadCloser, config any) (int64, error) { + return transposeBlocks(w, r, config, untranspose) +} + +// transposeBlocks streams r to w, applying f to one block at a time. +func transposeBlocks(w io.WriteCloser, r io.ReadCloser, config any, f func(dst, src []byte, n int)) (int64, error) { + n := cfgInt(config) + if n < 1 { + return 0, fmt.Errorf("element size must be at least 1, got %d", n) + } + + blockSize := shuffleTargetBlock / n * n + if blockSize == 0 { + // an element larger than the target block still gets its own block + blockSize = n + } + src := make([]byte, blockSize) + dst := make([]byte, blockSize) + + var written int64 + for { + read, err := io.ReadFull(r, src) + if read > 0 { + f(dst[:read], src[:read], n) + nw, werr := w.Write(dst[:read]) + written += int64(nw) + if werr != nil { + return written, werr + } + } + switch err { + case nil: + case io.EOF, io.ErrUnexpectedEOF: + return written, nil + default: + return written, err + } + } +} + +// transpose groups src by byte position: every first byte of an element, then +// every second one, and so on. Trailing bytes that do not fill an element are +// copied as-is, and untranspose leaves them alone too, so they round-trip. +func transpose(dst, src []byte, n int) { + m := len(src) / n + k := 0 + for j := range n { + for i := range m { + dst[k] = src[i*n+j] + k++ + } + } + copy(dst[k:], src[m*n:]) +} + +// untranspose is the inverse permutation of transpose. +func untranspose(dst, src []byte, n int) { + m := len(src) / n + k := 0 + for j := range n { + for i := range m { + dst[i*n+j] = src[k] + k++ + } + } + copy(dst[k:], src[m*n:]) +} diff --git a/pkg/mutators/single/shuffle_test.go b/pkg/mutators/single/shuffle_test.go new file mode 100644 index 000000000..b09d43f5e --- /dev/null +++ b/pkg/mutators/single/shuffle_test.go @@ -0,0 +1,85 @@ +package mutators_test + +import ( + "bytes" + "encoding/binary" + "fmt" + "math" + "strings" + "testing" + + "github.com/batmac/ccat/pkg/mutators" +) + +// float64s builds a byte stream of count doubles, as ccat would receive it. +func float64s(count int, f func(i int) float64) string { + var b bytes.Buffer + for i := range count { + _ = binary.Write(&b, binary.LittleEndian, f(i)) + } + return b.String() +} + +func TestShuffleRoundTrip(t *testing.T) { + tests := []struct { + name string + size int + input string + }{ + {"empty", 8, ""}, + {"single element", 8, "abcdefgh"}, + {"shorter than one element", 8, "abc"}, + {"not a multiple of the element size", 8, "abcdefghij"}, + {"element size 1 is a no-op", 1, "hello world"}, + {"element size 3", 3, "aaabbbcccddd"}, + {"element size 4", 4, float64s(64, func(i int) float64 { return float64(i) })}, + {"doubles", 8, float64s(1000, func(i int) float64 { return 100 + math.Sin(float64(i)/10) })}, + // larger than one 64KiB block, to exercise the block loop + {"multi-block", 8, float64s(20000, func(i int) float64 { return float64(i) * 1.5 })}, + {"multi-block, partial last block", 8, float64s(20000, func(i int) float64 { return float64(i) }) + "xyz"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + shuffled := mutators.Run(fmt.Sprintf("shuffle:%d", tt.size), tt.input) + if len(shuffled) != len(tt.input) { + t.Fatalf("shuffle changed the length: got %d, want %d", len(shuffled), len(tt.input)) + } + got := mutators.Run(fmt.Sprintf("unshuffle:%d", tt.size), shuffled) + if got != tt.input { + t.Errorf("round-trip mismatch: got %d bytes, want %d", len(got), len(tt.input)) + } + }) + } +} + +func TestShuffleGroupsBytePositions(t *testing.T) { + // four 4-byte elements: the transform must emit all the first bytes, then + // all the second ones, and so on + input := "AbcdAbcdAbcdAbcd" + want := "AAAAbbbbccccdddd" + if got := mutators.Run("shuffle:4", input); got != want { + t.Errorf("shuffle:4 = %q, want %q", got, want) + } +} + +func TestShuffleHelpsCompression(t *testing.T) { + // the point of the filter: a compressor that cannot do anything with + // interleaved doubles does well once they are grouped by byte position + input := float64s(4000, func(i int) float64 { return 1000 + math.Sin(float64(i)/50) }) + + plain := len(mutators.Run("zstd", input)) + shuffled := len(mutators.Run("zstd", mutators.Run("shuffle:8", input))) + + t.Logf("%d bytes -> zstd %d, shuffle+zstd %d", len(input), plain, shuffled) + if shuffled >= plain { + t.Errorf("shuffle did not help: shuffle+zstd = %d, zstd alone = %d", shuffled, plain) + } +} + +func TestShuffleDefaultsToEightBytes(t *testing.T) { + input := strings.Repeat("Abcdefgh", 4) + if mutators.Run("shuffle", input) != mutators.Run("shuffle:8", input) { + t.Error("shuffle should default to an element size of 8") + } +}