Added per-request namespace capability (on-premises only). (#51)

Added per-request namespace capability (on-premises only).

Also modified internal timing to use nanoTime() instead of
currentTimeMillis(), which can go backwards in specific instances,
and made per-request-iteration timeout values more accurate.

Author: connelly38

CHANGELOG.md

@@ -2,6 +2,16 @@
 All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/).
 
+## [Unreleased]
+
+### Added
+- On-premises only: added support for setting namespace on a per-request basis
+
+### Fixed
+- Internal: changed to use nanoTime() instead of currentTimeMillis() to avoid
+ possible issue if system clock rolls backwards
+- Additional internal per-request-iteration timeout corrections
+
 ## [5.4.9] 2023-02-14
 
 ### Added

driver/src/main/java/oracle/nosql/driver/DefaultRetryHandler.java

@@ -98,9 +98,6 @@ public void delay(Request request,
      *         do not delay at all.
      */
     public static int computeBackoffDelay(Request request, int fixedDelayMs) {
-        int timeoutMs = request.getTimeoutInternal();
-        long startTimeMs = request.getStartTimeMs();
-
         int delayMs = fixedDelayMs;
         if (delayMs == 0) {
             /* add 200ms plus a small random amount */
@@ -114,10 +111,12 @@ public static int computeBackoffDelay(Request request, int fixedDelayMs) {
          * if the delay would put us over the timeout, reduce it to just before
          * the timeout would occur.
          */
-        long nowMs = System.currentTimeMillis();
-        long msLeft = (startTimeMs + (long)timeoutMs) - nowMs;
-        if ((int)msLeft < delayMs) {
-            delayMs = (int)msLeft;
+        long nanosUsed = System.nanoTime() - request.getStartNanos();
+        int msUsed = Math.toIntExact(nanosUsed / 1_000_000);
+        int timeoutMs = request.getTimeoutInternal();
+        int msLeft = (timeoutMs - msUsed) - 1;
+        if (msLeft < delayMs) {
+            delayMs = msLeft;
             if (delayMs < 1) {
                 return 0;
             }

driver/src/main/java/oracle/nosql/driver/NoSQLHandleConfig.java

@@ -205,7 +205,7 @@ public class NoSQLHandleConfig implements Cloneable {
     private String compartment;
 
     /**
-     * On-premise only.
+     * On-premises only.
      *
      * The default namespace to use for all requests. If this is null (the
      * default), no namespace is used unless specified in table names in
@@ -1087,9 +1087,9 @@ public String getDefaultCompartment() {
     }
 
     /**
-     * @hidden
+     * Set the default namespace to use for the handle.
      *
-     * On-premise only.
+     * On-premises only.
      *
      * Sets the default namespace to use for requests sent using the
      * handle. This is an optional convenience method to avoid having to
@@ -1098,23 +1098,30 @@ public String getDefaultCompartment() {
      * Any non-namespace qualified table name in requests and/or SQL
      * statements will be resolved/qualified to the specified namespace.
      *
+     * This value can be overridden on a per-request basis by calling
+     * setNamespace() on individual requests.
+     *
      * @param defaultNamespace the default namespace to use
      *
      * @return this
+     *
+     * @since 5.4.10
      */
     public NoSQLHandleConfig setDefaultNamespace(String defaultNamespace) {
         this.defaultNamespace = defaultNamespace;
         return this;
     }
 
     /**
-     * @hidden
+     * Get the default namespace.
      *
-     * On-premise only.
+     * On-premises only.
      *
      * Returns the default namespace to use for requests or null if not set.
      *
      * @return the default namespace
+     *
+     * @since 5.4.10
      */
     public String getDefaultNamespace() {
         return defaultNamespace;

driver/src/main/java/oracle/nosql/driver/http/Client.java

@@ -470,16 +470,16 @@ public Result execute(Request kvRequest) {
             }
         }
 
-        final long startTime = System.currentTimeMillis();
-        kvRequest.setStartTimeMs(startTime);
+        final long startNanos = System.nanoTime();
+        kvRequest.setStartNanos(startNanos);
         final String requestClass = kvRequest.getClass().getSimpleName();
 
         String requestId = "";
         int thisIterationTimeoutMs = 0;
 
         do {
-            long thisTime = System.currentTimeMillis();
-            thisIterationTimeoutMs = timeoutMs - (int)(thisTime - startTime);
+            thisIterationTimeoutMs =
+                getIterationTimeoutMs(timeoutMs, startNanos);
 
             /*
              * Check rate limiters before executing the request.
@@ -513,15 +513,15 @@ public Result execute(Request kvRequest) {
                 }
             }
 
+            /* update iteration timeout in case limiters slept for some time */
+            thisIterationTimeoutMs =
+                getIterationTimeoutMs(timeoutMs, startNanos);
+
             /* ensure limiting didn't throw us over the timeout */
-            if (timeoutRequest(startTime, timeoutMs, exception)) {
+            if (thisIterationTimeoutMs <= 0) {
                 break;
             }
 
-            /* update iteration timeout in case limiters slept for some time */
-            thisTime = System.currentTimeMillis();
-            thisIterationTimeoutMs = timeoutMs - (int)(thisTime - startTime);
-
             final String authString =
                 authProvider.getAuthorizationString(kvRequest);
             authProvider.validateAuthString(authString);
@@ -550,7 +550,6 @@ public Result execute(Request kvRequest) {
             ResponseHandler responseHandler = null;
             short serialVersionUsed = serialVersion;
             ByteBuf buffer = null;
-            long networkLatency;
             try {
                 /*
                  * NOTE: the ResponseHandler will release the Channel
@@ -559,6 +558,14 @@ public Result execute(Request kvRequest) {
                  * operations in the loop.
                  */
                 Channel channel = httpClient.getChannel(thisIterationTimeoutMs);
+                /* update iteration timeout in case channel took some time */
+                thisIterationTimeoutMs =
+                    getIterationTimeoutMs(timeoutMs, startNanos);
+                /* ensure limiting didn't throw us over the timeout */
+                if (thisIterationTimeoutMs <= 0) {
+                    break;
+                }
+
                 requestId = Long.toString(nextRequestId());
                 responseHandler =
                     new ResponseHandler(httpClient, logger, channel,
@@ -573,10 +580,16 @@ public Result execute(Request kvRequest) {
                  */
                 kvRequest.setCheckRequestSize(false);
 
-                /* update timeout in request to match this iteration timeout */
+                /*
+                 * Temporarily change the timeout in the request object so
+                 * the serialized timeout sent to the server is correct for
+                 * this iteration. After serializing the request, set the
+                 * timeout back to the overall request timeout so that other
+                 * processing (retry delays, etc) work correctly.
+                 */
                 kvRequest.setTimeoutInternal(thisIterationTimeoutMs);
-
                 serialVersionUsed = writeContent(buffer, kvRequest);
+                kvRequest.setTimeoutInternal(timeoutMs);
 
                 /*
                  * If on-premises the authProvider will always be a
@@ -627,17 +640,20 @@ public Result execute(Request kvRequest) {
                 }
                 authProvider.setRequiredHeaders(authString, kvRequest, headers);
 
-                if (config.getDefaultNamespace() != null) {
-                    headers.add(REQUEST_NAMESPACE_HEADER,
-                                config.getDefaultNamespace());
+                String namespace = kvRequest.getNamespace();
+                if (namespace == null) {
+                    namespace = config.getDefaultNamespace();
+                }
+                if (namespace != null) {
+                    headers.add(REQUEST_NAMESPACE_HEADER, namespace);
                 }
 
                 if (isLoggable(logger, Level.FINE) &&
                     !kvRequest.getIsRefresh()) {
                     logTrace(logger, "Request: " + requestClass +
                                      ", requestId=" + requestId);
                 }
-                networkLatency = System.currentTimeMillis();
+                long latencyNanos = System.nanoTime();
                 httpClient.runRequest(request, responseHandler, channel);
 
                 boolean isTimeout =
@@ -663,7 +679,9 @@ public Result execute(Request kvRequest) {
                 rateDelayedMs += getRateDelayedFromHeader(
                                        responseHandler.getHeaders());
                 int resSize = wireContent.readerIndex();
-                networkLatency = System.currentTimeMillis() - networkLatency;
+                long networkLatency =
+                    (System.nanoTime() - latencyNanos) / 1_000_000;
+
 
                 if (serialVersionUsed < 3) {
                     /* so we can emit a one-time message if the app */
@@ -813,7 +831,8 @@ public Result execute(Request kvRequest) {
             } catch (UnsupportedProtocolException upe) {
                 /* reduce protocol version and try again */
                 if (decrementSerialVersion(serialVersionUsed) == true) {
-                    exception = upe;
+                    /* Don't set this exception: it's misleading */
+                    /* exception = upe; */
                     logFine(logger, "Got unsupported protocol error " +
                             "from server: decrementing serial version to " +
                             serialVersion + " and trying again.");
@@ -900,7 +919,7 @@ public Result execute(Request kvRequest) {
                     responseHandler.close();
                 }
             }
-        } while (! timeoutRequest(startTime, timeoutMs, exception));
+        } while (! timeoutRequest(startNanos, timeoutMs, exception));
 
         kvRequest.setRateLimitDelayedMs(rateDelayedMs);
         statsControl.observeError(kvRequest);
@@ -912,6 +931,18 @@ public Result execute(Request kvRequest) {
                 kvRequest.getRetryStats() : ""), exception);
     }
 
+    /**
+     * Calculate the timeout for the next iteration.
+     * This is basically the given timeout minus the time
+     * elapsed since the start of the request processing.
+     * If this returns zero or negative, the request should be
+     * aborted with a timeout exception.
+     */
+    private int getIterationTimeoutMs(long timeoutMs, long startNanos) {
+        long diffNanos = System.nanoTime() - startNanos;
+        return ((int)timeoutMs - Math.toIntExact(diffNanos / 1_000_000));
+    }
+
     /**
      * Returns a rate limiter for a query operation, if the query op has
      * a prepared statement and a limiter exists in the rate limiter map
@@ -1046,18 +1077,18 @@ public boolean updateRateLimiters(String tableName, TableLimits limits) {
 
     /**
      * Determine if the request should be timed out.
-     * Check if the request exceed the timeout given.
+     * Check if the request exceeds the timeout given.
      *
-     * @param startTime when the request starts
-     * @param requestTimeout the default timeout of this request
+     * @param startNanos when the request starts
+     * @param requestTimeoutMs the timeout of this request in ms
      * @param exception the last exception
      *
-     * @return true the request need to be timed out.
+     * @return true if the request needs to be timed out.
      */
-    boolean timeoutRequest(long startTime,
-                           long requestTimeout,
+    boolean timeoutRequest(long startNanos,
+                           long requestTimeoutMs,
                            Throwable exception) {
-        return ((System.currentTimeMillis() - startTime) >= requestTimeout);
+        return (getIterationTimeoutMs(requestTimeoutMs, startNanos) <= 0);
     }
 
     /**

driver/src/main/java/oracle/nosql/driver/ops/DeleteRequest.java

@@ -211,8 +211,8 @@ public DeleteRequest setReturnRow(boolean value) {
 
     /**
      * Sets the optional request timeout value, in milliseconds. This overrides
-     * any default value set in {@link NoSQLHandleConfig}. The value must be
-     * positive.
+     * any default value set with {@link NoSQLHandleConfig#setRequestTimeout}.
+     * The value must be positive.
      *
      * @param timeoutMs the timeout value, in milliseconds
      *
@@ -226,6 +226,27 @@ public DeleteRequest setTimeout(int timeoutMs) {
         return this;
     }
 
+    /**
+     * Sets the optional namespace.
+     * On-premises only.
+     *
+     * This overrides any default value set with
+     * {@link NoSQLHandleConfig#setDefaultNamespace}.
+     * Note: if a namespace is specified in the table name for the request
+     * (using the namespace:tablename format), that value will override this
+     * setting.
+     *
+     * @param namespace the namespace to use for the operation
+     *
+     * @return this
+     *
+     * @since 5.4.10
+     */
+    public DeleteRequest setNamespace(String namespace) {
+        super.setNamespaceInternal(namespace);
+        return this;
+    }
+
     /**
      * @hidden
      */

driver/src/main/java/oracle/nosql/driver/ops/GetIndexesRequest.java

@@ -87,8 +87,8 @@ public String getIndexName() {
 
     /**
      * Sets the request timeout value, in milliseconds. This overrides any
-     * default value set in {@link NoSQLHandleConfig}. The value must be
-     * positive.
+     * default value set with {@link NoSQLHandleConfig#setRequestTimeout}.
+     * The value must be positive.
      *
      * @param timeoutMs the timeout value, in milliseconds
      *
@@ -102,6 +102,27 @@ public GetIndexesRequest setTimeout(int timeoutMs) {
         return this;
     }
 
+    /**
+     * Sets the optional namespace.
+     * On-premises only.
+     *
+     * This overrides any default value set with
+     * {@link NoSQLHandleConfig#setDefaultNamespace}.
+     * Note: if a namespace is specified in the table name for the request
+     * (using the namespace:tablename format), that value will override this
+     * setting.
+     *
+     * @param namespace the namespace to use for the operation
+     *
+     * @return this
+     *
+     * @since 5.4.10
+     */
+    public GetIndexesRequest setNamespace(String namespace) {
+        super.setNamespaceInternal(namespace);
+        return this;
+    }
+
     /**
      * @hidden
      */

driver/src/main/java/oracle/nosql/driver/ops/GetRequest.java

@@ -132,8 +132,8 @@ public GetRequest setConsistency(Consistency consistency) {
 
     /**
      * Sets the request timeout value, in milliseconds. This overrides any
-     * default value set in {@link NoSQLHandleConfig}. The value must be
-     * positive.
+     * default value set with {@link NoSQLHandleConfig#setRequestTimeout}.
+     * The value must be positive.
      *
      * @param timeoutMs the timeout value, in milliseconds
      *
@@ -147,6 +147,27 @@ public GetRequest setTimeout(int timeoutMs) {
         return this;
     }
 
+    /**
+     * Sets the optional namespace.
+     * On-premises only.
+     *
+     * This overrides any default value set with
+     * {@link NoSQLHandleConfig#setDefaultNamespace}.
+     * Note: if a namespace is specified in the table name for the request
+     * (using the namespace:tablename format), that value will override this
+     * setting.
+     *
+     * @param namespace the namespace to use for the operation
+     *
+     * @return this
+     *
+     * @since 5.4.10
+     */
+    public GetRequest setNamespace(String namespace) {
+        super.setNamespaceInternal(namespace);
+        return this;
+    }
+
     /**
      * @hidden
      * Internal use only