feat: data wrapper to wrap the raw value in the underlying storage

Makefile

@@ -6,6 +6,7 @@ API_DIR=api
 V=v1
 GEN_DIR=${API_DIR}/server/${V}
 PROTO_DIR=${API_DIR}/proto/server/${V}
+DATA_PROTO_DIR=internal
 
 # Needed to be able to build amd64 binaries on MacOS M1
 DOCKER_PLATFORM="linux/amd64"
@@ -40,7 +41,13 @@ ${API_DIR}/client/${V}/%/http.go: ${PROTO_DIR}/%_openapi.yaml
 		-o ${API_DIR}/client/${V}/$(*F)/http.go \
 		${PROTO_DIR}/$(*F)_openapi.yaml
 
-generate: ${GEN_DIR}/api.pb.go ${GEN_DIR}/api.pb.gw.go ${GEN_DIR}/health.pb.go ${GEN_DIR}/health.pb.gw.go
+${DATA_PROTO_DIR}/%.pb.go ${DATA_PROTO_DIR}/%.pb.gw.go: ${DATA_PROTO_DIR}/%.proto
+	protoc -Iinternal \
+		--go_out=${DATA_PROTO_DIR} --go_opt=paths=source_relative \
+		--go-grpc_out=${DATA_PROTO_DIR} --go-grpc_opt=paths=source_relative \
+		$<
+
+generate: ${GEN_DIR}/api.pb.go ${GEN_DIR}/api.pb.gw.go ${GEN_DIR}/health.pb.go ${GEN_DIR}/health.pb.gw.go ${DATA_PROTO_DIR}/data.pb.gw.go
 
 test_client: ${API_DIR}/client/${V}/api/http.go
 

go.mod

@@ -69,6 +69,7 @@ require (
 	github.com/spf13/cast v1.4.1 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/subosito/gotenv v1.2.0 // indirect
+	github.com/ugorji/go/codec v1.2.7 // indirect
 	github.com/valyala/fasthttp v1.34.0 // indirect
 	github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f // indirect
 	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect

go.sum

@@ -383,8 +383,12 @@ github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69
 github.com/tigrisdata/tigrisdb-client-go v1.0.0-alpha.4 h1:bxmxXRVzsD9vGlFgfpMe1WtWB0wzUV2rE66ir7wacdY=
 github.com/tigrisdata/tigrisdb-client-go v1.0.0-alpha.4/go.mod h1:lP4RhNml4kRHULYY+nb+nmvGz0THlwUx9w5BI7Qus7I=
 github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
+github.com/ugorji/go v1.2.6/go.mod h1:anCg0y61KIhDlPZmnH+so+RQbysYVyDko0IMgJv0Nn0=
+github.com/ugorji/go v1.2.7 h1:qYhyWUUd6WbiM+C6JZAUkIJt/1WrjzNHY9+KCIjVqTo=
 github.com/ugorji/go v1.2.7/go.mod h1:nF9osbDWLy6bDVv/Rtoh6QgnvNDpmCalQV5urGCCS6M=
 github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
+github.com/ugorji/go/codec v1.2.6/go.mod h1:V6TCNZ4PHqoHGFZuSG1W8nrCzzdgA2DozYxWFFpvxTw=
+github.com/ugorji/go/codec v1.2.7 h1:YPXUKf7fYbp/y8xloBqZOw2qaVggbfwMlI8WM3wZUJ0=
 github.com/ugorji/go/codec v1.2.7/go.mod h1:WGN1fab3R1fzQlVQTkfxVtIBhWDRqOviHU95kRgeqEY=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=

internal/data.go

@@ -0,0 +1,92 @@
+// Copyright 2022 Tigris Data, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+	"bytes"
+	"time"
+
+	api "github.com/tigrisdata/tigrisdb/api/server/v1"
+	ulog "github.com/tigrisdata/tigrisdb/util/log"
+	"github.com/ugorji/go/codec"
+	"google.golang.org/grpc/codes"
+)
+
+var (
+	bh codec.BincHandle
+)
+
+// DataType is to define the different data types for the data stored in the storage engine.
+type DataType byte
+
+// Note: Do not change the order. Order is important because encoder is adding the type as the first byte. Check the
+// Encode/Decode method to see how it is getting used.
+const (
+	Unknown DataType = iota
+	TableDataType
+)
+
+func NewTimestamp() *Timestamp {
+	return &Timestamp{
+		Nano: time.Now().UnixNano(),
+	}
+}
+
+// NewTableData returns a table data type by setting the ts to the current value.
+func NewTableData(data []byte) *TableData {
+	return &TableData{
+		Ts:      NewTimestamp(),
+		RawData: data,
+	}
+}
+
+// Encode is used to encode data to the raw bytes which is used to store in storage as value. The first byte is storing
+// the type corresponding to this Data. This is important and used by the decoder later to decode back.
+func Encode(data *TableData) ([]byte, error) {
+	var buf bytes.Buffer
+	// this is added so that we can evolve the DataTypes and have more dataTypes in future
+	err := buf.WriteByte(byte(TableDataType))
+	if err != nil {
+		return nil, err
+	}
+	enc := codec.NewEncoder(&buf, &bh)
+	if err := enc.Encode(data); ulog.E(err) {
+		return nil, err
+	}
+
+	return buf.Bytes(), nil
+}
+
+// Decode is used to decode the raw bytes to TableData. The raw bytes are returned from the storage and the kvStore is
+// calling Decode to convert these raw bytes back to TableData.
+func Decode(b []byte) (*TableData, error) {
+	dataType := DataType(b[0])
+	return decodeInternal(dataType, b[1:])
+}
+
+func decodeInternal(dataType DataType, encoded []byte) (*TableData, error) {
+	dec := codec.NewDecoderBytes(encoded, &bh)
+
+	switch dataType {
+	case TableDataType:
+		var v *TableData
+		if err := dec.Decode(&v); err != nil {
+			return nil, err
+		}
+		return v, nil
+	}
+
+	return nil, api.Errorf(codes.Internal, "unable to decode '%v'", dataType)
+}

internal/data.proto

@@ -0,0 +1,34 @@
+// Copyright 2022 Tigris Data, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+option go_package = "github.com/tigrisdata/tigrisdb/internal";
+
+message Timestamp {
+  int64 nano=1;
+}
+
+// TableData is a wrapper around underlying storage raw bytes. This is used to store user collection data, internal
+// metadata, schemas etc. Caller can use raw_data to store any raw bytes.
+message TableData {
+  // ver is the version for the raw bytes, this may be schema version in case of user data.
+  int32 ver = 1;
+  // encoding represents encoding of the data field.
+  int32 encoding = 2;
+  // ts represents the commit timestamp of the row.
+  Timestamp ts = 3;
+  // raw_data is the raw bytes stored, caller controls how they want to store these raw bytes in database.
+  bytes raw_data = 4;
+}

internal/data_test.go

@@ -0,0 +1,53 @@
+// Copyright 2022 Tigris Data, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	api "github.com/tigrisdata/tigrisdb/api/server/v1"
+	"google.golang.org/grpc/codes"
+)
+
+func TestEncode_Decode(t *testing.T) {
+	t.Run("table_data", func(t *testing.T) {
+		d := NewTableData([]byte(`{"a": 1, "b": "foo"}`))
+		encoded, err := Encode(d)
+		require.NoError(t, err)
+		require.NotNil(t, encoded)
+
+		data, err := Decode(encoded)
+		require.NoError(t, err)
+		require.Equal(t, d, data)
+	})
+	t.Run("not_implemented", func(t *testing.T) {
+		data, err := Decode([]byte(`{"a": 1, "b": "foo"}`))
+		require.Equal(t, api.Error(codes.Internal, "unable to decode '123'"), err)
+		require.Nil(t, data)
+	})
+
+}
+
+func Benchmark_MsgPack(b *testing.B) {
+	v := &TableData{
+		RawData: []byte(`"K1": "vK1", "K2": 1, "D1": "vD1", "random", "this is a long string, with many characters"}`),
+	}
+	for n := 0; n < b.N; n++ {
+		enc, err := Encode(v)
+		require.NoError(b, err)
+		require.NotNil(b, enc)
+	}
+}

query/aggregation/aggregation.go

@@ -49,7 +49,7 @@ func Unmarshal(input jsoniter.RawMessage) (expression.Expr, error) {
 	return expression.Unmarshal(input, UnmarshalAggObject)
 }
 
-// UnmarshalObject unmarshals the input to the aggregation. Note the return after the first check, this is mainly
+// UnmarshalAggObject unmarshal the input to the aggregation. Note the return after the first check, this is mainly
 // because an aggregation object can have nested objects but top level it will be one expression.
 func UnmarshalAggObject(input jsoniter.RawMessage) (expression.Expr, error) {
 	var err error

server/main.go

@@ -38,13 +38,13 @@ func main() {
 
 	log.Info().Str("version", Version).Str("BuildHash", BuildHash).Msgf("Starting server")
 
-	kv, err := kv.NewFoundationDB(&config.DefaultConfig.FoundationDB)
+	kvStore, err := kv.NewKeyValueStore(&config.DefaultConfig.FoundationDB)
 	if err != nil {
 		log.Fatal().Err(err).Msg("error initializing kv store")
 	}
 
 	mx := muxer.NewMuxer(&config.DefaultConfig)
-	mx.RegisterServices(kv)
+	mx.RegisterServices(kvStore)
 	if err := mx.Start(config.DefaultConfig.Server.Host, config.DefaultConfig.Server.Port); err != nil {
 		log.Fatal().Err(err).Msgf("error starting server")
 	}

server/metadata/encoding/dictionary.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/tigrisdata/tigrisdb/internal"
+
 	"github.com/rs/zerolog/log"
 	api "github.com/tigrisdata/tigrisdb/api/server/v1"
 	"github.com/tigrisdata/tigrisdb/keys"
@@ -133,8 +135,9 @@ func (r *reservedSubspace) reload(ctx context.Context, tx transaction.Tx) error
 			return api.Errorf(codes.Internal, "unable to deduce the encoded key from fdb key %T", allocatedTo)
 		}
 
-		r.allocated[ByteToUInt32(row.Value)] = allocatedTo.(string)
-		r.namespaceToId[allocatedTo.(string)] = ByteToUInt32(row.Value)
+		allocatedValue := ByteToUInt32(row.Data.RawData)
+		r.allocated[allocatedValue] = allocatedTo.(string)
+		r.namespaceToId[allocatedTo.(string)] = allocatedValue
 	}
 
 	return it.Err()
@@ -162,7 +165,7 @@ func (r *reservedSubspace) reserveNamespace(ctx context.Context, tx transaction.
 
 	key := keys.NewKey(ReservedSubspaceKey, namespaceKey, namespace, keyEnd)
 	// now do an insert to fail if namespace already exists.
-	if err := tx.Insert(ctx, key, UInt32ToByte(id)); err != nil {
+	if err := tx.Insert(ctx, key, internal.NewTableData(UInt32ToByte(id))); err != nil {
 		log.Debug().Interface("key", key).Uint32("value", id).Err(err).Msg("reserving namespace failed")
 		return err
 	}
@@ -181,14 +184,14 @@ func (r *reservedSubspace) allocateToken(ctx context.Context, tx transaction.Tx,
 	newReservedValue := reservedBaseValue
 	var row kv.KeyValue
 	if it.Next(&row) {
-		newReservedValue = ByteToUInt32(row.Value) + 1
+		newReservedValue = ByteToUInt32(row.Data.RawData) + 1
 	}
 
 	if err := it.Err(); err != nil {
 		return 0, err
 	}
 
-	if err := tx.Replace(ctx, key, UInt32ToByte(newReservedValue)); err != nil {
+	if err := tx.Replace(ctx, key, internal.NewTableData(UInt32ToByte(newReservedValue))); err != nil {
 		log.Debug().Str("key", key.String()).Uint32("value", newReservedValue).Msg("allocating token failed")
 		return 0, err
 	}
@@ -329,7 +332,7 @@ func (k *DictionaryEncoder) encodeAsDropped(ctx context.Context, tx transaction.
 	log.Debug().Str("key", toDeleteKey.String()).Str("type", encName).Msg("existing entry deletion succeed")
 
 	// now do insert because we need to fail if token is already assigned
-	if err := tx.Replace(ctx, newKey, UInt32ToByte(newValue)); err != nil {
+	if err := tx.Replace(ctx, newKey, internal.NewTableData(UInt32ToByte(newValue))); err != nil {
 		log.Debug().Str("key", newKey.String()).Uint32("value", newValue).Err(err).Str("type", encName).Msg("encoding failed")
 		return err
 	}
@@ -345,7 +348,7 @@ func (k *DictionaryEncoder) encode(ctx context.Context, tx transaction.Tx, key k
 	}
 
 	// now do insert because we need to fail if token is already assigned
-	if err := tx.Insert(ctx, key, UInt32ToByte(reserveToken)); err != nil {
+	if err := tx.Insert(ctx, key, internal.NewTableData(UInt32ToByte(reserveToken))); err != nil {
 		log.Debug().Str("type", encName).Str("key", key.String()).Uint32("value", reserveToken).Err(err).Msg("encoding failed for")
 		return InvalidId, err
 	}
@@ -402,11 +405,11 @@ func (k *DictionaryEncoder) GetDatabases(ctx context.Context, tx transaction.Tx,
 
 			if end == keyDroppedEnd {
 				log.Debug().Str("database", name).Msg("dropped database found, ignoring")
-				droppedDatabase[name] = ByteToUInt32(v.Value)
+				droppedDatabase[name] = ByteToUInt32(v.Data.RawData)
 				continue
 			}
 
-			databases[name] = ByteToUInt32(v.Value)
+			databases[name] = ByteToUInt32(v.Data.RawData)
 		}
 	}
 
@@ -450,11 +453,11 @@ func (k *DictionaryEncoder) GetCollections(ctx context.Context, tx transaction.T
 
 			if end == keyDroppedEnd {
 				log.Debug().Str("collection", name).Msg("dropped collection found, ignoring")
-				droppedCollection[name] = ByteToUInt32(v.Value)
+				droppedCollection[name] = ByteToUInt32(v.Data.RawData)
 				continue
 			}
 
-			collections[name] = ByteToUInt32(v.Value)
+			collections[name] = ByteToUInt32(v.Data.RawData)
 		}
 	}
 
@@ -498,11 +501,11 @@ func (k *DictionaryEncoder) GetIndexes(ctx context.Context, tx transaction.Tx, n
 
 			if end == keyDroppedEnd {
 				log.Debug().Str("index", name).Msg("dropped index found, ignoring")
-				droppedIndexes[name] = ByteToUInt32(v.Value)
+				droppedIndexes[name] = ByteToUInt32(v.Data.RawData)
 				continue
 			}
 
-			indexes[name] = ByteToUInt32(v.Value)
+			indexes[name] = ByteToUInt32(v.Data.RawData)
 		}
 	}
 
@@ -541,7 +544,7 @@ func (k *DictionaryEncoder) getId(ctx context.Context, tx transaction.Tx, key ke
 
 	var row kv.KeyValue
 	if it.Next(&row) {
-		return ByteToUInt32(row.Value), nil
+		return ByteToUInt32(row.Data.RawData), nil
 	}
 
 	if err := it.Err(); err != nil {