command.go

// Copyright 2013-2019 Aerospike, 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 aerospike

import (
	"bytes"
	"compress/zlib"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"time"

	. "github.com/aerospike/aerospike-client-go/logger"
	. "github.com/aerospike/aerospike-client-go/types"

	ParticleType "github.com/aerospike/aerospike-client-go/internal/particle_type"
	Buffer "github.com/aerospike/aerospike-client-go/utils/buffer"
)

const (
	// Flags commented out are not supported by cmd client.
	// Contains a read operation.
	_INFO1_READ int = (1 << 0)
	// Get all bins.
	_INFO1_GET_ALL int = (1 << 1)
	// Batch read or exists.
	_INFO1_BATCH int = (1 << 3)

	// Do not read the bins
	_INFO1_NOBINDATA int = (1 << 5)

	// Involve all replicas in read operation.
	_INFO1_CONSISTENCY_ALL = (1 << 6)

	// Tell server to compress its response.
	_INFO1_COMPRESS_RESPONSE = (1 << 7)

	// Create or update record
	_INFO2_WRITE int = (1 << 0)
	// Fling a record into the belly of Moloch.
	_INFO2_DELETE int = (1 << 1)
	// Update if expected generation == old.
	_INFO2_GENERATION int = (1 << 2)
	// Update if new generation >= old, good for restore.
	_INFO2_GENERATION_GT int = (1 << 3)
	// Transaction resulting in record deletion leaves tombstone (Enterprise only).
	_INFO2_DURABLE_DELETE int = (1 << 4)
	// Create only. Fail if record already exists.
	_INFO2_CREATE_ONLY int = (1 << 5)

	// Return a result for every operation.
	_INFO2_RESPOND_ALL_OPS int = (1 << 7)

	// This is the last of a multi-part message.
	_INFO3_LAST int = (1 << 0)
	// Commit to master only before declaring success.
	_INFO3_COMMIT_MASTER int = (1 << 1)
	// Update only. Merge bins.
	_INFO3_UPDATE_ONLY int = (1 << 3)

	// Create or completely replace record.
	_INFO3_CREATE_OR_REPLACE int = (1 << 4)
	// Completely replace existing record only.
	_INFO3_REPLACE_ONLY int = (1 << 5)
	// Linearize read when in CP mode.
	_INFO3_LINEARIZE_READ int = (1 << 6)

	_MSG_TOTAL_HEADER_SIZE     uint8 = 30
	_FIELD_HEADER_SIZE         uint8 = 5
	_OPERATION_HEADER_SIZE     uint8 = 8
	_MSG_REMAINING_HEADER_SIZE uint8 = 22
	_DIGEST_SIZE               uint8 = 20
	_COMPRESS_THRESHOLD        int   = 128
	_CL_MSG_VERSION            int64 = 2
	_AS_MSG_TYPE               int64 = 3
	_AS_MSG_TYPE_COMPRESSED    int64 = 4
)

// command interface describes all commands available
type command interface {
	getPolicy(ifc command) Policy

	writeBuffer(ifc command) error
	getNode(ifc command) (*Node, error)
	getConnection(policy Policy) (*Connection, error)
	putConnection(conn *Connection)
	parseResult(ifc command, conn *Connection) error
	parseRecordResults(ifc command, receiveSize int) (bool, error)

	execute(ifc command, isRead bool) error
	// Executes the command
	Execute() error
}

// Holds data buffer for the command
type baseCommand struct {
	node *Node
	conn *Connection

	// dataBufferCompress is not a second buffer; it is just a pointer to
	// the beginning of the dataBuffer.
	// To avoid allocating multiple buffers before compression, the dataBuffer
	// will be referencing to a padded buffer. After the command is written to
	// the buffer, this padding will be used to compress the command in-place,
	// and then the compressed proto header will be written.
	dataBufferCompress []byte
	dataBuffer         []byte
	dataOffset         int

	// oneShot determines if streaming commands like query, scan or queryAggregate
	// are not retried if they error out mid-parsing
	oneShot bool

	// will determine if the buffer will be compressed
	// before being sent to the server
	compressed bool
}

// Writes the command for write operations
func (cmd *baseCommand) setWrite(policy *WritePolicy, operation OperationType, key *Key, bins []*Bin, binMap BinMap) error {
	cmd.begin()
	fieldCount, err := cmd.estimateKeySize(key, policy.SendKey)
	if err != nil {
		return err
	}

	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}

	if binMap == nil {
		for i := range bins {
			if err := cmd.estimateOperationSizeForBin(bins[i]); err != nil {
				return err
			}
		}
	} else {
		for name, value := range binMap {
			if err := cmd.estimateOperationSizeForBinNameAndValue(name, value); err != nil {
				return err
			}
		}
	}

	if err := cmd.sizeBuffer(policy.compress()); err != nil {
		return err
	}

	if binMap == nil {
		cmd.writeHeaderWithPolicy(policy, 0, _INFO2_WRITE, fieldCount, len(bins))
	} else {
		cmd.writeHeaderWithPolicy(policy, 0, _INFO2_WRITE, fieldCount, len(binMap))
	}

	cmd.writeKey(key, policy.SendKey)

	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return err
		}
	}

	if binMap == nil {
		for i := range bins {
			if err := cmd.writeOperationForBin(bins[i], operation); err != nil {
				return err
			}
		}
	} else {
		for name, value := range binMap {
			if err := cmd.writeOperationForBinNameAndValue(name, value, operation); err != nil {
				return err
			}
		}
	}

	cmd.end()
	cmd.markCompressed(policy)

	return nil
}

// Writes the command for delete operations
func (cmd *baseCommand) setDelete(policy *WritePolicy, key *Key) error {
	cmd.begin()
	fieldCount, err := cmd.estimateKeySize(key, false)
	if err != nil {
		return err
	}

	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}

	if err := cmd.sizeBuffer(policy.compress()); err != nil {
		return err
	}
	cmd.writeHeaderWithPolicy(policy, 0, _INFO2_WRITE|_INFO2_DELETE, fieldCount, 0)
	cmd.writeKey(key, false)
	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return err
		}
	}
	cmd.end()
	cmd.markCompressed(policy)

	return nil

}

// Writes the command for touch operations
func (cmd *baseCommand) setTouch(policy *WritePolicy, key *Key) error {
	cmd.begin()
	fieldCount, err := cmd.estimateKeySize(key, policy.SendKey)
	if err != nil {
		return err
	}

	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}

	cmd.estimateOperationSize()
	if err := cmd.sizeBuffer(false); err != nil {
		return err
	}
	cmd.writeHeaderWithPolicy(policy, 0, _INFO2_WRITE, fieldCount, 1)
	cmd.writeKey(key, policy.SendKey)
	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return err
		}
	}
	cmd.writeOperationForOperationType(_TOUCH)
	cmd.end()
	return nil

}

// Writes the command for exist operations
func (cmd *baseCommand) setExists(policy *BasePolicy, key *Key) error {
	cmd.begin()
	fieldCount, err := cmd.estimateKeySize(key, false)
	if err != nil {
		return err
	}

	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}

	if err := cmd.sizeBuffer(false); err != nil {
		return err
	}
	cmd.writeHeader(policy, _INFO1_READ|_INFO1_NOBINDATA, 0, fieldCount, 0)
	cmd.writeKey(key, false)
	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return err
		}
	}
	cmd.end()
	return nil

}

// Writes the command for get operations (all bins)
func (cmd *baseCommand) setReadForKeyOnly(policy *BasePolicy, key *Key) error {
	cmd.begin()
	fieldCount, err := cmd.estimateKeySize(key, false)
	if err != nil {
		return err
	}
	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}
	if err := cmd.sizeBuffer(false); err != nil {
		return err
	}
	cmd.writeHeader(policy, _INFO1_READ|_INFO1_GET_ALL, 0, fieldCount, 0)
	cmd.writeKey(key, false)
	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return err
		}
	}
	cmd.end()
	return nil

}

// Writes the command for get operations (specified bins)
func (cmd *baseCommand) setRead(policy *BasePolicy, key *Key, binNames []string) (err error) {
	if len(binNames) > 0 {
		cmd.begin()
		fieldCount, err := cmd.estimateKeySize(key, false)
		if err != nil {
			return err
		}

		predSize := 0
		if len(policy.PredExp) > 0 {
			predSize = cmd.estimatePredExpSize(policy.PredExp)
			fieldCount++
		}

		for i := range binNames {
			cmd.estimateOperationSizeForBinName(binNames[i])
		}
		if err = cmd.sizeBuffer(false); err != nil {
			return nil
		}
		cmd.writeHeader(policy, _INFO1_READ, 0, fieldCount, len(binNames))
		cmd.writeKey(key, false)

		if len(policy.PredExp) > 0 {
			cmd.writePredExp(policy.PredExp, predSize)
		}

		for i := range binNames {
			cmd.writeOperationForBinName(binNames[i], _READ)
		}
		cmd.end()
	} else {
		err = cmd.setReadForKeyOnly(policy, key)
	}

	return err
}

// Writes the command for getting metadata operations
func (cmd *baseCommand) setReadHeader(policy *BasePolicy, key *Key) error {
	cmd.begin()
	fieldCount, err := cmd.estimateKeySize(key, false)
	if err != nil {
		return err
	}

	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}

	cmd.estimateOperationSizeForBinName("")
	if err := cmd.sizeBuffer(policy.compress()); err != nil {
		return err
	}

	cmd.writeHeader(policy, _INFO1_READ|_INFO1_NOBINDATA, 0, fieldCount, 1)

	cmd.writeKey(key, false)
	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return err
		}
	}
	cmd.writeOperationForBinName("", _READ)
	cmd.end()
	return nil

}

// Implements different command operations
func (cmd *baseCommand) setOperate(policy *WritePolicy, key *Key, operations []*Operation) (bool, error) {
	if len(operations) == 0 {
		return false, NewAerospikeError(PARAMETER_ERROR, "No operations were passed.")
	}

	cmd.begin()
	fieldCount := 0
	readAttr := 0
	writeAttr := 0
	hasWrite := false
	readBin := false
	readHeader := false
	RespondPerEachOp := policy.RespondPerEachOp

	for i := range operations {
		switch operations[i].opType {
		case _BIT_READ:
			fallthrough
		case _HLL_READ:
			fallthrough
		case _MAP_READ:
			// Map operations require RespondPerEachOp to be true.
			RespondPerEachOp = true
			// Fall through to read.
			fallthrough
		case _READ, _CDT_READ:
			if !operations[i].headerOnly {
				readAttr |= _INFO1_READ

				// Read all bins if no bin is specified.
				if operations[i].binName == "" {
					readAttr |= _INFO1_GET_ALL
				}
				readBin = true
			} else {
				readAttr |= _INFO1_READ
				readHeader = true
			}
		case _BIT_MODIFY:
			fallthrough
		case _HLL_MODIFY:
			fallthrough
		case _MAP_MODIFY:
			// Map operations require RespondPerEachOp to be true.
			RespondPerEachOp = true
			// Fall through to default.
			fallthrough
		default:
			writeAttr = _INFO2_WRITE
			hasWrite = true
		}
		cmd.estimateOperationSizeForOperation(operations[i])
	}

	ksz, err := cmd.estimateKeySize(key, policy.SendKey && hasWrite)
	if err != nil {
		return hasWrite, err
	}
	fieldCount += ksz

	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}

	if err := cmd.sizeBuffer(policy.compress()); err != nil {
		return hasWrite, err
	}

	if readHeader && !readBin {
		readAttr |= _INFO1_NOBINDATA
	}

	if RespondPerEachOp {
		writeAttr |= _INFO2_RESPOND_ALL_OPS
	}

	if writeAttr != 0 {
		cmd.writeHeaderWithPolicy(policy, readAttr, writeAttr, fieldCount, len(operations))
	} else {
		cmd.writeHeader(&policy.BasePolicy, readAttr, writeAttr, fieldCount, len(operations))
	}
	cmd.writeKey(key, policy.SendKey && hasWrite)

	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return hasWrite, err
		}
	}

	for _, operation := range operations {
		if err := cmd.writeOperationForOperation(operation); err != nil {
			return hasWrite, err
		}
	}

	cmd.end()
	cmd.markCompressed(policy)

	return hasWrite, nil
}

func (cmd *baseCommand) setUdf(policy *WritePolicy, key *Key, packageName string, functionName string, args *ValueArray) error {
	cmd.begin()
	fieldCount, err := cmd.estimateKeySize(key, policy.SendKey)
	if err != nil {
		return err
	}

	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}

	fc, err := cmd.estimateUdfSize(packageName, functionName, args)
	if err != nil {
		return err
	}
	fieldCount += fc

	if err := cmd.sizeBuffer(policy.compress()); err != nil {
		return err
	}

	cmd.writeHeaderWithPolicy(policy, 0, _INFO2_WRITE, fieldCount, 0)
	cmd.writeKey(key, policy.SendKey)
	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return err
		}
	}
	cmd.writeFieldString(packageName, UDF_PACKAGE_NAME)
	cmd.writeFieldString(functionName, UDF_FUNCTION)
	cmd.writeUdfArgs(args)
	cmd.end()
	cmd.markCompressed(policy)

	return nil
}

func (cmd *baseCommand) setBatchIndexReadCompat(policy *BatchPolicy, keys []*Key, batch *batchNode, binNames []string, readAttr int) error {
	offsets := batch.offsets
	max := len(batch.offsets)
	fieldCountRow := 1
	if policy.SendSetName {
		fieldCountRow = 2
	}

	binNameSize := 0
	operationCount := len(binNames)
	for _, binName := range binNames {
		binNameSize += len(binName) + int(_OPERATION_HEADER_SIZE)
	}

	// Estimate buffer size
	cmd.begin()
	fieldCount := 1
	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}

	cmd.dataOffset += int(_FIELD_HEADER_SIZE) + 5

	var prev *Key
	for i := 0; i < max; i++ {
		key := keys[offsets[i]]
		cmd.dataOffset += len(key.digest) + 4

		// Try reference equality in hope that namespace/set for all keys is set from fixed variables.
		if prev != nil && prev.namespace == key.namespace &&
			(!policy.SendSetName || prev.setName == key.setName) {
			// Can set repeat previous namespace/bin names to save space.
			cmd.dataOffset++
		} else {
			// Must write full header and namespace/set/bin names.
			cmd.dataOffset += len(key.namespace) + int(_FIELD_HEADER_SIZE) + 6

			if policy.SendSetName {
				cmd.dataOffset += len(key.setName) + int(_FIELD_HEADER_SIZE)
			}
			cmd.dataOffset += binNameSize
			prev = key
		}
	}

	if err := cmd.sizeBuffer(policy.compress()); err != nil {
		return err
	}

	if policy.ConsistencyLevel == CONSISTENCY_ALL {
		readAttr |= _INFO1_CONSISTENCY_ALL
	}

	if len(binNames) == 0 {
		readAttr |= _INFO1_GET_ALL
	}

	cmd.writeHeader(&policy.BasePolicy, readAttr|_INFO1_BATCH, 0, fieldCount, 0)

	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return err
		}
	}

	// Write real field size.
	fieldSizeOffset := cmd.dataOffset
	if policy.SendSetName {
		cmd.writeFieldHeader(0, BATCH_INDEX_WITH_SET)
	} else {
		cmd.writeFieldHeader(0, BATCH_INDEX)
	}

	cmd.WriteUint32(uint32(max))

	if policy.AllowInline {
		cmd.WriteByte(1)
	} else {
		cmd.WriteByte(0)
	}

	prev = nil
	for i := 0; i < max; i++ {
		index := offsets[i]
		cmd.WriteUint32(uint32(index))

		key := keys[index]
		cmd.Write(key.digest[:])
		// Try reference equality in hope that namespace/set for all keys is set from fixed variables.
		if prev != nil && prev.namespace == key.namespace &&
			(!policy.SendSetName || prev.setName == key.setName) {
			// Can set repeat previous namespace/bin names to save space.
			cmd.WriteByte(1) // repeat
		} else {
			// Write full header, namespace and bin names.
			cmd.WriteByte(0) // do not repeat
			cmd.WriteByte(byte(readAttr))
			cmd.WriteUint16(uint16(fieldCountRow))
			cmd.WriteUint16(uint16(operationCount))
			cmd.writeFieldString(key.namespace, NAMESPACE)

			if policy.SendSetName {
				cmd.writeFieldString(key.setName, TABLE)
			}

			for _, binName := range binNames {
				cmd.writeOperationForBinName(binName, _READ)
			}

			prev = key
		}
	}

	cmd.WriteUint32At(uint32(cmd.dataOffset)-uint32(_MSG_TOTAL_HEADER_SIZE)-4, fieldSizeOffset)
	cmd.end()
	cmd.markCompressed(policy)

	return nil
}

func (cmd *baseCommand) setBatchIndexRead(policy *BatchPolicy, records []*BatchRead, batch *batchNode) error {
	offsets := batch.offsets
	max := len(batch.offsets)
	fieldCountRow := 1
	if policy.SendSetName {
		fieldCountRow = 2
	}

	// Estimate buffer size
	cmd.begin()
	fieldCount := 1
	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}

	cmd.dataOffset += int(_FIELD_HEADER_SIZE) + 5

	var prev *BatchRead
	for i := 0; i < max; i++ {
		record := records[offsets[i]]
		key := record.Key
		binNames := record.BinNames

		cmd.dataOffset += len(key.digest) + 4

		// Try reference equality in hope that namespace/set for all keys is set from fixed variables.
		if prev != nil && prev.Key.namespace == key.namespace &&
			(!policy.SendSetName || prev.Key.setName == key.setName) &&
			&prev.BinNames == &binNames && prev.ReadAllBins == record.ReadAllBins {
			// Can set repeat previous namespace/bin names to save space.
			cmd.dataOffset++
		} else {
			// Must write full header and namespace/set/bin names.
			cmd.dataOffset += len(key.namespace) + int(_FIELD_HEADER_SIZE) + 6

			if policy.SendSetName {
				cmd.dataOffset += len(key.setName) + int(_FIELD_HEADER_SIZE)
			}

			if len(binNames) != 0 {
				for _, binName := range binNames {
					cmd.estimateOperationSizeForBinName(binName)
				}
			}

			prev = record
		}
	}

	if err := cmd.sizeBuffer(policy.compress()); err != nil {
		return err
	}

	readAttr := _INFO1_READ
	if policy.ConsistencyLevel == CONSISTENCY_ALL {
		readAttr |= _INFO1_CONSISTENCY_ALL
	}

	cmd.writeHeader(&policy.BasePolicy, readAttr|_INFO1_BATCH, 0, fieldCount, 0)
	// cmd.writeHeader(&policy.BasePolicy, _INFO1_READ|_INFO1_BATCH, 0, 1, 0)

	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return err
		}
	}

	// Write real field size.
	fieldSizeOffset := cmd.dataOffset
	if policy.SendSetName {
		cmd.writeFieldHeader(0, BATCH_INDEX_WITH_SET)
	} else {
		cmd.writeFieldHeader(0, BATCH_INDEX)
	}

	cmd.WriteUint32(uint32(max))

	if policy.AllowInline {
		cmd.WriteByte(1)
	} else {
		cmd.WriteByte(0)
	}

	prev = nil
	for i := 0; i < max; i++ {
		index := offsets[i]
		cmd.WriteUint32(uint32(index))

		record := records[index]
		key := record.Key
		binNames := record.BinNames
		cmd.Write(key.digest[:])

		// Try reference equality in hope that namespace/set for all keys is set from fixed variables.
		if prev != nil && prev.Key.namespace == key.namespace &&
			(!policy.SendSetName || prev.Key.setName == key.setName) &&
			&prev.BinNames == &binNames && prev.ReadAllBins == record.ReadAllBins {
			// Can set repeat previous namespace/bin names to save space.
			cmd.WriteByte(1) // repeat
		} else {
			// Write full header, namespace and bin names.
			cmd.WriteByte(0) // do not repeat
			if len(binNames) > 0 {
				cmd.WriteByte(byte(readAttr))
				cmd.WriteUint16(uint16(fieldCountRow))
				cmd.WriteUint16(uint16(len(binNames)))
				cmd.writeFieldString(key.namespace, NAMESPACE)

				if policy.SendSetName {
					cmd.writeFieldString(key.setName, TABLE)
				}

				for _, binName := range binNames {
					cmd.writeOperationForBinName(binName, _READ)
				}
			} else {
				attr := byte(readAttr)
				if record.ReadAllBins {
					attr |= byte(_INFO1_GET_ALL)
				} else {
					attr |= byte(_INFO1_NOBINDATA)
				}
				cmd.WriteByte(attr)

				cmd.WriteUint16(uint16(fieldCountRow))
				cmd.WriteUint16(0)
				cmd.writeFieldString(key.namespace, NAMESPACE)

				if policy.SendSetName {
					cmd.writeFieldString(key.setName, TABLE)
				}
			}

			prev = record
		}
	}

	cmd.WriteUint32At(uint32(cmd.dataOffset)-uint32(_MSG_TOTAL_HEADER_SIZE)-4, fieldSizeOffset)
	cmd.end()
	cmd.markCompressed(policy)

	return nil
}

func (cmd *baseCommand) setScan(policy *ScanPolicy, namespace *string, setName *string, binNames []string, taskID uint64) error {
	cmd.begin()
	fieldCount := 0

	predSize := 0
	if len(policy.PredExp) > 0 {
		predSize = cmd.estimatePredExpSize(policy.PredExp)
		fieldCount++
	}

	if namespace != nil {
		cmd.dataOffset += len(*namespace) + int(_FIELD_HEADER_SIZE)
		fieldCount++
	}

	if setName != nil {
		cmd.dataOffset += len(*setName) + int(_FIELD_HEADER_SIZE)
		fieldCount++
	}

	if policy.RecordsPerSecond > 0 {
		cmd.dataOffset += 4 + int(_FIELD_HEADER_SIZE)
		fieldCount++
	}

	// Estimate scan options size.
	cmd.dataOffset += 2 + int(_FIELD_HEADER_SIZE)
	fieldCount++

	// Estimate scan timeout size.
	cmd.dataOffset += 4 + int(_FIELD_HEADER_SIZE)
	fieldCount++

	// Allocate space for TaskId field.
	cmd.dataOffset += 8 + int(_FIELD_HEADER_SIZE)
	fieldCount++

	if binNames != nil {
		for i := range binNames {
			cmd.estimateOperationSizeForBinName(binNames[i])
		}
	}

	if err := cmd.sizeBuffer(false); err != nil {
		return err
	}
	readAttr := _INFO1_READ

	if !policy.IncludeBinData {
		readAttr |= _INFO1_NOBINDATA
	}

	operationCount := 0
	if binNames != nil {
		operationCount = len(binNames)
	}
	cmd.writeHeader(&policy.BasePolicy, readAttr, 0, fieldCount, operationCount)

	if namespace != nil {
		cmd.writeFieldString(*namespace, NAMESPACE)
	}

	if setName != nil {
		cmd.writeFieldString(*setName, TABLE)
	}

	if len(policy.PredExp) > 0 {
		if err := cmd.writePredExp(policy.PredExp, predSize); err != nil {
			return err
		}
	}

	if policy.RecordsPerSecond > 0 {
		cmd.writeFieldInt32(int32(policy.RecordsPerSecond), RECORDS_PER_SECOND)
	}

	cmd.writeFieldHeader(2, SCAN_OPTIONS)
	priority := byte(policy.Priority)
	priority <<= 4

	if policy.FailOnClusterChange {
		priority |= 0x08
	}

	cmd.WriteByte(priority)
	cmd.WriteByte(byte(policy.ScanPercent))

	// Write scan timeout
	cmd.writeFieldHeader(4, SCAN_TIMEOUT)
	cmd.WriteInt32(int32(policy.SocketTimeout / time.Millisecond)) // in milliseconds

	cmd.writeFieldHeader(8, TRAN_ID)
	cmd.WriteUint64(taskID)

	if binNames != nil {
		for i := range binNames {
			cmd.writeOperationForBinName(binNames[i], _READ)
		}
	}

	cmd.end()

	return nil
}

func (cmd *baseCommand) setQuery(policy *QueryPolicy, wpolicy *WritePolicy, statement *Statement, operations []*Operation, write bool) (err error) {
	fieldCount := 0
	filterSize := 0
	binNameSize := 0
	predSize := 0
	predExp := statement.predExps

	recordsPerSecond := 0
	if !write {
		recordsPerSecond = policy.RecordsPerSecond
	}

	cmd.begin()

	if statement.Namespace != "" {
		cmd.dataOffset += len(statement.Namespace) + int(_FIELD_HEADER_SIZE)
		fieldCount++
	}

	if statement.IndexName != "" {
		cmd.dataOffset += len(statement.IndexName) + int(_FIELD_HEADER_SIZE)
		fieldCount++
	}

	if statement.SetName != "" {
		cmd.dataOffset += len(statement.SetName) + int(_FIELD_HEADER_SIZE)
		fieldCount++
	}

	// Allocate space for TaskId field.
	cmd.dataOffset += 8 + int(_FIELD_HEADER_SIZE)
	fieldCount++

	if statement.Filter != nil {
		idxType := statement.Filter.IndexCollectionType()

		if idxType != ICT_DEFAULT {
			cmd.dataOffset += int(_FIELD_HEADER_SIZE) + 1
			fieldCount++
		}

		cmd.dataOffset += int(_FIELD_HEADER_SIZE)
		filterSize++ // num filters

		sz, err := statement.Filter.EstimateSize()
		if err != nil {
			return err
		}
		filterSize += sz

		cmd.dataOffset += filterSize
		fieldCount++

		// Query bin names are specified as a field (Scan bin names are specified later as operations)
		if len(statement.BinNames) > 0 {
			cmd.dataOffset += int(_FIELD_HEADER_SIZE)
			binNameSize++ // num bin names

			for _, binName := range statement.BinNames {
				binNameSize += len(binName) + 1
			}
			cmd.dataOffset += binNameSize
			fieldCount++
		}
	} else {
		// Calling query with no filters is more efficiently handled by a primary index scan.
		// Estimate scan options size.
		cmd.dataOffset += (2 + int(_FIELD_HEADER_SIZE))
		fieldCount++

		// Estimate scan timeout size.
		cmd.dataOffset += (4 + int(_FIELD_HEADER_SIZE))
		fieldCount++

		// Estimate records per second size.
		if recordsPerSecond > 0 {
			cmd.dataOffset += 4 + int(_FIELD_HEADER_SIZE)
			fieldCount++
		}
	}

	if len(policy.PredExp) > 0 && len(predExp) == 0 {
		predExp = policy.PredExp
	}

	if len(predExp) > 0 {
		predSize = cmd.estimatePredExpSize(predExp)
		fieldCount++
	}

	var functionArgs *ValueArray
	if statement.functionName != "" {
		cmd.dataOffset += int(_FIELD_HEADER_SIZE) + 1 // udf type
		cmd.dataOffset += len(statement.packageName) + int(_FIELD_HEADER_SIZE)
		cmd.dataOffset += len(statement.functionName) + int(_FIELD_HEADER_SIZE)

		fasz := 0
		if len(statement.functionArgs) > 0 {
			functionArgs = NewValueArray(statement.functionArgs)
			fasz, err = functionArgs.EstimateSize()
			if err != nil {
				return err
			}
		}

		cmd.dataOffset += int(_FIELD_HEADER_SIZE) + fasz
		fieldCount += 4
	}

	// Operations (used in query execute) and bin names (used in scan/query) are mutually exclusive.
	if len(operations) > 0 {
		for _, op := range operations {
			cmd.estimateOperationSizeForOperation(op)
		}
	} else if len(statement.BinNames) > 0 && statement.Filter == nil {
		for _, binName := range statement.BinNames {
			cmd.estimateOperationSizeForBinName(binName)
		}
	}

	if err := cmd.sizeBuffer(false); err != nil {
		return err
	}

	operationCount := 0
	if len(operations) > 0 {
		operationCount = len(operations)
	} else if statement.Filter == nil && len(statement.BinNames) > 0 {
		operationCount = len(statement.BinNames)
	}

	if write {
		cmd.writeHeaderWithPolicy(wpolicy, 0, _INFO2_WRITE, fieldCount, operationCount)
	} else {
		readAttr := _INFO1_READ | _INFO1_NOBINDATA
		if policy.IncludeBinData {
			readAttr = _INFO1_READ
		}
		cmd.writeHeader(&policy.BasePolicy, readAttr, 0, fieldCount, operationCount)
	}

	if statement.Namespace != "" {
		cmd.writeFieldString(statement.Namespace, NAMESPACE)
	}

	if statement.IndexName != "" {
		cmd.writeFieldString(statement.IndexName, INDEX_NAME)
	}

	if statement.SetName != "" {
		cmd.writeFieldString(statement.SetName, TABLE)
	}

	cmd.writeFieldHeader(8, TRAN_ID)
	cmd.WriteUint64(statement.TaskId)

	if statement.Filter != nil {
		idxType := statement.Filter.IndexCollectionType()

		if idxType != ICT_DEFAULT {
			cmd.writeFieldHeader(1, INDEX_TYPE)
			cmd.WriteByte(byte(idxType))
		}

		cmd.writeFieldHeader(filterSize, INDEX_RANGE)
		cmd.WriteByte(byte(1)) // number of filters

		_, err := statement.Filter.write(cmd)
		if err != nil {
			return err
		}

		if len(statement.BinNames) > 0 {
			cmd.writeFieldHeader(binNameSize, QUERY_BINLIST)
			cmd.WriteByte(byte(len(statement.BinNames)))

			for _, binName := range statement.BinNames {
				len := copy(cmd.dataBuffer[cmd.dataOffset+1:], binName)
				cmd.dataBuffer[cmd.dataOffset] = byte(len)
				cmd.dataOffset += len + 1
			}
		}
	} else {
		// Calling query with no filters is more efficiently handled by a primary index scan.
		cmd.writeFieldHeader(2, SCAN_OPTIONS)
		priority := byte(policy.Priority)
		priority <<= 4

		if !write && policy.FailOnClusterChange {
			priority |= 0x08
		}

		cmd.WriteByte(priority)
		cmd.WriteByte(byte(100))

		// Write scan timeout
		cmd.writeFieldHeader(4, SCAN_TIMEOUT)
		cmd.WriteInt32(int32(policy.SocketTimeout / time.Millisecond)) // in milliseconds

		// Write records per second.
		if recordsPerSecond > 0 {
			cmd.writeFieldInt32(int32(recordsPerSecond), RECORDS_PER_SECOND)
		}
	}

	if len(predExp) > 0 {
		if err := cmd.writePredExp(predExp, predSize); err != nil {
			return err
		}
	}

	if statement.functionName != "" {
		cmd.writeFieldHeader(1, UDF_OP)
		if statement.returnData {
			cmd.dataBuffer[cmd.dataOffset] = byte(1)
		} else {
			cmd.dataBuffer[cmd.dataOffset] = byte(2)
		}
		cmd.dataOffset++

		cmd.writeFieldString(statement.packageName, UDF_PACKAGE_NAME)
		cmd.writeFieldString(statement.functionName, UDF_FUNCTION)
		cmd.writeUdfArgs(functionArgs)
	}

	if len(operations) > 0 {
		for _, op := range operations {
			cmd.writeOperationForOperation(op)
		}
	} else if len(statement.BinNames) > 0 && statement.Filter == nil {
		// scan binNames come last
		for _, binName := range statement.BinNames {
			cmd.writeOperationForBinName(binName, _READ)
		}
	}

	cmd.end()

	return nil
}

func (cmd *baseCommand) estimateKeySize(key *Key, sendKey bool) (int, error) {
	fieldCount := 0

	if key.namespace != "" {
		cmd.dataOffset += len(key.namespace) + int(_FIELD_HEADER_SIZE)
		fieldCount++
	}

	if key.setName != "" {
		cmd.dataOffset += len(key.setName) + int(_FIELD_HEADER_SIZE)
		fieldCount++
	}

	cmd.dataOffset += int(_DIGEST_SIZE + _FIELD_HEADER_SIZE)
	fieldCount++

	if sendKey {
		// field header size + key size
		sz, err := key.userKey.EstimateSize()
		if err != nil {
			return sz, err
		}
		cmd.dataOffset += sz + int(_FIELD_HEADER_SIZE) + 1
		fieldCount++
	}

	return fieldCount, nil
}

func (cmd *baseCommand) estimateUdfSize(packageName string, functionName string, args *ValueArray) (int, error) {
	cmd.dataOffset += len(packageName) + int(_FIELD_HEADER_SIZE)
	cmd.dataOffset += len(functionName) + int(_FIELD_HEADER_SIZE)

	sz, err := args.EstimateSize()
	if err != nil {
		return 0, err
	}

	// fmt.Println(args, sz)

	cmd.dataOffset += sz + int(_FIELD_HEADER_SIZE)
	return 3, nil
}

func (cmd *baseCommand) estimateOperationSizeForBin(bin *Bin) error {
	cmd.dataOffset += len(bin.Name) + int(_OPERATION_HEADER_SIZE)
	sz, err := bin.Value.EstimateSize()
	if err != nil {
		return err
	}
	cmd.dataOffset += sz
	return nil
}

func (cmd *baseCommand) estimateOperationSizeForBinNameAndValue(name string, value interface{}) error {
	cmd.dataOffset += len(name) + int(_OPERATION_HEADER_SIZE)
	sz, err := NewValue(value).EstimateSize()
	if err != nil {
		return err
	}
	cmd.dataOffset += sz
	return nil
}

func (cmd *baseCommand) estimateOperationSizeForOperation(operation *Operation) error {
	binLen := len(operation.binName)
	cmd.dataOffset += binLen + int(_OPERATION_HEADER_SIZE)

	if operation.encoder == nil {
		if operation.binValue != nil {
			sz, err := operation.binValue.EstimateSize()
			if err != nil {
				return err
			}
			cmd.dataOffset += sz
		}
	} else {
		sz, err := operation.encoder(operation, nil)
		if err != nil {
			return err
		}
		cmd.dataOffset += sz
	}
	return nil
}

func (cmd *baseCommand) estimateOperationSizeForBinName(binName string) {
	cmd.dataOffset += len(binName) + int(_OPERATION_HEADER_SIZE)
}

func (cmd *baseCommand) estimateOperationSize() {
	cmd.dataOffset += int(_OPERATION_HEADER_SIZE)
}

func (cmd *baseCommand) estimatePredExpSize(predExp []PredExp) int {
	sz := 0
	for _, predexp := range predExp {
		sz += predexp.marshaledSize()
	}
	cmd.dataOffset += sz + int(_FIELD_HEADER_SIZE)
	return sz
}

// Generic header write.
func (cmd *baseCommand) writeHeader(policy *BasePolicy, readAttr int, writeAttr int, fieldCount int, operationCount int) {
	infoAttr := 0
	if policy.LinearizeRead {
		infoAttr |= _INFO3_LINEARIZE_READ
	}

	if policy.ConsistencyLevel == CONSISTENCY_ALL {
		readAttr |= _INFO1_CONSISTENCY_ALL
	}

	if policy.UseCompression {
		readAttr |= _INFO1_COMPRESS_RESPONSE
	}

	// Write all header data except total size which must be written last.
	cmd.dataBuffer[8] = _MSG_REMAINING_HEADER_SIZE // Message header length.
	cmd.dataBuffer[9] = byte(readAttr)
	cmd.dataBuffer[10] = byte(writeAttr)
	cmd.dataBuffer[11] = byte(infoAttr)

	for i := 12; i < 26; i++ {
		cmd.dataBuffer[i] = 0
	}
	cmd.dataOffset = 26
	cmd.WriteInt16(int16(fieldCount))
	cmd.WriteInt16(int16(operationCount))
	cmd.dataOffset = int(_MSG_TOTAL_HEADER_SIZE)
}

// Header write for write operations.
func (cmd *baseCommand) writeHeaderWithPolicy(policy *WritePolicy, readAttr int, writeAttr int, fieldCount int, operationCount int) {
	// Set flags.
	generation := uint32(0)
	infoAttr := 0

	switch policy.RecordExistsAction {
	case UPDATE:
	case UPDATE_ONLY:
		infoAttr |= _INFO3_UPDATE_ONLY
	case REPLACE:
		infoAttr |= _INFO3_CREATE_OR_REPLACE
	case REPLACE_ONLY:
		infoAttr |= _INFO3_REPLACE_ONLY
	case CREATE_ONLY:
		writeAttr |= _INFO2_CREATE_ONLY
	}

	switch policy.GenerationPolicy {
	case NONE:
	case EXPECT_GEN_EQUAL:
		generation = policy.Generation
		writeAttr |= _INFO2_GENERATION
	case EXPECT_GEN_GT:
		generation = policy.Generation
		writeAttr |= _INFO2_GENERATION_GT
	}

	if policy.CommitLevel == COMMIT_MASTER {
		infoAttr |= _INFO3_COMMIT_MASTER
	}

	if policy.LinearizeRead {
		infoAttr |= _INFO3_LINEARIZE_READ
	}

	if policy.ConsistencyLevel == CONSISTENCY_ALL {
		readAttr |= _INFO1_CONSISTENCY_ALL
	}

	if policy.DurableDelete {
		writeAttr |= _INFO2_DURABLE_DELETE
	}

	if policy.UseCompression {
		readAttr |= _INFO1_COMPRESS_RESPONSE
	}

	// Write all header data except total size which must be written last.
	cmd.dataBuffer[8] = _MSG_REMAINING_HEADER_SIZE // Message header length.
	cmd.dataBuffer[9] = byte(readAttr)
	cmd.dataBuffer[10] = byte(writeAttr)
	cmd.dataBuffer[11] = byte(infoAttr)
	cmd.dataBuffer[12] = 0 // unused
	cmd.dataBuffer[13] = 0 // clear the result code
	cmd.dataOffset = 14
	cmd.WriteUint32(generation)
	cmd.dataOffset = 18
	cmd.WriteUint32(policy.Expiration)

	// Initialize timeout. It will be written later.
	cmd.dataBuffer[22] = 0
	cmd.dataBuffer[23] = 0
	cmd.dataBuffer[24] = 0
	cmd.dataBuffer[25] = 0

	cmd.dataOffset = 26
	cmd.WriteInt16(int16(fieldCount))
	cmd.WriteInt16(int16(operationCount))
	cmd.dataOffset = int(_MSG_TOTAL_HEADER_SIZE)
}

func (cmd *baseCommand) writeKey(key *Key, sendKey bool) {
	// Write key into buffer.
	if key.namespace != "" {
		cmd.writeFieldString(key.namespace, NAMESPACE)
	}

	if key.setName != "" {
		cmd.writeFieldString(key.setName, TABLE)
	}

	cmd.writeFieldBytes(key.digest[:], DIGEST_RIPE)

	if sendKey {
		cmd.writeFieldValue(key.userKey, KEY)
	}
}

func (cmd *baseCommand) writeOperationForBin(bin *Bin, operation OperationType) error {
	nameLength := copy(cmd.dataBuffer[(cmd.dataOffset+int(_OPERATION_HEADER_SIZE)):], bin.Name)

	// check for float support
	cmd.checkServerCompatibility(bin.Value)

	valueLength, err := bin.Value.EstimateSize()
	if err != nil {
		return err
	}

	cmd.WriteInt32(int32(nameLength + valueLength + 4))
	cmd.WriteByte((operation.op))
	cmd.WriteByte((byte(bin.Value.GetType())))
	cmd.WriteByte((byte(0)))
	cmd.WriteByte((byte(nameLength)))
	cmd.dataOffset += nameLength
	_, err = bin.Value.write(cmd)
	return err
}

func (cmd *baseCommand) writeOperationForBinNameAndValue(name string, val interface{}, operation OperationType) error {
	nameLength := copy(cmd.dataBuffer[(cmd.dataOffset+int(_OPERATION_HEADER_SIZE)):], name)

	v := NewValue(val)

	// check for float support
	cmd.checkServerCompatibility(v)

	valueLength, err := v.EstimateSize()
	if err != nil {
		return err
	}

	cmd.WriteInt32(int32(nameLength + valueLength + 4))
	cmd.WriteByte((operation.op))
	cmd.WriteByte((byte(v.GetType())))
	cmd.WriteByte((byte(0)))
	cmd.WriteByte((byte(nameLength)))
	cmd.dataOffset += nameLength
	_, err = v.write(cmd)
	return err
}

func (cmd *baseCommand) writeOperationForOperation(operation *Operation) error {
	nameLength := copy(cmd.dataBuffer[(cmd.dataOffset+int(_OPERATION_HEADER_SIZE)):], operation.binName)

	// check for float support
	cmd.checkServerCompatibility(operation.binValue)

	if operation.used {
		// cahce will set the used flag to false again
		operation.cache()
	}

	if operation.encoder == nil {
		valueLength, err := operation.binValue.EstimateSize()
		if err != nil {
			return err
		}

		cmd.WriteInt32(int32(nameLength + valueLength + 4))
		cmd.WriteByte((operation.opType.op))
		cmd.WriteByte((byte(operation.binValue.GetType())))
		cmd.WriteByte((byte(0)))
		cmd.WriteByte((byte(nameLength)))
		cmd.dataOffset += nameLength
		_, err = operation.binValue.write(cmd)
		return err
	}

	valueLength, err := operation.encoder(operation, nil)
	if err != nil {
		return err
	}

	cmd.WriteInt32(int32(nameLength + valueLength + 4))
	cmd.WriteByte((operation.opType.op))
	cmd.WriteByte((byte(ParticleType.BLOB)))
	cmd.WriteByte((byte(0)))
	cmd.WriteByte((byte(nameLength)))
	cmd.dataOffset += nameLength
	_, err = operation.encoder(operation, cmd)
	//mark the operation as used, so that it will be cached the next time it is used
	operation.used = err == nil
	return err
}

func (cmd *baseCommand) writeOperationForBinName(name string, operation OperationType) {
	nameLength := copy(cmd.dataBuffer[(cmd.dataOffset+int(_OPERATION_HEADER_SIZE)):], name)
	cmd.WriteInt32(int32(nameLength + 4))
	cmd.WriteByte((operation.op))
	cmd.WriteByte(byte(0))
	cmd.WriteByte(byte(0))
	cmd.WriteByte(byte(nameLength))
	cmd.dataOffset += nameLength
}

func (cmd *baseCommand) writeOperationForOperationType(operation OperationType) {
	cmd.WriteInt32(int32(4))
	cmd.WriteByte(operation.op)
	cmd.WriteByte(0)
	cmd.WriteByte(0)
	cmd.WriteByte(0)
}

func (cmd *baseCommand) writePredExp(predExp []PredExp, predSize int) error {
	cmd.writeFieldHeader(predSize, PREDEXP)
	for i := range predExp {
		if err := predExp[i].marshal(cmd); err != nil {
			return err
		}
	}
	return nil
}

// TODO: Remove this method and move it to the appropriate VALUE method
func (cmd *baseCommand) checkServerCompatibility(val Value) {
	if val == nil {
		return
	}

	// check for float support
	switch val.GetType() {
	case ParticleType.FLOAT:
		if !cmd.node.supportsFloat.Get() {
			panic("This cluster node doesn't support double precision floating-point values.")
		}
	case ParticleType.GEOJSON:
		if !cmd.node.supportsGeo.Get() {
			panic("This cluster node doesn't support geo-spatial features.")
		}
	}
}

func (cmd *baseCommand) writeFieldValue(value Value, ftype FieldType) error {
	// check for float support
	cmd.checkServerCompatibility(value)

	vlen, err := value.EstimateSize()
	if err != nil {
		return err
	}
	cmd.writeFieldHeader(vlen+1, ftype)
	cmd.WriteByte(byte(value.GetType()))

	_, err = value.write(cmd)
	return err
}

func (cmd *baseCommand) writeUdfArgs(value *ValueArray) error {
	if value != nil {
		vlen, err := value.EstimateSize()
		if err != nil {
			return err
		}
		cmd.writeFieldHeader(vlen, UDF_ARGLIST)
		_, err = value.pack(cmd)
		return err
	}

	cmd.writeFieldHeader(0, UDF_ARGLIST)
	return nil

}

func (cmd *baseCommand) writeFieldInt32(val int32, ftype FieldType) {
	cmd.writeFieldHeader(4, ftype)
	cmd.WriteInt32(val)
}

func (cmd *baseCommand) writeFieldInt64(val int64, ftype FieldType) {
	cmd.writeFieldHeader(8, ftype)
	cmd.WriteInt64(val)
}

func (cmd *baseCommand) writeFieldString(str string, ftype FieldType) {
	len := copy(cmd.dataBuffer[(cmd.dataOffset+int(_FIELD_HEADER_SIZE)):], str)
	cmd.writeFieldHeader(len, ftype)
	cmd.dataOffset += len
}

func (cmd *baseCommand) writeFieldBytes(bytes []byte, ftype FieldType) {
	copy(cmd.dataBuffer[cmd.dataOffset+int(_FIELD_HEADER_SIZE):], bytes)

	cmd.writeFieldHeader(len(bytes), ftype)
	cmd.dataOffset += len(bytes)
}

func (cmd *baseCommand) writeFieldHeader(size int, ftype FieldType) {
	cmd.WriteInt32(int32(size + 1))
	cmd.WriteByte((byte(ftype)))
}

// Int64ToBytes converts an int64 into slice of Bytes.
func (cmd *baseCommand) WriteInt64(num int64) (int, error) {
	return cmd.WriteUint64(uint64(num))
}

// Uint64ToBytes converts an uint64 into slice of Bytes.
func (cmd *baseCommand) WriteUint64(num uint64) (int, error) {
	binary.BigEndian.PutUint64(cmd.dataBuffer[cmd.dataOffset:cmd.dataOffset+8], num)
	cmd.dataOffset += 8
	return 8, nil
}

// Int32ToBytes converts an int32 to a byte slice of size 4
func (cmd *baseCommand) WriteInt32(num int32) (int, error) {
	return cmd.WriteUint32(uint32(num))
}

// Uint32ToBytes converts an uint32 to a byte slice of size 4
func (cmd *baseCommand) WriteUint32(num uint32) (int, error) {
	binary.BigEndian.PutUint32(cmd.dataBuffer[cmd.dataOffset:cmd.dataOffset+4], num)
	cmd.dataOffset += 4
	return 4, nil
}

// Uint32ToBytes converts an uint32 to a byte slice of size 4
func (cmd *baseCommand) WriteUint32At(num uint32, index int) (int, error) {
	binary.BigEndian.PutUint32(cmd.dataBuffer[index:index+4], num)
	return 4, nil
}

// Int16ToBytes converts an int16 to slice of bytes
func (cmd *baseCommand) WriteInt16(num int16) (int, error) {
	return cmd.WriteUint16(uint16(num))
}

// Int16ToBytes converts an int16 to slice of bytes
func (cmd *baseCommand) WriteUint16(num uint16) (int, error) {
	binary.BigEndian.PutUint16(cmd.dataBuffer[cmd.dataOffset:cmd.dataOffset+2], num)
	cmd.dataOffset += 2
	return 2, nil
}

func (cmd *baseCommand) WriteFloat32(float float32) (int, error) {
	bits := math.Float32bits(float)
	binary.BigEndian.PutUint32(cmd.dataBuffer[cmd.dataOffset:cmd.dataOffset+4], bits)
	cmd.dataOffset += 4
	return 4, nil
}

func (cmd *baseCommand) WriteFloat64(float float64) (int, error) {
	bits := math.Float64bits(float)
	binary.BigEndian.PutUint64(cmd.dataBuffer[cmd.dataOffset:cmd.dataOffset+8], bits)
	cmd.dataOffset += 8
	return 8, nil
}

func (cmd *baseCommand) WriteByte(b byte) error {
	cmd.dataBuffer[cmd.dataOffset] = b
	cmd.dataOffset++
	return nil
}

func (cmd *baseCommand) WriteString(s string) (int, error) {
	copy(cmd.dataBuffer[cmd.dataOffset:cmd.dataOffset+len(s)], s)
	cmd.dataOffset += len(s)
	return len(s), nil
}

func (cmd *baseCommand) Write(b []byte) (int, error) {
	copy(cmd.dataBuffer[cmd.dataOffset:cmd.dataOffset+len(b)], b)
	cmd.dataOffset += len(b)
	return len(b), nil
}

func (cmd *baseCommand) begin() {
	cmd.dataOffset = int(_MSG_TOTAL_HEADER_SIZE)
}

func (cmd *baseCommand) sizeBuffer(compress bool) error {
	return cmd.sizeBufferSz(cmd.dataOffset, compress)
}

func (cmd *baseCommand) validateHeader(header int64) error {
	msgVersion := (uint64(header) & 0xFF00000000000000) >> 56
	if msgVersion != 2 {
		return NewAerospikeError(PARSE_ERROR, fmt.Sprintf("Invalid Message Header: Expected version to be 2, but got %v", msgVersion))
	}

	msgType := (uint64(header) & 0x00FF000000000000) >> 49
	if !(msgType == 1 || msgType == 3 || msgType == 4) {
		return NewAerospikeError(PARSE_ERROR, fmt.Sprintf("Invalid Message Header: Expected type to be 1, 3 or 4, but got %v", msgType))
	}

	msgSize := header & 0x0000FFFFFFFFFFFF
	if msgSize > int64(MaxBufferSize) {
		return NewAerospikeError(PARSE_ERROR, fmt.Sprintf("Invalid Message Header: Expected size to be under 10MiB, but got %v", msgSize))
	}

	return nil
}

var (
	// MaxBufferSize protects against allocating massive memory blocks
	// for buffers. Tweak this number if you are returning a lot of
	// LDT elements in your queries.
	MaxBufferSize = 1024 * 1024 * 120 // 120 MB
)

const (
	msgHeaderPad  = 16
	zlibHeaderPad = 2
)

func (cmd *baseCommand) sizeBufferSz(size int, willCompress bool) error {

	if willCompress {
		// adds zlib and proto pads to the size of the buffer
		size += msgHeaderPad + zlibHeaderPad
	}

	// Corrupted data streams can result in a huge length.
	// Do a sanity check here.
	if size > MaxBufferSize || size < 0 {
		return NewAerospikeError(PARSE_ERROR, fmt.Sprintf("Invalid size for buffer: %d", size))
	}

	if size <= len(cmd.dataBuffer) {
		// don't touch the buffer
	} else if size <= cap(cmd.dataBuffer) {
		cmd.dataBuffer = cmd.dataBuffer[:size]
	} else {
		// not enough space
		cmd.dataBuffer = make([]byte, size)
	}

	// The trick here to keep a ref to the buffer, and set the buffer itself
	// to a padded version of the original:
	// | Proto Header | Original Compressed Size | compressed message |
	// |    8 Bytes   |          8 Bytes         |                    |
	if willCompress {
		cmd.dataBufferCompress = cmd.dataBuffer
		cmd.dataBuffer = cmd.dataBufferCompress[msgHeaderPad+zlibHeaderPad:]
	}

	return nil
}

func (cmd *baseCommand) end() {
	var proto = int64(cmd.dataOffset-8) | (_CL_MSG_VERSION << 56) | (_AS_MSG_TYPE << 48)
	binary.BigEndian.PutUint64(cmd.dataBuffer[0:], uint64(proto))
}

func (cmd *baseCommand) markCompressed(policy Policy) {
	cmd.compressed = policy.compress()
}

func (cmd *baseCommand) compress() error {
	if cmd.compressed && cmd.dataOffset > _COMPRESS_THRESHOLD {
		b := bytes.NewBuffer(cmd.dataBufferCompress[msgHeaderPad:])
		b.Reset()
		w := zlib.NewWriter(b)

		// There seems to be a bug either in Go's zlib or in zlibc
		// which messes up a single write block of bigger than 64KB to
		// the deflater.
		// Things work in multiple writes of 64KB though, so this is
		// how we're going to do it.
		i := 0
		const step = 64 * 1024
		for i+step < cmd.dataOffset {
			n, err := w.Write(cmd.dataBuffer[i : i+step])
			i += n
			if err != nil {
				return err
			}
		}

		if i < cmd.dataOffset {
			_, err := w.Write(cmd.dataBuffer[i:cmd.dataOffset])
			if err != nil {
				return err
			}
		}

		// flush
		w.Close()

		compressedSz := b.Len()

		// Use compressed buffer if compression completed within original buffer size.
		var proto = int64(compressedSz+8) | (_CL_MSG_VERSION << 56) | (_AS_MSG_TYPE_COMPRESSED << 48)
		binary.BigEndian.PutUint64(cmd.dataBufferCompress[0:], uint64(proto))
		binary.BigEndian.PutUint64(cmd.dataBufferCompress[8:], uint64(cmd.dataOffset))

		cmd.dataBuffer = cmd.dataBufferCompress
		cmd.dataOffset = compressedSz + 16
		cmd.dataBufferCompress = nil
	}

	return nil
}

// isCompressed returns the length of the compressed buffer.
// If the buffer is not compressed, the result will be -1
func (cmd *baseCommand) compressedSize() int {
	proto := Buffer.BytesToInt64(cmd.dataBuffer, 0)
	size := proto & 0xFFFFFFFFFFFF

	msgType := (proto >> 48) & 0xff

	if msgType != _AS_MSG_TYPE_COMPRESSED {
		return -1
	}

	return int(size)
}

////////////////////////////////////

func setInDoubt(err error, isRead bool, commandSentCounter int) error {
	// set inDoubt flag
	if ae, ok := err.(AerospikeError); ok {
		ae.SetInDoubt(isRead, commandSentCounter)
		return ae
	}

	return err
}

func (cmd *baseCommand) execute(ifc command, isRead bool) error {
	policy := ifc.getPolicy(ifc).GetBasePolicy()
	iterations := -1
	commandSentCounter := 0

	// for exponential backoff
	interval := policy.SleepBetweenRetries

	// set timeout outside the loop
	deadline := policy.deadline()

	var err error

	shouldSleep := false

	// Execute command until successful, timed out or maximum iterations have been reached.
	for {
		// too many retries
		if iterations++; (policy.MaxRetries <= 0 && iterations > 0) || (policy.MaxRetries > 0 && iterations > policy.MaxRetries) {
			if ae, ok := err.(AerospikeError); ok {
				err = NewAerospikeError(ae.ResultCode(), fmt.Sprintf("command execution timed out on client: Exceeded number of retries. See `Policy.MaxRetries`. (last error: %s)", err.Error()))
			}

			return setInDoubt(err, isRead, commandSentCounter)
		}

		// Sleep before trying again, after the first iteration
		if policy.SleepBetweenRetries > 0 && shouldSleep {
			// Do not sleep if you know you'll wake up after the deadline
			if policy.TotalTimeout > 0 && time.Now().Add(interval).After(deadline) {
				break
			}

			time.Sleep(interval)
			if policy.SleepMultiplier > 1 {
				interval = time.Duration(float64(interval) * policy.SleepMultiplier)
			}
		}

		shouldSleep = true

		// check for command timeout
		if policy.TotalTimeout > 0 && time.Now().After(deadline) {
			break
		}

		// set command node, so when you return a record it has the node
		cmd.node, err = ifc.getNode(ifc)
		if cmd.node == nil || !cmd.node.IsActive() || err != nil {
			// Node is currently inactive. Retry.
			continue
		}

		cmd.conn, err = ifc.getConnection(policy)
		if err != nil {
			if err == ErrConnectionPoolEmpty {
				// if the connection pool is empty, we still haven't tried
				// the transaction to increase the iteration count.
				iterations--
			}
			Logger.Debug("Node " + cmd.node.String() + ": " + err.Error())
			continue
		}

		// Assign the connection buffer to the command buffer
		cmd.dataBuffer = cmd.conn.dataBuffer

		// Set command buffer.
		err = ifc.writeBuffer(ifc)
		if err != nil {
			// All runtime exceptions are considered fatal. Do not retry.
			// Close socket to flush out possible garbage. Do not put back in pool.
			cmd.conn.Close()
			cmd.conn = nil
			return err
		}

		// Reset timeout in send buffer (destined for server) and socket.
		binary.BigEndian.PutUint32(cmd.dataBuffer[22:], 0)
		if !deadline.IsZero() {
			serverTimeout := deadline.Sub(time.Now())
			if serverTimeout < time.Millisecond {
				serverTimeout = time.Millisecond
			}
			binary.BigEndian.PutUint32(cmd.dataBuffer[22:], uint32(serverTimeout/time.Millisecond))
		}

		// now that the deadline has been set in the buffer, compress the contents
		if err = cmd.compress(); err != nil {
			return NewAerospikeError(SERIALIZE_ERROR, err.Error())
		}

		// Send command.
		_, err = cmd.conn.Write(cmd.dataBuffer[:cmd.dataOffset])
		if err != nil {
			// IO errors are considered temporary anomalies. Retry.
			// Close socket to flush out possible garbage. Do not put back in pool.
			cmd.conn.Close()
			cmd.conn = nil

			Logger.Debug("Node " + cmd.node.String() + ": " + err.Error())
			continue
		}
		commandSentCounter++

		// Parse results.
		err = ifc.parseResult(ifc, cmd.conn)
		if err != nil {
			if _, ok := err.(net.Error); err == ErrTimeout || err == io.EOF || ok {
				// IO errors are considered temporary anomalies. Retry.
				// Close socket to flush out possible garbage. Do not put back in pool.
				cmd.conn.Close()

				Logger.Debug("Node " + cmd.node.String() + ": " + err.Error())

				// retry only for non-streaming commands
				if !cmd.oneShot {
					cmd.conn = nil
					continue
				}
			}

			// close the connection
			// cancelling/closing the batch/multi commands will return an error, which will
			// close the connection to throw away its data and signal the server about the
			// situation. We will not put back the connection in the buffer.
			if cmd.conn.IsConnected() && KeepConnection(err) {
				// Put connection back in pool.
				cmd.node.PutConnection(cmd.conn)
			} else {
				cmd.conn.Close()
				cmd.conn = nil
			}

			return setInDoubt(err, isRead, commandSentCounter)
		}

		// in case it has grown and re-allocated
		if len(cmd.dataBufferCompress) > len(cmd.dataBuffer) {
			cmd.conn.dataBuffer = cmd.dataBufferCompress
		} else {
			cmd.conn.dataBuffer = cmd.dataBuffer
		}

		// Put connection back in pool.
		// cmd.node.PutConnection(cmd.conn)
		ifc.putConnection(cmd.conn)

		// command has completed successfully.  Exit method.
		return nil

	}

	// execution timeout
	return ErrTimeout
}

func (cmd *baseCommand) parseRecordResults(ifc command, receiveSize int) (bool, error) {
	panic(errors.New("Abstract method. Should not end up here"))
}