add capnproto recordio pkg

Makefile

@@ -26,6 +26,11 @@ compile-proto:
 	protoc --go_out=. --go_opt=paths=source_relative _examples/proto/mutation.proto
 	protoc --go_out=. --go_opt=paths=source_relative benchmark/proto/bench.proto
 	protoc --go_out=. --go_opt=paths=source_relative sstables/proto/sstable.proto
+	@echo
+	@echo "==> Compiling capnproto files <=="
+	capnp compile -ogo -I ../go-capnp/std/ benchmark/capnproto/bench.capnp
+	capnp compile -ogo -I ../go-capnp/std/ recordio/test_files/text_line.capnp
+
 
 .PHONY: release
 release:

benchmark/capnproto/bench.capnp

@@ -0,0 +1,8 @@
+@0xbf8381dfd6a0d017;
+using Go = import "/go.capnp";
+$Go.package("capnproto");
+$Go.import("benchmark/capnproto");
+
+struct BytesMsg {
+  key @0 :Data;
+}

benchmark/recordio_read_test.go

@@ -1,10 +1,14 @@
 package benchmark
 
 import (
+	"capnproto.org/go/capnp/v3"
 	"errors"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	bCapnProto "github.com/thomasjungblut/go-sstables/benchmark/capnproto"
 	bProto "github.com/thomasjungblut/go-sstables/benchmark/proto"
 	"github.com/thomasjungblut/go-sstables/recordio"
+	rCapnProto "github.com/thomasjungblut/go-sstables/recordio/capnproto"
 	rProto "github.com/thomasjungblut/go-sstables/recordio/proto"
 	"io"
 	"os"
@@ -117,3 +121,65 @@ func BenchmarkRecordIOProtoRead(b *testing.B) {
 	}
 
 }
+
+func BenchmarkRecordIOCapnProtoRead(b *testing.B) {
+	benchmarks := []struct {
+		name     string
+		fileSize int
+	}{
+		{"32mb", 1024 * 1024 * 32},
+		{"64mb", 1024 * 1024 * 64},
+		{"128mb", 1024 * 1024 * 128},
+		{"256mb", 1024 * 1024 * 256},
+		{"512mb", 1024 * 1024 * 512},
+		{"1024mb", 1024 * 1024 * 1024},
+		{"2048mb", 1024 * 1024 * 1024 * 2},
+		{"4096mb", 1024 * 1024 * 1024 * 4},
+		{"8192mb", 1024 * 1024 * 1024 * 8},
+	}
+
+	for _, bm := range benchmarks {
+		b.Run(bm.name, func(b *testing.B) {
+			bytes := randomRecordOfSize(1024)
+			tmpFile, err := os.CreateTemp("", "recordio_Bench")
+			assert.NoError(b, err)
+			defer os.Remove(tmpFile.Name())
+
+			writer, err := rCapnProto.NewWriter(rCapnProto.File(tmpFile))
+			assert.NoError(b, err)
+			assert.NoError(b, writer.Open())
+
+			arena := capnp.SingleSegment(nil)
+			msg, seg, err := capnp.NewMessage(arena)
+			require.NoError(b, err)
+			bytesMsg, err := bCapnProto.NewBytesMsg(seg)
+			require.NoError(b, err)
+			require.NoError(b, bytesMsg.SetKey(bytes))
+			for writer.Size() < uint64(bm.fileSize) {
+				_, _ = writer.Write(msg)
+			}
+			b.SetBytes(int64(writer.Size()))
+			assert.NoError(b, writer.Close())
+
+			b.ResetTimer()
+			for n := 0; n < b.N; n++ {
+				reader, err := rCapnProto.NewReader(rCapnProto.ReaderPath(tmpFile.Name()))
+				assert.NoError(b, err)
+				assert.NoError(b, reader.Open())
+
+				for {
+					msg, err := reader.ReadNext()
+					if errors.Is(err, io.EOF) {
+						break
+					}
+
+					_, err = bCapnProto.ReadRootBytesMsg(msg)
+					require.NoError(b, err)
+				}
+
+				assert.NoError(b, reader.Close())
+			}
+		})
+	}
+
+}

go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/steakknife/hamming v0.0.0-20180906055917-c99c65617cd3 // indirect
+	golang.org/x/sync v0.7.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 

go.sum

@@ -10,6 +10,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/ncw/directio v1.0.5 h1:JSUBhdjEvVaJvOoyPAbcW0fnd0tvRXD76wEfZ1KcQz4=
 github.com/ncw/directio v1.0.5/go.mod h1:rX/pKEYkOXBGOggmcyJeJGloCkleSvphPx2eV3t6ROk=
+github.com/philhofer/fwd v1.1.2 h1:bnDivRJ1EWPjUIRXV5KfORO897HTbpFAQddBdE8t7Gw=
+github.com/philhofer/fwd v1.1.2/go.mod h1:qkPdfjR2SIEbspLqpe1tO4n5yICnr2DY7mqEx2tUTP0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/steakknife/bloomfilter v0.0.0-20180922174646-6819c0d2a570 h1:gIlAHnH1vJb5vwEjIp5kBj/eu99p/bl0Ay2goiPe5xE=
@@ -18,10 +20,16 @@ github.com/steakknife/hamming v0.0.0-20180906055917-c99c65617cd3 h1:njlZPzLwU639
 github.com/steakknife/hamming v0.0.0-20180906055917-c99c65617cd3/go.mod h1:hpGUWaI9xL8pRQCTXQgocU38Qw1g0Us7n5PxxTwTCYU=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/tinylib/msgp v1.1.9 h1:SHf3yoO2sGA0veCJeCBYLHuttAVFHGm2RHgNodW7wQU=
+github.com/tinylib/msgp v1.1.9/go.mod h1:BCXGB54lDD8qUEPmiG0cQQUANC4IUQyB2ItS2UDlO/k=
+github.com/tj/assert v0.0.3 h1:Df/BlaZ20mq6kuai7f5z2TvPFiwC3xaWJSDQNiIS3Rk=
+github.com/tj/assert v0.0.3/go.mod h1:Ne6X72Q+TB1AteidzQncjw9PabbMp4PBMZ1k+vd1Pvk=
 github.com/tjungblu/porcupine v0.0.0-20221116095144-377185aa0569 h1:acDBvgSBtnyBidmpmTEgxStjXeuYyfG3fF72khP24/Y=
 github.com/tjungblu/porcupine v0.0.0-20221116095144-377185aa0569/go.mod h1:+z336r1WR0gcwl1ALfoNBpDTCW06vO5DzBwunEcSvcs=
 golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8 h1:yixxcjnhBmY0nkL253HFVIm0JsFHwrHdT3Yh6szTnfY=
 golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8/go.mod h1:jj3sYF3dwk5D+ghuXyeI3r5MFf+NT2An6/9dOA95KSI=
+golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
+golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
 google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=

recordio/capnproto/capnproto_reader.go

@@ -0,0 +1,85 @@
+package capnproto
+
+import (
+	"capnproto.org/go/capnp/v3"
+	"errors"
+	"os"
+
+	"github.com/thomasjungblut/go-sstables/recordio"
+)
+
+type Reader struct {
+	recordio.ReaderI
+}
+
+func (r *Reader) ReadNext() (*capnp.Message, error) {
+	bytes, err := r.ReaderI.ReadNext()
+	if err != nil {
+		return nil, err
+	}
+
+	return capnp.Unmarshal(bytes)
+}
+
+// options
+
+type ReaderOptions struct {
+	path         string
+	file         *os.File
+	bufSizeBytes int
+}
+
+type ReaderOption func(*ReaderOptions)
+
+func ReaderPath(p string) ReaderOption {
+	return func(args *ReaderOptions) {
+		args.path = p
+	}
+}
+
+func ReaderFile(p *os.File) ReaderOption {
+	return func(args *ReaderOptions) {
+		args.file = p
+	}
+}
+
+func ReadBufferSizeBytes(p int) ReaderOption {
+	return func(args *ReaderOptions) {
+		args.bufSizeBytes = p
+	}
+}
+
+// NewReader creates a new reader with the given options. Either Path or File must be supplied
+func NewReader(readerOptions ...ReaderOption) (ReaderI, error) {
+	opts := &ReaderOptions{
+		path:         "",
+		file:         nil,
+		bufSizeBytes: 1024 * 1024 * 4,
+	}
+
+	for _, readerOption := range readerOptions {
+		readerOption(opts)
+	}
+
+	if (opts.file != nil) && (opts.path != "") {
+		return nil, errors.New("either os.File or string path must be supplied, never both")
+	}
+
+	if opts.file == nil {
+		if opts.path == "" {
+			return nil, errors.New("path was not supplied")
+		}
+	}
+	reader, err := recordio.NewFileReader(
+		recordio.ReaderPath(opts.path),
+		recordio.ReaderFile(opts.file),
+		recordio.ReaderBufferSizeBytes(opts.bufSizeBytes))
+	if err != nil {
+		return nil, err
+	}
+
+	return &Reader{
+		reader,
+	}, nil
+
+}