Added support for on-demand tables in the cloud.

- Cloud only: Added support for creating on-demand tables
- On-Prem only: Added support for setting Durability in put/delete operations
- Added support for returning row modification time in get operations

- TableLimits now includes a CapacityMode field, to allow for specifying OnDemand. The default is Provisioned. This may affect existing code if the TableLimits struct was created without using named fields.
- Internal logic now detects differences in server protocol version, and decrements its internal serial version to match. This is to maintain compatibility when a new driver is used with an old server.

Author: connelly38

CHANGELOG.md

@@ -3,10 +3,17 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](http://keepachangelog.com/).
 
-## Unpublished
+## [Unpublished]
 
 ### Added
 - Added latest Oracle Cloud Infrastructure regions and region codes: LIN, MTZ, VCP, BRS, UKB, JNB, SIN, MRS, ARN, AUH, MCT, WGA.
+- Cloud only: Added support for creating on-demand tables
+- On-Prem only: Added support for setting Durability in put/delete operations
+- Added support for returning row modification time in get operations
+
+### Changed
+- TableLimits now includes a CapacityMode field, to allow for specifying OnDemand. The default is Provisioned. This may affect existing code if the TableLimits struct was created without using named fields.
+- Internal logic now detects differences in server protocol version, and decrements its internal serial version to match. This is to maintain compatibility when a new driver is used with an old server.
 
 ## 1.2.2 - 2021-06-08
 

internal/test/config.go

@@ -148,6 +148,10 @@ func createClient(cfg *Config) (*nosqldb.Client, error) {
 		return nil, err
 	}
 
+	// this will set the protocol serial version according to the connected server.
+	// ignore errors here, they may be expected.
+	client.VerifyConnection()
+
 	if interceptor != nil {
 		err = interceptor.OnSetupClient(client)
 		if err != nil {

nosqldb/bad_protocol_test.go

@@ -53,6 +53,9 @@ func (suite *BadProtocolTestSuite) SetupSuite() {
 	suite.bpTestClient, err = nosqldb.NewClient(suite.Client.Config)
 	suite.Require().NoErrorf(err, "failed to create a client, got error %v", err)
 
+	// this will set the serial protocol version. Ignore errors from it.
+	suite.bpTestClient.VerifyConnection()
+
 	// Disable retry handling.
 	suite.bpTestClient.RetryHandler = nil
 	suite.bpTestClient.AuthorizationProvider = suite.Client.AuthorizationProvider
@@ -580,6 +583,7 @@ func (suite *BadProtocolTestSuite) TestBadPutRequest() {
 		3,              // RequestTimeout: packed int
 		suite.tableLen, // TableName: String
 		1,              // ReturnRow: boolean
+		1,              // Durability: 1 byte (serialVersion > 2)
 		1,              // ExactMatch: boolean
 		1,              // IdentityCacheSize: packed int
 		suite.valueLen, // Record: MapValue
@@ -629,7 +633,10 @@ func (suite *BadProtocolTestSuite) TestBadPutRequest() {
 	}
 
 	// Invalid TTL value/unit.
-	off = seekPos(lengths, 9)
+	off = seekPos(lengths, 10)
+	if suite.bpTestClient.GetSerialVersion() < 3 {
+		off -= 1 // Durability
+	}
 	desc = "invalid TTL value"
 	copy(data, origData)
 	suite.wr.Reset()
@@ -718,6 +725,7 @@ func (suite *BadProtocolTestSuite) TestBadWriteMultipleRequest() {
 		3,              // RequestTimeout: packed int
 		suite.tableLen, // TableName: string
 		1,              // OperationNum: packed int
+		1,              // Durability: 1 byte (serialVersion > 2)
 		1,              // abortOnFail: boolean
 		0,              // Sub requests: the size does not matter for this test.
 	}
@@ -744,7 +752,10 @@ func (suite *BadProtocolTestSuite) TestBadWriteMultipleRequest() {
 	}
 
 	// Invalid opcode for sub requests.
-	off = seekPos(lengths, 6)
+	off = seekPos(lengths, 7)
+	if suite.bpTestClient.GetSerialVersion() < 3 {
+		off -= 1 // Durability
+	}
 	testOpCodes := []proto.OpCode{proto.OpCode(-1), proto.Get}
 	for _, v := range testOpCodes {
 		desc = fmt.Sprintf("invalid opcode %v", v)
@@ -772,6 +783,7 @@ func (suite *BadProtocolTestSuite) TestBadMultiDeleteRequest() {
 		1,              // OpCode: byte
 		3,              // RequestTimeout: packed int
 		suite.tableLen, // TableName: string
+		1,              // Durability: 1 byte (serialVersion > 2)
 		suite.keyLen,   // Key: MapValue
 		1,              // HasFieldRange: boolean
 		3,              // MaxWriteKB: packed int
@@ -788,7 +800,10 @@ func (suite *BadProtocolTestSuite) TestBadMultiDeleteRequest() {
 	suite.doBadProtoTest(req, data, desc, 0)
 
 	// Invalid MaxWriteKB.
-	off = seekPos(lengths, 6)
+	off = seekPos(lengths, 7)
+	if suite.bpTestClient.GetSerialVersion() < 3 {
+		off -= 1 // Durability
+	}
 	var tests []int
 	if test.IsOnPrem() {
 		// There is no limit on MaxWriteKB for the on-premise server.
@@ -807,7 +822,10 @@ func (suite *BadProtocolTestSuite) TestBadMultiDeleteRequest() {
 	}
 
 	// Invalid length of ContinuationKey.
-	off = seekPos(lengths, 7)
+	off = seekPos(lengths, 8)
+	if suite.bpTestClient.GetSerialVersion() < 3 {
+		off -= 1 // Durability
+	}
 	tests = []int{-2, 100}
 	for _, v := range tests {
 		desc = fmt.Sprintf("invalid length of ContinuationKey %v", v)
@@ -823,9 +841,14 @@ func (suite *BadProtocolTestSuite) TestBadTableRequest() {
 	newTable := "ABCD"
 	stmt := "create table if not exists " + newTable + " (id integer, primary key(id))"
 	stmtLen, _ := suite.wr.WriteString(&stmt)
+	limits := &nosqldb.TableLimits{
+		ReadUnits:  50,
+		WriteUnits: 50,
+		StorageGB:  2,
+	}
 	req := &nosqldb.TableRequest{
 		Statement:   stmt,
-		TableLimits: &nosqldb.TableLimits{50, 50, 5},
+		TableLimits: limits,
 	}
 
 	var desc string
@@ -839,6 +862,7 @@ func (suite *BadProtocolTestSuite) TestBadTableRequest() {
 		4,       // ReadKB: int
 		4,       // WriteKB: int
 		4,       // StorageGB: int
+		1,       // LimitMode: byte (serialVersion > 2)
 		1,       // HasTableName: boolean
 	}
 

nosqldb/client.go

@@ -31,6 +31,7 @@ import (
 	"github.com/oracle/nosql-go-sdk/nosqldb/jsonutil"
 	"github.com/oracle/nosql-go-sdk/nosqldb/logger"
 	"github.com/oracle/nosql-go-sdk/nosqldb/nosqlerr"
+	"github.com/oracle/nosql-go-sdk/nosqldb/types"
 )
 
 // Client represents an Oracle NoSQL database client used to access the Oracle
@@ -79,6 +80,12 @@ type Client struct {
 	// Keep an internal map of tablename to next limits update time
 	tableLimitUpdateMap map[string]int64
 	limitMux            sync.Mutex
+
+	// (possibly negotiated) version of the protocol in use
+	serialVersion int16
+
+	// for managing one-time messaging
+	oneTimeMessages map[string]struct{}
 }
 
 var (
@@ -112,14 +119,15 @@ func NewClient(cfg Config) (*Client, error) {
 	}
 
 	c := &Client{
-		Config:     cfg,
-		HTTPClient: cfg.httpClient,
-		requestURL: cfg.Endpoint + sdkutil.DataServiceURI,
-		requestID:  0,
-		serverHost: cfg.host,
-		executor:   cfg.httpClient,
-		logger:     cfg.Logger,
-		isCloud:    cfg.IsCloud() || cfg.IsCloudSim(),
+		Config:        cfg,
+		HTTPClient:    cfg.httpClient,
+		requestURL:    cfg.Endpoint + sdkutil.DataServiceURI,
+		requestID:     0,
+		serverHost:    cfg.host,
+		executor:      cfg.httpClient,
+		logger:        cfg.Logger,
+		isCloud:       cfg.IsCloud() || cfg.IsCloudSim(),
+		serialVersion: proto.DefaultSerialVersion,
 	}
 	c.handleResponse = c.processResponse
 	c.queryLogger, err = newQueryLogger()
@@ -132,6 +140,8 @@ func NewClient(cfg Config) (*Client, error) {
 		c.rateLimiterMap = make(map[string]common.RateLimiterPair)
 	}
 
+	c.oneTimeMessages = make(map[string]struct{})
+
 	return c, nil
 }
 
@@ -145,9 +155,8 @@ func (c *Client) Close() error {
 		c.queryLogger.Close()
 	}
 
-	if c.logger != nil {
-		c.logger.Close()
-	}
+	// do not close logger; it may have been passed to us and
+	// may still be in use by the application
 
 	return nil
 }
@@ -800,7 +809,7 @@ func (c *Client) processRequest(req Request) (data []byte, err error) {
 		return nil, err
 	}
 
-	data, err = serializeRequest(req)
+	data, err = c.serializeRequest(req)
 	if err != nil || !c.isCloud {
 		return
 	}
@@ -952,7 +961,16 @@ func (c *Client) doExecute(ctx context.Context, req Request, data []byte) (resul
 				}
 			}
 
-			if !c.handleError(err, req, numThrottleRetries) {
+			if nosqlerr.Is(err, nosqlerr.UnsupportedProtocol) {
+				if c.decrementSerialVersion() == false {
+					return nil, err
+				}
+				// if serial version mismatch, we must re-serialize the request
+				data, err = c.serializeRequest(req)
+				if err != nil {
+					return nil, err
+				}
+			} else if !c.handleError(err, req, numThrottleRetries) {
 				return nil, err
 			}
 
@@ -1035,6 +1053,35 @@ func (c *Client) doExecute(ctx context.Context, req Request, data []byte) (resul
 			return nil, err
 		}
 
+		// warn if using features not implemented at the connected server
+		// currently cloud does not support Durability
+		if c.serialVersion < 3 || c.isCloud {
+			needMsg := false
+			if pReq, ok := req.(*PutRequest); ok && pReq.Durability.IsSet() {
+				needMsg = true
+			} else if dReq, ok := req.(*DeleteRequest); ok && dReq.Durability.IsSet() {
+				needMsg = true
+			} else if mReq, ok := req.(*MultiDeleteRequest); ok && mReq.Durability.IsSet() {
+				needMsg = true
+			} else if wReq, ok := req.(*WriteMultipleRequest); ok && wReq.Durability.IsSet() {
+				needMsg = true
+			}
+			if needMsg {
+				c.oneTimeMessage("The requested feature is not supported " +
+					"by the connected server: Durability")
+			}
+		}
+
+		// OnDemand is not available in V2
+		if c.serialVersion < 3 {
+			if tReq, ok := req.(*TableRequest); ok && tReq.TableLimits != nil {
+				if tReq.TableLimits.CapacityMode == types.OnDemand {
+					c.oneTimeMessage("The requested feature is not supported " +
+						"by the connected server: on demand capacity table")
+				}
+			}
+		}
+
 		reqCtx, reqCancel := context.WithTimeout(ctx, reqTimeout)
 		httpReq = httpReq.WithContext(reqCtx)
 		httpResp, err = c.executor.Do(httpReq)
@@ -1287,13 +1334,13 @@ func (c *Client) signHTTPRequest(httpReq *http.Request) error {
 // serializeRequest serializes the specified request into a slice of bytes that
 // will be sent to the server. The serial version is always written followed by
 // the actual request payload.
-func serializeRequest(req Request) (data []byte, err error) {
+func (c *Client) serializeRequest(req Request) (data []byte, err error) {
 	wr := binary.NewWriter()
-	if _, err = wr.WriteSerialVersion(proto.SerialVersion); err != nil {
+	if _, err = wr.WriteSerialVersion(c.serialVersion); err != nil {
 		return
 	}
 
-	if err = req.serialize(wr); err != nil {
+	if err = req.serialize(wr, c.serialVersion); err != nil {
 		return
 	}
 
@@ -1329,7 +1376,7 @@ func (c *Client) processOKResponse(data []byte, req Request) (Result, error) {
 
 	// A zero byte represents the operation succeeded.
 	if code == 0 {
-		res, err := req.deserialize(rd)
+		res, err := req.deserialize(rd, c.serialVersion)
 		if err != nil {
 			return nil, err
 		}
@@ -1372,6 +1419,10 @@ func wrapResponseErrors(code int, msg string) error {
 		return nosqlerr.New(errCode, "unknown error: %s", msg)
 
 	case nosqlerr.BadProtocolMessage:
+		// V2 proxy will return this message if V3 is used in the driver
+		if strings.Contains(msg, "Invalid driver serial version") {
+			return nosqlerr.New(nosqlerr.UnsupportedProtocol, msg)
+		}
 		return nosqlerr.NewIllegalArgument("bad protocol message: %s", msg)
 
 	default:
@@ -1423,3 +1474,50 @@ func (c *Client) ResetRateLimiters(tableName string) {
 	rp.WriteLimiter.Reset()
 	rp.ReadLimiter.Reset()
 }
+
+// VerifyConnection attempts to verify that the connection is useable.
+// It may check auth credentials, and may negotiate the protocol level
+// to use with the server.
+// This is typically only used in tests.
+func (c *Client) VerifyConnection() error {
+
+	// issue a GetTable call for a (probably) nonexistent table.
+	// expect a TableNotFound error (or success in the unlikely event a
+	// table exists with this name). Any other errors will be returned here.
+	// Internally, this may result in the client negotiating a lower
+	// protocol version, if connected to an older server.
+	req := &GetTableRequest{
+		TableName: "noop",
+		Timeout:   20 * time.Second,
+	}
+
+	_, err := c.GetTable(req)
+	if err != nil && nosqlerr.IsTableNotFound(err) == false {
+		return err
+	}
+
+	return nil
+}
+
+// decrementSerialVersion attempts to reduce the serial version used for
+// communicating with the server. If the version is already at its lowest
+// value, it will not be decremented and false will be returned.
+func (c *Client) decrementSerialVersion() bool {
+	if c.serialVersion > 2 {
+		c.serialVersion--
+		return true
+	}
+	return false
+}
+
+// GetSerialVersion is used for tests.
+func (c *Client) GetSerialVersion() int16 {
+	return c.serialVersion
+}
+
+func (c *Client) oneTimeMessage(msg string) {
+	if _, ok := c.oneTimeMessages[msg]; ok == false {
+		c.oneTimeMessages[msg] = struct{}{}
+		c.logger.Warn(msg)
+	}
+}

nosqldb/data_ops_test.go

@@ -941,9 +941,9 @@ func (suite *DataOpsTestSuite) TestNonNumericDataTypes() {
 	// int/int64 values are valid for TIMESTAMP data type starting with NoSQL
 	// Server 20.2.14 on-premise release and Cloud Simulator 1.3.2 release.
 	//if suite.IsOnPrem() && suite.Version >= "20.2.14" || suite.IsCloudSim() && suite.Version >= "1.3.2" {
-		tsTest.validValues = append(tsTest.validValues, []types.FieldValue{intVal, int64Val}...)
+	tsTest.validValues = append(tsTest.validValues, []types.FieldValue{intVal, int64Val}...)
 	//} else {
-		//tsTest.invalidValues = append(tsTest.invalidValues, []types.FieldValue{intVal, int64Val}...)
+	//tsTest.invalidValues = append(tsTest.invalidValues, []types.FieldValue{intVal, int64Val}...)
 	//}
 
 	testCases = append(testCases, tsTest)