Merge pull request #433 from aerospike/stage

Go Client v7.3.0

.github/workflows/build.yml

@@ -1,26 +1,25 @@
 name: Aerospike Go Client Tests
-'on':
+"on":
   push:
   pull_request:
 
 env:
-  AEROSPIKE_HOSTS: '127.0.0.1:3000'
+  AEROSPIKE_HOSTS: "127.0.0.1:3000"
 jobs:
   build:
     runs-on: ubuntu-latest
     strategy:
       matrix:
         go-version:
-         # - '1.18'
-          # - '1.19'
-          # - '1.20'
-          - '1.21'
+          - "1.20"
+          - "1.21"
+          - "1.22"
     steps:
       - uses: actions/checkout@v3
-      - name: 'Setup Go ${{ matrix.go-version }}'
+      - name: "Setup Go ${{ matrix.go-version }}"
         uses: actions/setup-go@v3
         with:
-          go-version: '${{ matrix.go-version }}'
+          go-version: "${{ matrix.go-version }}"
           cache: true
       - name: Display Go version
         run: go version
@@ -40,9 +39,9 @@ jobs:
         run: cd tools/cli | go build -o cli .
       - name: Build example files
         run: find examples -name "*.go" -type f -print0 | xargs -0 -n1 go build
-#      - name: Build with Reflection code removed
-#        run: go run github.com/onsi/ginkgo/v2/ginkgo build -tags="as_performance" .
-#      - name: Build for Google App Engine (unsafe package removed)
-#        run: go run github.com/onsi/ginkgo/v2/ginkgo -tags="app_engine" .
+      #      - name: Build with Reflection code removed
+      #        run: go run github.com/onsi/ginkgo/v2/ginkgo build -tags="as_performance" .
+      #      - name: Build for Google App Engine (unsafe package removed)
+      #        run: go run github.com/onsi/ginkgo/v2/ginkgo -tags="app_engine" .
       - name: Run the tests
         run: go run github.com/onsi/ginkgo/v2/ginkgo -cover -race -keep-going -succinct -randomize-suites -skip="HyperLogLog"

.gitignore

@@ -24,4 +24,6 @@ Makefile
 testdata/
 Dockerfile*
 .dockerignore
-docker-compose.yml
+docker-compose.yml
+golangci.yml
+cover.out

CHANGELOG.md

@@ -1,5 +1,52 @@
 # Change History
 
+## May 3 2024: v7.3.0
+
+This is a major feature release of the Go client and touches some of the fundamental aspects of the inner workings of it.
+We suggest complete testing of your application before using it in production.
+
+- **New Features**
+  - [CLIENT-2238] Convert batch calls with just one key per node in sub-batches to Get requests.
+    If the number keys for a sub-batch to a node is equal or less then the value set in BatchPolicy.DirectGetThreshold, the client use direct get instead of batch commands to reduce the load on the server.
+
+  - [CLIENT-2274] Use constant sized connection buffers and resize the connection buffers over time.
+
+    The client would use a single buffer on the connection and would grow it
+    per demand in case it needed a bigger buffer, but would not shrink it.
+    This helped with avoiding using buffer pools and the associated
+    synchronization, but resulted in excessive memory use in case there were a
+    few large records in the results, even if they were infrequent.
+    This changeset does two things:
+            1. Will use a memory pool for large records only. Large records
+               are defined as records bigger than `aerospike.PoolCutOffBufferSize`.
+               This is a tiered pool with different buffer sizes. The pool
+               uses `sync.Pool` under the cover, releasing unused buffers back to the
+               runtime.
+            2. By using bigger `aerospike.DefaultBufferSize` values, the user can
+               imitate the old behavior, so no memory pool is used most of the time.
+            3. Setting `aerospike.MinBufferSize` will prevent the pool using buffer sizes too small,
+               having to grow them frequently.
+            4. Buffers are resized every 5 seconds to the median size of buffers used over the previous period,
+               within the above limits.
+
+    This change should result in much lower memory use by the client.
+
+  - [CLIENT-2702] Support Client Transaction Metrics. The native client can now track transaction latencies using histograms. Enable using the `Client.EnableMetrics` API.
+
+- **Improvements**
+  - [CLIENT-2862] Use default batch policies when the record level batch policy is nil.
+  - [CLIENT-2889] Increase grpc `MaxRecvMsgSize` to handle big records for the proxy client.
+  - [CLIENT-2891] Export various batch operation struct fields. Resolves #247.
+  - Remove dependency on `xrand` sub-package since the native API is fast enough.
+  - Linter Clean up.
+  - `WritePolicy.SendKey` documentation, thanks to [Rishabh Sairawat](https://github.com/rishabhsairawat)
+  - Replaced the deprecated `ioutil.ReadFile` with `os.ReadFile`. PR #430, thanks to [Swarit Pandey](https://github.com/swarit-pandey)
+
+- **Fixes**
+  - [CLIENT-2905] Fix inconsistency of handling in-doubt flag in errors.
+  - [CLIENT-2890] Support `[]MapPair` return in reflection.
+    This fix supports unmarshalling ordered maps into `map[K]V` and `[]MapPair` in the structs.
+
 ## April 10 2024: v7.2.1
 
 This release updates the dependencies to mitigate security issues.

README.md

@@ -13,17 +13,20 @@ You can refer to the test files for idiomatic use cases.
 
 Please refer to [`CHANGELOG.md`](CHANGELOG.md) for release notes, or if you encounter breaking changes.
 
-- [Usage](#usage)
-- [Prerequisites](#Prerequisites)
-- [Installation](#Installation)
-- [Tweaking Performance](#Performance)
-- [Benchmarks](#Benchmarks)
-- [API Documentation](#API-Documentation)
-- [Google App Engine](#App-Engine)
-- [Reflection](#Reflection)
-- [Tests](#Tests)
-- [Examples](#Examples)
-  - [Tools](#Tools)
+- [Aerospike Go Client v7](#aerospike-go-client-v7)
+  - [Usage](#usage)
+  - [Prerequisites](#prerequisites)
+  - [Installation](#installation)
+    - [Some Hints:](#some-hints)
+  - [Performance Tweaking](#performance-tweaking)
+  - [Tests](#tests)
+  - [Examples](#examples)
+    - [Tools](#tools)
+  - [Benchmarks](#benchmarks)
+  - [API Documentation](#api-documentation)
+  - [Google App Engine](#google-app-engine)
+  - [Reflection, and Object API](#reflection-and-object-api)
+  - [License](#license)
 
 ## Usage
 
@@ -155,7 +158,7 @@ See the [`tools/benchmark/README.md`](tools/benchmark/README.md) for details.
 <a name="API-Documentation"></a>
 ## API Documentation
 
-A simple API documentation is available in the [`docs`](docs/README.md) directory. The latest up-to-date docs can be found in [![Godoc](https://godoc.org/github.com/aerospike/aerospike-client-go?status.svg)](https://pkg.go.dev/github.com/aerospike/aerospike-client-go).
+A simple API documentation is available in the [`docs`](docs/README.md) directory. The latest up-to-date docs can be found in [![Godoc](https://godoc.org/github.com/aerospike/aerospike-client-go?status.svg)](https://pkg.go.dev/github.com/aerospike/aerospike-client-go/v7).
 
 <a name="App-Engine"></a>
 ## Google App Engine

aerospike_suite_test.go

@@ -21,7 +21,6 @@ import (
 	"encoding/pem"
 	"flag"
 	"fmt"
-	"io/ioutil"
 	"log"
 	"math/rand"
 	"os"
@@ -380,7 +379,7 @@ func initTLS() *tls.Config {
 
 // Read content from file
 func readFromFile(filePath string) ([]byte, error) {
-	dataBytes, err := ioutil.ReadFile(filePath)
+	dataBytes, err := os.ReadFile(filePath)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to read from file `%s`: `%v`", filePath, err)
 	}

batch_attr.go

@@ -166,17 +166,6 @@ func (ba *batchAttr) adjustReadForAllBins(readAllBins bool) {
 	}
 }
 
-func (ba *batchAttr) setWrite(wp *BasePolicy) {
-	ba.filterExp = nil
-	ba.readAttr = 0
-	ba.writeAttr = _INFO2_WRITE | _INFO2_RESPOND_ALL_OPS
-	ba.infoAttr = 0
-	ba.expiration = 0
-	ba.generation = 0
-	ba.hasWrite = true
-	ba.sendKey = wp.SendKey
-}
-
 func (ba *batchAttr) setBatchWrite(wp *BatchWritePolicy) {
 	ba.filterExp = wp.FilterExpression
 	ba.readAttr = 0
@@ -254,17 +243,6 @@ func (ba *batchAttr) adjustWrite(ops []*Operation) {
 	}
 }
 
-func (ba *batchAttr) setUDF(up *BasePolicy) {
-	ba.filterExp = nil
-	ba.readAttr = 0
-	ba.writeAttr = _INFO2_WRITE
-	ba.infoAttr = 0
-	ba.expiration = 0
-	ba.generation = 0
-	ba.hasWrite = true
-	ba.sendKey = up.SendKey
-}
-
 func (ba *batchAttr) setBatchUDF(up *BatchUDFPolicy) {
 	ba.filterExp = up.FilterExpression
 	ba.readAttr = 0
@@ -284,17 +262,6 @@ func (ba *batchAttr) setBatchUDF(up *BatchUDFPolicy) {
 	}
 }
 
-func (ba *batchAttr) setDelete(dp *BasePolicy) {
-	ba.filterExp = nil
-	ba.readAttr = 0
-	ba.writeAttr = _INFO2_WRITE | _INFO2_RESPOND_ALL_OPS | _INFO2_DELETE
-	ba.infoAttr = 0
-	ba.expiration = 0
-	ba.generation = 0
-	ba.hasWrite = true
-	ba.sendKey = dp.SendKey
-}
-
 func (ba *batchAttr) setBatchDelete(dp *BatchDeletePolicy) {
 	ba.filterExp = dp.FilterExpression
 	ba.readAttr = 0

batch_command.go

@@ -24,9 +24,11 @@ type batcher interface {
 	cloneBatchCommand(batch *batchNode) batcher
 	filteredOut() int
 
-	retryBatch(ifc batcher, cluster *Cluster, deadline time.Time, iteration int, commandWasSent bool) (bool, Error)
+	retryBatch(ifc batcher, cluster *Cluster, deadline time.Time, iteration int) (bool, Error)
 	generateBatchNodes(*Cluster) ([]*batchNode, Error)
 	setSequence(int, int)
+
+	executeSingle(*Client) Error
 }
 
 type batchCommand struct {
@@ -56,7 +58,7 @@ func (cmd *batchCommand) prepareRetry(ifc command, isTimeout bool) bool {
 	return false
 }
 
-func (cmd *batchCommand) retryBatch(ifc batcher, cluster *Cluster, deadline time.Time, iteration int, commandWasSent bool) (bool, Error) {
+func (cmd *batchCommand) retryBatch(ifc batcher, cluster *Cluster, deadline time.Time, iteration int) (bool, Error) {
 	// Retry requires keys for this node to be split among other nodes.
 	// This is both recursive and exponential.
 	batchNodes, err := ifc.generateBatchNodes(cluster)
@@ -74,7 +76,7 @@ func (cmd *batchCommand) retryBatch(ifc batcher, cluster *Cluster, deadline time
 	for _, batchNode := range batchNodes {
 		command := ifc.cloneBatchCommand(batchNode)
 		command.setSequence(cmd.sequenceAP, cmd.sequenceSC)
-		if err := command.executeAt(command, cmd.policy.GetBasePolicy(), deadline, iteration, commandWasSent); err != nil {
+		if err := command.executeAt(command, cmd.policy.GetBasePolicy(), deadline, iteration); err != nil {
 			ferr = chainErrors(err, ferr)
 			if !cmd.policy.AllowPartialResults {
 				return false, ferr
@@ -93,6 +95,10 @@ func (cmd *batchCommand) getPolicy(ifc command) Policy {
 	return cmd.policy
 }
 
+func (cmd *batchCommand) transactionType() transactionType {
+	return ttNone
+}
+
 func (cmd *batchCommand) Execute() Error {
 	return cmd.execute(cmd)
 }
@@ -102,13 +108,13 @@ func (cmd *batchCommand) filteredOut() int {
 }
 
 func (cmd *batchCommand) generateBatchNodes(cluster *Cluster) ([]*batchNode, Error) {
-	panic("Unreachable")
+	panic(unreachable)
 }
 
 func (cmd *batchCommand) cloneBatchCommand(batch *batchNode) batcher {
-	panic("Unreachable")
+	panic(unreachable)
 }
 
 func (cmd *batchCommand) writeBuffer(ifc command) Error {
-	panic("Unreachable")
+	panic(unreachable)
 }

batch_command_delete.go

@@ -22,15 +22,17 @@ import (
 type batchCommandDelete struct {
 	batchCommand
 
-	keys    []*Key
-	records []*BatchRecord
-	attr    *batchAttr
+	batchDeletePolicy *BatchDeletePolicy
+	keys              []*Key
+	records           []*BatchRecord
+	attr              *batchAttr
 }
 
 func newBatchCommandDelete(
 	node *Node,
 	batch *batchNode,
 	policy *BatchPolicy,
+	batchDeletePolicy *BatchDeletePolicy,
 	keys []*Key,
 	records []*BatchRecord,
 	attr *batchAttr,
@@ -41,9 +43,10 @@ func newBatchCommandDelete(
 			policy:           policy,
 			batch:            batch,
 		},
-		keys:    keys,
-		records: records,
-		attr:    attr,
+		batchDeletePolicy: batchDeletePolicy,
+		keys:              keys,
+		records:           records,
+		attr:              attr,
 	}
 	return res
 }
@@ -108,8 +111,8 @@ func (cmd *batchCommandDelete) parseRecordResults(ifc command, receiveSize int)
 				return false, err
 			}
 		} else {
-			cmd.records[batchIndex].setError(cmd.node, resultCode, cmd.batchInDoubt(cmd.attr.hasWrite, cmd.commandWasSent))
 			cmd.records[batchIndex].Err = chainErrors(newCustomNodeError(cmd.node, resultCode), cmd.records[batchIndex].Err)
+			cmd.records[batchIndex].setError(cmd.node, resultCode, cmd.batchInDoubt(cmd.attr.hasWrite, cmd.commandSentCounter))
 		}
 	}
 	return true, nil
@@ -161,7 +164,41 @@ func (cmd *batchCommandDelete) parseRecord(rec *BatchRecord, key *Key, opCount i
 	return nil
 }
 
+func (cmd *batchCommandDelete) transactionType() transactionType {
+	return ttBatchWrite
+}
+
+func (cmd *batchCommandDelete) executeSingle(client *Client) Error {
+	policy := cmd.batchDeletePolicy.toWritePolicy(cmd.policy)
+	for i, key := range cmd.keys {
+		res, err := client.Operate(policy, key, DeleteOp())
+		cmd.records[i].setRecord(res)
+		if err != nil {
+			cmd.records[i].setRawError(err)
+
+			// Key not found is NOT an error for batch requests
+			if err.resultCode() == types.KEY_NOT_FOUND_ERROR {
+				continue
+			}
+
+			if err.resultCode() == types.FILTERED_OUT {
+				cmd.filteredOutCnt++
+				continue
+			}
+
+			if cmd.policy.AllowPartialResults {
+				continue
+			}
+			return err
+		}
+	}
+	return nil
+}
+
 func (cmd *batchCommandDelete) Execute() Error {
+	if len(cmd.keys) == 1 {
+		return cmd.executeSingle(cmd.node.cluster.client)
+	}
 	return cmd.execute(cmd)
 }