ronantakizawa/codereview-bench huggingface dataset:
Repository: codenotary/immudb
File: cmd/immuclient/immuc/misc.go (Go)

Code before:
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 immuc

import (
	"context"
	"strings"

	"github.com/codenotary/immudb/pkg/api/schema"
	"github.com/codenotary/immudb/pkg/client"
)

func (i *immuc) HealthCheck(args []string) (string, error) {
	ctx := context.Background()

	if _, err := i.Execute(func(immuClient client.ImmuClient) (interface{}, error) {
		_, err := immuClient.ServerInfo(ctx, &schema.ServerInfoRequest{})
		return nil, err
	}); err != nil {
		rpcerrors := strings.SplitAfter(err.Error(), "=")

		if len(rpcerrors) > 1 {
			return rpcerrors[len(rpcerrors)-1], nil
		}

		return "", err
	}

	return "Health check OK", nil
}


Review comment (suggestion):
This one should also still use old, deprecated `HealthCheck` call

Code after:
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 immuc

import (
	"context"
	"strings"

	"github.com/codenotary/immudb/pkg/client"
)

func (i *immuc) HealthCheck(args []string) (string, error) {
	ctx := context.Background()

	if _, err := i.Execute(func(immuClient client.ImmuClient) (interface{}, error) {
		return nil, immuClient.HealthCheck(ctx)
	}); err != nil {
		rpcerrors := strings.SplitAfter(err.Error(), "=")

		if len(rpcerrors) > 1 {
			return rpcerrors[len(rpcerrors)-1], nil
		}

		return "", err
	}

	return "Health check OK", nil
}
Repository: codenotary/immudb
File: embedded/store/immustore.go (Go)

Code before:
			return nil, err
		}
	}

	// early check to reduce amount of garbage when tx is not finally committed
	s.mutex.Lock()

	if s.closed {
		s.mutex.Unlock()
		return nil, ErrAlreadyClosed
	}

	if !otx.IsWriteOnly() && otx.snap.Ts() <= s.lastPreCommittedTxID() {
		s.mutex.Unlock()
		return nil, ErrTxReadConflict
	}

	s.mutex.Unlock()

	var ts int64
	var blTxID uint64
	var version int

	if expectedHeader == nil {
		ts = s.timeFunc().Unix()
		version = s.writeTxHeaderVersion
	} else {
		ts = expectedHeader.Ts
		blTxID = expectedHeader.BlTxID
		version = expectedHeader.Version

		//TxHeader is validated against current store

		currTxID, currAlh := s.Alh()

		var blRoot [sha256.Size]byte

		if blTxID > 0 {
			blRoot, err = s.aht.RootAt(blTxID)
			if err != nil && err != ahtree.ErrEmptyTree {
				return nil, err
			}
		}

		if otx.metadata != nil {
			if !otx.metadata.Equal(expectedHeader.Metadata) {
				return nil, ErrIllegalArguments
			}
		} else if expectedHeader.Metadata != nil {
			if !expectedHeader.Metadata.Equal(otx.metadata) {
				return nil, ErrIllegalArguments

Review comment (bug):
it may be better to leave only the header version out of the critical section. Moving timestamp and validations during replicated commit inside it, to prevent potential edge cases when switching between master and replica modes. Please take a look at the commit in branch `fix/bltx-ordering-rep` as reference

Code after:
		if err != nil {
			return nil, err
		}
	}

	// early check to reduce amount of garbage when tx is not finally committed
	s.mutex.Lock()

	if s.closed {
		s.mutex.Unlock()
		return nil, ErrAlreadyClosed
	}

	if !otx.IsWriteOnly() && otx.snap.Ts() <= s.lastPreCommittedTxID() {
		s.mutex.Unlock()
		return nil, ErrTxReadConflict
	}

	s.mutex.Unlock()

	var version int

	if expectedHeader == nil {
		version = s.writeTxHeaderVersion
	} else {
		version = expectedHeader.Version

		// TxHeader is validated against current store

		currTxID, currAlh := s.Alh()

		if currTxID != expectedHeader.ID-1 {
			return nil, ErrTxAlreadyCommitted
		}

		var blRoot [sha256.Size]byte

		if expectedHeader.BlTxID > 0 {
			blRoot, err = s.aht.RootAt(expectedHeader.BlTxID)
			if err != nil && err != ahtree.ErrEmptyTree {
				return nil, err
			}
		}

		if otx.metadata != nil {
			if !otx.metadata.Equal(expectedHeader.Metadata) {
				return nil, ErrIllegalArguments
			}
		} else if expectedHeader.Metadata != nil {
			if !expectedHeader.Metadata.Equal(otx.metadata) {
				return nil, ErrIllegalArguments
Repository: codenotary/immudb
File: pkg/client/clienttest/token_service_mock.go (Go)

Code before:
/*
Copyright 2022 Codenotary Inc. All rights reserved.

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 clienttest

import (
	"github.com/codenotary/immudb/pkg/client/homedir"
	"github.com/codenotary/immudb/pkg/client/tokenservice"
)

var (
	_ tokenservice.TokenService = &TokenServiceMock{}
)

type TokenServiceMock struct {
	tokenservice.TokenService
	GetTokenF       func() (string, error)
	SetTokenF       func(database string, token string) error
	IsTokenPresentF func() (bool, error)
	DeleteTokenF    func() error
}

func (ts TokenServiceMock) GetToken() (string, error) {
	return ts.GetTokenF()
}

func (ts TokenServiceMock) SetToken(database string, token string) error {
	return ts.SetTokenF(database, token)
}

func (ts TokenServiceMock) DeleteToken() error {
	return ts.DeleteTokenF()
}

func (ts TokenServiceMock) IsTokenPresent() (bool, error) {
	return ts.IsTokenPresentF()
}

Review comment (suggestion):
```suggestion
	_ tokenservice.TokenService = (*TokenServiceMock)(nil)
```

Code after:
/*
Copyright 2022 Codenotary Inc. All rights reserved.

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 clienttest

import (
	"github.com/codenotary/immudb/pkg/client/homedir"
	"github.com/codenotary/immudb/pkg/client/tokenservice"
)

var (
	_ tokenservice.TokenService = (*TokenServiceMock)(nil)
)

type TokenServiceMock struct {
	tokenservice.TokenService
	GetTokenF       func() (string, error)
	SetTokenF       func(database string, token string) error
	IsTokenPresentF func() (bool, error)
	DeleteTokenF    func() error
}

func (ts TokenServiceMock) GetToken() (string, error) {
	return ts.GetTokenF()
}

func (ts TokenServiceMock) SetToken(database string, token string) error {
	return ts.SetTokenF(database, token)
}

func (ts TokenServiceMock) DeleteToken() error {
	return ts.DeleteTokenF()
}

func (ts TokenServiceMock) IsTokenPresent() (bool, error) {
	return ts.IsTokenPresentF()
}
Repository: codenotary/immudb
File: pkg/integration/replication/docker.go (Go)

Code before:
				Source: "/tmp",
				Target: os.Getenv("TMPDIR"),
				Type:   "bind",
			},
		}
	})
	require.NoError(t, err)
	return resource
}

func (s *dockerTestServer) Start(t *testing.T) {
	require.Nil(t, s.srv)

	var (
		name     = fmt.Sprintf("immudb-%d", rand.Intn(50))
		repo     = "codenotary/immudb"
		tag      = "latest"
		dirFlag  = fmt.Sprintf("--dir=%s", s.dir)
		hostPort = ""
	)

	if s.port > 0 {
		hostPort = strconv.Itoa(s.port)
	}

	container := s.startContainer(t, &dockertest.RunOptions{
		Name:       name,
		Repository: repo,
		Tag:        tag,
		Cmd: []string{
			dirFlag,
			"--pgsql-server=false",
			"--metrics-server=false",
		},
		ExposedPorts: []string{"3322"},
		PortBindings: map[docker.Port][]docker.PortBinding{"3322/tcp": {{HostPort: hostPort}}},
	})

	port, err := strconv.Atoi(container.GetPort("3322/tcp"))
	require.Nil(t, err)

	s.srv = container
	s.port = port

	// Wait for the server to initialize
	// TODO: Active notification that the server has started
	time.Sleep(5 * time.Second)

}

// getLocalIP returns the non loopback local IP of the host

Review comment (suggestion):
I  think we should also bind data volume here from the temporary folder - that will be needed when restarting container by completely recreating the docker container (e.g. after shutting it down forcibly)

Code after:
				Source: "/tmp",
				Target: os.Getenv("TMPDIR"),
				Type:   "bind",
			},
		}
	})
	require.NoError(t, err)
	return resource
}

func (s *dockerTestServer) Start(t *testing.T) {
	require.Nil(t, s.srv)

	var (
		name     = fmt.Sprintf("immudb-%d", rand.Intn(50))
		repo     = "immudb/e2e"
		tag      = "latest"
		dirFlag  = fmt.Sprintf("--dir=%s", s.dir)
		hostPort = ""
	)

	if s.port > 0 {
		hostPort = strconv.Itoa(s.port)
	}

	container := s.startContainer(t, &dockertest.RunOptions{
		Name:       name,
		Repository: repo,
		Tag:        tag,
		Cmd: []string{
			dirFlag,
			"--pgsql-server=false",
			"--metrics-server=false",
		},
		ExposedPorts: []string{"3322"},
		PortBindings: map[docker.Port][]docker.PortBinding{"3322/tcp": {{HostPort: hostPort}}},
	})

	port, err := strconv.Atoi(container.GetPort("3322/tcp"))
	require.Nil(t, err)

	s.srv = container
	s.port = port

	// Wait for the server to initialize
	// TODO: Active notification that the server has started
	time.Sleep(5 * time.Second)

}

// getLocalIP returns the non loopback local IP of the host
Repository: codenotary/immudb
File: pkg/client/types.go (Go)

Code before:

// Deprecated: will be removed in future versions.
func (c *immuClient) WithTokenService(tokenService tokenservice.TokenService) *immuClient {
	c.Tkns = tokenService
	return c
}

// WithServerSigningPubKey sets up public key for server's state validation.
func (c *immuClient) WithServerSigningPubKey(publicKey *ecdsa.PublicKey) *immuClient {
	c.serverSigningPubKey = publicKey
	return c
}

// WithStreamServiceFactory sets up stream factory for the client.
func (c *immuClient) WithStreamServiceFactory(ssf stream.ServiceFactory) *immuClient {
	c.StreamServiceFactory = ssf
	return c
}

// WithOptions sets up client options for the instance.
func (c *immuClient) WithOptions(options *Options) *immuClient {
	c.Options = options
	return c
}

func (c *immuClient) WithErrorHandler(handler heartbeater.ErrorHandler) *immuClient {
	c.errorHandler = handler
	return c
}


Review comment (suggestion):
error handler should not be in package heartbeater, in the future more client errors can be handled and not just the ones related to heartbeats

Code after:
// Deprecated: will be removed in future versions.
func (c *immuClient) WithTokenService(tokenService tokenservice.TokenService) *immuClient {
	c.Tkns = tokenService
	return c
}

// WithServerSigningPubKey sets up public key for server's state validation.
func (c *immuClient) WithServerSigningPubKey(publicKey *ecdsa.PublicKey) *immuClient {
	c.serverSigningPubKey = publicKey
	return c
}

// WithStreamServiceFactory sets up stream factory for the client.
func (c *immuClient) WithStreamServiceFactory(ssf stream.ServiceFactory) *immuClient {
	c.StreamServiceFactory = ssf
	return c
}

// WithOptions sets up client options for the instance.
func (c *immuClient) WithOptions(options *Options) *immuClient {
	c.Options = options
	return c
}

func (c *immuClient) WithErrorHandler(handler ErrorHandler) *immuClient {
	c.errorHandler = handler
	return c
}
Repository: codenotary/immudb
File: embedded/ahtree/ahtree.go (Go)

Code before:
		return err
	}

	cLogSize := int64(newSize * cLogEntrySize)
	pLogSize := int64(0)
	dLogSize := int64(0)

	if newSize > 0 {
		var b [cLogEntrySize]byte
		_, err := t.cLog.ReadAt(b[:], cLogSize-cLogEntrySize)
		if err != nil {
			return err
		}

		pOff := binary.BigEndian.Uint64(b[:])
		pSize := binary.BigEndian.Uint32(b[offsetSize:])

		pLogSize = int64(pOff) + int64(szSize+pSize)

		pLogFileSize, err := t.pLog.Size()
		if err != nil {
			return err
		}

		if pLogFileSize < pLogSize {
			return ErrorCorruptedData
		}

		dLogSize = int64(nodesUpto(uint64(cLogSize/cLogEntrySize)) * sha256.Size)

		dLogFileSize, err := t.dLog.Size()
		if err != nil {
			return err
		}

		if dLogFileSize < dLogSize {
			return ErrorCorruptedDigests
		}
	}

	// Invalidate caches
	for i := cLogSize; i < t.cLogSize; i += cLogEntrySize {
		t.pCache.Pop(uint64(i / cLogEntrySize))
	}
	for i := dLogSize; i < t.dLogSize; i += sha256.Size {
		t.dCache.Pop(uint64(i / sha256.Size))
	}

	t.cLogSize = cLogSize
	t.pLogSize = pLogSize
	t.dLogSize = dLogSize

Review comment (bug):
just a comment on what this bug was, and why this change

Code after:
	}

	err := t.sync()
	if err != nil {
		return err
	}

	cLogSize := int64(newSize * cLogEntrySize)
	pLogSize := int64(0)
	dLogSize := int64(0)

	if newSize > 0 {
		var b [cLogEntrySize]byte
		_, err := t.cLog.ReadAt(b[:], cLogSize-cLogEntrySize)
		if err != nil {
			return err
		}

		pOff := binary.BigEndian.Uint64(b[:])
		pSize := binary.BigEndian.Uint32(b[offsetSize:])

		// pOff denotes the latest payload
		// pSize denotes the size of the latest payload
		// as payloads are prefixed with the size when written into pLog
		// pLogSize is calculated with the offset, the size description of the payload and the payload itself
		pLogSize = int64(pOff) + int64(szSize+pSize)

		pLogFileSize, err := t.pLog.Size()
		if err != nil {
			return err
		}

		if pLogFileSize < pLogSize {
			return ErrorCorruptedData
		}

		dLogSize = int64(nodesUpto(uint64(cLogSize/cLogEntrySize)) * sha256.Size)

		dLogFileSize, err := t.dLog.Size()
		if err != nil {
			return err
		}

		if dLogFileSize < dLogSize {
			return ErrorCorruptedDigests
		}
	}

	// Invalidate caches
	for i := cLogSize; i < t.cLogSize; i += cLogEntrySize {
		t.pCache.Pop(uint64(i / cLogEntrySize))
Repository: codenotary/immudb
File: test/performance-test-suite/pkg/benchmarks/writetxs/benchmark.go (Go)

Code before:
	b.server.Server.Srv.WithLogger(logger.NewMemoryLoggerWithLevel(logger.LogDebug))

	err = b.server.Start()
	if err != nil {
		return err
	}

	if b.cfg.Replica == "async" || b.cfg.Replica == "sync" {
		replicaDirName := fmt.Sprintf("%s-replica-tx-test", b.cfg.Replica)

		replicaOptions := server.
			ReplicationOptions{}

		options2 := server.
			DefaultOptions().
			WithDir(replicaDirName).
			WithLogFormat(logger.LogFormatJSON)

		if b.cfg.Replica == "async" {
			options2.
				WithReplicationOptions(replicaOptions.WithIsReplica(true))
		}

		if b.cfg.Replica == "sync" {
			options2.
				WithReplicationOptions(replicaOptions.WithIsReplica(true).WithSyncReplication(true))
		}

		b.replicaServer = servertest.NewBufconnServer(options2)
		b.replicaServer.Server.Srv.WithLogger(logger.NewMemoryLoggerWithLevel(logger.LogDebug))

		err = b.replicaServer.Start()
		if err != nil {
			return err
		}
	}

	b.clients = []client.ImmuClient{}
	for i := 0; i < b.cfg.Workers; i++ {
		c, err := b.server.NewAuthenticatedClient(client.DefaultOptions())
		if err != nil {
			return err
		}
		b.clients = append(b.clients, c)
	}

	return nil
}

func (b *benchmark) Cleanup() error {


Review comment (suggestion):
primary server also needs to be configured for sync replication

e.g. `WithReplicationOptions(replicaOptions.WithIsReplica(false).WithSyncReplication(true).WithSyncAcks(1))`

Code after:
		WithLogFormat(logger.LogFormatJSON).
		WithLogfile("./immudb.log")

	primaryServerReplicaOptions := server.ReplicationOptions{}

	if b.cfg.Replica == "async" {
		primaryServerOpts.WithReplicationOptions(primaryServerReplicaOptions.WithIsReplica(false))
	}

	if b.cfg.Replica == "sync" {
		primaryServerOpts.WithReplicationOptions(primaryServerReplicaOptions.WithIsReplica(false).WithSyncReplication(true).WithSyncAcks(1))
	}

	b.primaryServer = server.DefaultServer().WithOptions(primaryServerOpts).(*server.ImmuServer)

	err = b.primaryServer.Initialize()
	if err != nil {
		return err
	}

	go func() {
		b.primaryServer.Start()
	}()

	time.Sleep(1 * time.Second)

	primaryPort := b.primaryServer.Listener.Addr().(*net.TCPAddr).Port

	if b.cfg.Replica == "async" || b.cfg.Replica == "sync" {
		replicaPath, err := os.MkdirTemp("", fmt.Sprintf("%s-tx-test-replica", b.cfg.Replica))
		if err != nil {
			return err
		}

		defer os.RemoveAll(replicaPath)

		replicaServerOptions := server.
			DefaultOptions().
			WithDir(replicaPath).
			WithPort(0).
			WithLogFormat(logger.LogFormatJSON).
			WithLogfile("./replica.log")

		replicaServerOptions.PgsqlServer = false
		replicaServerOptions.MetricsServer = false
		replicaServerOptions.WebServer = false

		replicaServerReplicaOptions := server.ReplicationOptions{}

		replicaServerReplicaOptions.PrimaryHost = "127.0.0.1"
		replicaServerReplicaOptions.PrimaryPort = primaryPort
Repository: codenotary/immudb
File: embedded/store/immustore.go (Go)

Code before:

	if entry.vLen == 0 {
		return nil, nil
	}

	b := make([]byte, entry.vLen)

	_, err := s.readValueAt(b, entry.vOff, entry.hVal, false)
	if err != nil {
		return nil, err
	}

	return b, nil
}

func (s *ImmuStore) readValueAt(b []byte, off int64, hvalue [sha256.Size]byte, skipIntegrityCheck bool) (n int, err error) {
	if s.vLogCache != nil {
		val, err := s.vLogCache.Get(off)
		if err == nil {
			// the requested value was found in the value cache
			bval := val.([]byte)
			n := len(bval)

			copy(b, bval)

			// length is compared to cover the corner case where empty value is stored in the cache
			if !skipIntegrityCheck && (len(b) != n || hvalue != sha256.Sum256(b[:n])) {
				return n, ErrCorruptedData
			}

			return n, nil
		} else {
			if !errors.Is(err, cache.ErrKeyNotFound) {
				return 0, err
			}
		}
	}

	vLogID, offset := decodeOffset(off)

	if vLogID == 0 {
		// vLogID == 0 means value was not stored on any vlog i.e. empty value
		// integrity check may still be done in such case
		n = 0
	} else {
		vLog := s.fetchVLog(vLogID)
		defer s.releaseVLog(vLogID)

		n, err = vLog.ReadAt(b, offset)
		if err == multiapp.ErrAlreadyClosed || err == singleapp.ErrAlreadyClosed {
			return n, ErrAlreadyClosed

Review comment (performance):
just curious, when can an empty value be stored in the cache? Because the vlogcache is being set when appendData is called. Shouldn't we put a check there too to avoid this scenario?

Code after:
}

// ReadValue returns the actual associated value to a key at a specific transaction
// ErrExpiredEntry is be returned if the specified time has already elapsed
func (s *ImmuStore) ReadValue(entry *TxEntry) ([]byte, error) {
	if entry == nil || !entry.readonly {
		return nil, ErrIllegalArguments
	}

	if entry.md != nil && !entry.md.readonly {
		return nil, ErrIllegalArguments
	}

	if entry.md != nil && entry.md.ExpiredAt(time.Now()) {
		return nil, ErrExpiredEntry
	}

	if entry.vLen == 0 {
		// while not required, nil is returned instead of an empty slice

		// TODO: this step should be done after reading the value to ensure proper validations are made
		// But current changes in ExportTx with truncated transactions are not providing the value length
		// for truncated transactions, making it impossible to differentiate an empty value with a truncated one
		return nil, nil
	}

	b := make([]byte, entry.vLen)

	_, err := s.readValueAt(b, entry.vOff, entry.hVal, false)
	if err != nil {
		return nil, err
	}

	return b, nil
}

// readValueAt fills b with the value referenced by off
// expected value size and digest may be required for validations to pass
func (s *ImmuStore) readValueAt(b []byte, off int64, hvalue [sha256.Size]byte, skipIntegrityCheck bool) (n int, err error) {
	vLogID, offset := decodeOffset(off)

	if vLogID == 0 && len(b) > 0 {
		// it means value was not stored on any vlog i.e. a truncated transaction was replicated
		return 0, io.EOF
	}

	if vLogID > 0 {
		foundInTheCache := false

		if s.vLogCache != nil {
			val, err := s.vLogCache.Get(off)
Repository: codenotary/immudb
File: embedded/document/engine.go (Go)

Code before:
		}
		if err != nil {
			return nil, err
		}

		document := &structpb.Struct{Fields: map[string]*structpb.Value{}}
		for i := range colDescriptors {
			colName := colDescriptors[i].Column
			tv := row.ValuesByPosition[i]
			transformedTV := transformTypedValue(colName, tv)
			vtype, err := valueTypeToFieldValue(transformedTV)
			if err != nil {
				return nil, err
			}
			document.Fields[colName] = vtype
		}
		results = append(results, document)
	}

	return results, nil
}

func (d *Engine) getKey(ctx context.Context, tx *sql.SQLTx, collectionName string, documentID DocumentID) ([]byte, error) {
	table, err := tx.Catalog().GetTableByName(collectionName)
	if err != nil {
		return nil, err
	}

	var searchKey []byte

	valbuf := bytes.Buffer{}
	for _, col := range table.PrimaryIndex().Cols() {
		if col.Name() == defaultDocumentIDField {
			rval := sql.NewBlob(documentID[:])
			encVal, err := sql.EncodeRawValueAsKey(rval.RawValue(), col.Type(), col.MaxLen())
			if err != nil {
				return nil, err
			}
			_, err = valbuf.Write(encVal)
			if err != nil {
				return nil, err
			}
		}
	}
	pkEncVals := valbuf.Bytes()

	searchKey = sql.MapKey(
		d.GetPrefix(),
		sql.PIndexPrefix,
		sql.EncodeID(1),
		sql.EncodeID(table.ID()),

Review comment (suggestion):
it might be good having a *document.DocTx or similar struct, avoiding exposures of internal implementation

Code after:
		}
		if err != nil {
			return nil, err
		}

		document := &structpb.Struct{Fields: map[string]*structpb.Value{}}
		for i := range colDescriptors {
			colName := colDescriptors[i].Column
			switch colName {
			case defaultDocumentIDField, defaultDocumentBLOBField:
				tv := row.ValuesByPosition[i]
				transformedTV := transformTypedValue(colName, tv)
				vtype, err := valueTypeToFieldValue(transformedTV)
				if err != nil {
					return nil, err
				}
				document.Fields[colName] = vtype
			}
		}
		results = append(results, document)
	}

	return results, nil
}

func (d *Engine) getKey(ctx context.Context, tx *sql.SQLTx, collectionName string, documentID DocumentID) ([]byte, error) {
	table, err := tx.Catalog().GetTableByName(collectionName)
	if err != nil {
		return nil, err
	}

	var searchKey []byte

	valbuf := bytes.Buffer{}
	for _, col := range table.PrimaryIndex().Cols() {
		if col.Name() == defaultDocumentIDField {
			rval := sql.NewBlob(documentID[:])
			encVal, err := sql.EncodeRawValueAsKey(rval.RawValue(), col.Type(), col.MaxLen())
			if err != nil {
				return nil, err
			}
			_, err = valbuf.Write(encVal)
			if err != nil {
				return nil, err
			}
		}
	}
	pkEncVals := valbuf.Bytes()

	searchKey = sql.MapKey(
		d.GetPrefix(),