From 3b1e6bf8f0ecb3411e81577be74cde6add9cd6b4 Mon Sep 17 00:00:00 2001 From: Jin Zhao Date: Fri, 10 Jul 2026 14:16:55 -0400 Subject: [PATCH] Add per-store topology support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A proxy can serve more than one NoSQL store. Keeping one shared topology for the proxy can cause a query to use shard information from the wrong store. Track topology separately for each store and send the cached topology version for every known store with each request. Read topology updates and query branch store names from the proxy, then cache them by store. For advanced queries, select the needed store topology once when execution starts and use that same snapshot for all later query batches. Keep the existing topology behavior for older proxies that do not support per-store topology. Protocol changes: - Add per-store topology sequence numbers in the NSON request header: h.sts: [{ sid: , ts: }, ...] The SDK sends both the existing legacy topology sequence and h.sts, allowing proxies to use per-store topology when supported while preserving rolling compatibility. - Add per-store topology refreshes in responses: stp: [{ sid: , pn: , sa: [] }, ...] The SDK caches these immutable topology entries by store name. Existing legacy tp response handling remains unchanged. - Add "qbs" to prepare/unprepared-query response: qbs: [string,..] It is a store-name array aligned with prepared-query branches (qb). The SDK records this branch-to-store mapping in PreparedStatement. - For each prepared query batch, serialize the selected branch’s store name as "sid"", allowing the proxy to validate the execution branch’s store identity. - At advanced-query execution start, the SDK freezes the topology for each query branch from the per-store cache. After a prepare response refreshes the cache, it updates the original request so internal batch requests carry the same topology snapshot used by local scanners. - For legacy prepared plans without "qbs", compiled by old proxy, the SDK uses the legacy topology. If it is unavailable, the SDK requires reprepare rather than guessing from cached per-store topology. - Binary/V3 topology handling remains legacy-only and unchanged. --- .../main/java/oracle/nosql/driver/Nson.java | 5 +- .../java/oracle/nosql/driver/http/Client.java | 97 +++++++- .../nosql/driver/ops/PreparedStatement.java | 37 ++- .../oracle/nosql/driver/ops/QueryRequest.java | 25 ++- .../java/oracle/nosql/driver/ops/Request.java | 31 +++ .../java/oracle/nosql/driver/ops/Result.java | 24 ++ .../driver/ops/serde/BinaryProtocol.java | 2 +- .../ops/serde/PrepareRequestSerializer.java | 3 +- .../driver/ops/serde/nson/NsonProtocol.java | 8 + .../ops/serde/nson/NsonSerializerFactory.java | 212 +++++++++++++++++- .../nosql/driver/query/QueryDriver.java | 66 ++++++ .../driver/query/RuntimeControlBlock.java | 18 +- .../nosql/driver/query/TopologyInfo.java | 50 ++++- .../oracle/nosql/driver/query/UnionIter.java | 6 +- .../java/oracle/nosql/driver/RequestTest.java | 3 +- 15 files changed, 542 insertions(+), 45 deletions(-) diff --git a/driver/src/main/java/oracle/nosql/driver/Nson.java b/driver/src/main/java/oracle/nosql/driver/Nson.java index 8d55ece5..12bf0f9b 100644 --- a/driver/src/main/java/oracle/nosql/driver/Nson.java +++ b/driver/src/main/java/oracle/nosql/driver/Nson.java @@ -436,7 +436,10 @@ public static String toJsonString(byte[] nson, JsonOptions options) { if (nson == null) { return null; } - JsonSerializer jsonSerializer = new JsonSerializer(options); + JsonSerializer jsonSerializer = + (options != null && options.getPrettyPrint()) ? + new JsonPrettySerializer(options) : + new JsonSerializer(options); try (NettyByteInputStream bis = NettyByteInputStream.createFromBytes(nson)) { Nson.generateEventsFromNson(jsonSerializer, bis, false); diff --git a/driver/src/main/java/oracle/nosql/driver/http/Client.java b/driver/src/main/java/oracle/nosql/driver/http/Client.java index 711bd7d6..3dd9dc20 100644 --- a/driver/src/main/java/oracle/nosql/driver/http/Client.java +++ b/driver/src/main/java/oracle/nosql/driver/http/Client.java @@ -43,8 +43,11 @@ import java.io.FileOutputStream; import java.io.IOException; import java.net.URL; +import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.Properties; import java.util.concurrent.ConcurrentHashMap; @@ -229,6 +232,9 @@ public class Client { private volatile TopologyInfo topology; + /* Per store topology */ + private final Map storeTopologies; + /* for internal testing */ private final String prepareFilename; @@ -300,6 +306,7 @@ public Client(Logger logger, } oneTimeMessages = new HashSet(); + storeTopologies = new ConcurrentHashMap(); statsControl = new StatsControlImpl(config, logger, httpClient, httpConfig.getRateLimitingEnabled()); @@ -360,6 +367,8 @@ public void shutdown() { if (threadPool != null) { threadPool.shutdown(); } + topology = null; + storeTopologies.clear(); } public int getAcquiredChannelCount() { @@ -441,9 +450,11 @@ public Result execute(Request kvRequest) { QueryRequest qreq = (QueryRequest)kvRequest; - /* Set the topo seq num in the request, if it has not been set - * already */ - kvRequest.setTopoSeqNum(getTopoSeqNum()); + /* + * Stamp the query request with the cached legacy topology sequence + * and the per-store topology sequences. + */ + setRequestTopology(kvRequest); statsControl.observeQuery(qreq); @@ -655,11 +666,15 @@ public Result execute(Request kvRequest) { */ kvRequest.setCheckRequestSize(false); - /* Set the topo seq num in the request, if it has not been set - * already */ + /* + * Stamp non-query and user query requests with the legacy + * topology sequence and the per-store topology sequences + * before serialization. The request preserves any topology + * captured earlier for a query execution. + */ if (!(kvRequest instanceof QueryRequest) || kvRequest.isQueryRequest()) { - kvRequest.setTopoSeqNum(getTopoSeqNum()); + setRequestTopology(kvRequest); } /* @@ -776,7 +791,8 @@ public Result execute(Request kvRequest) { long networkLatency = (System.nanoTime() - latencyNanos) / 1_000_000; - setTopology(res.getTopology()); + /* Update cached legacy or per-store topology from the response. */ + updateCachedTopology(res); if (serialVersionUsed < 3) { /* so we can emit a one-time message if the app */ @@ -1189,8 +1205,8 @@ public boolean updateRateLimiters(String tableName, TableLimits limits) { int durationSeconds = Integer.getInteger("test.rldurationsecs", 30) .intValue(); - double RUs = (double)limits.getReadUnits(); - double WUs = (double)limits.getWriteUnits(); + double RUs = limits.getReadUnits(); + double WUs = limits.getWriteUnits(); /* if there's a specified rate limiter percentage, use that */ double rlPercent = config.getDefaultRateLimitingPercentage(); @@ -1412,7 +1428,6 @@ private void setEnabledFeatures(HttpHeaders headers) { if (space <= 0) { space = v.length(); } - long val = 0; try { this.features = Long.parseLong(v, eq+1, space, 16); } catch (Exception e) { @@ -2042,6 +2057,68 @@ public TopologyInfo getTopology() { return topology; } + /** + * @hidden + * Returns an immutable snapshot of the per-store topology cache. + */ + public Map getStoreTopoSnapshot() { + return Collections.unmodifiableMap(new HashMap<>(storeTopologies)); + } + + /* + * Sets both the legacy topology sequence and a snapshot of the per-store + * topology sequences on the request before serialization. Both are set + * because the client cannot determine whether the proxy supports per-store + * topology sequences. + */ + private void setRequestTopology(Request kvRequest) { + /* legacy topology sequence */ + kvRequest.setTopoSeqNum(getTopoSeqNum()); + + /* per store topology sequence */ + Map topos = getStoreTopoSnapshot(); + Map seqNums = new HashMap<>(); + for (Map.Entry e : topos.entrySet()) { + seqNums.put(e.getKey(), e.getValue().getSeqNum()); + } + kvRequest.setStoreTopoSeqNums(seqNums); + } + + /* + * Updates the cached topology information: + * - Updates the per-store cache from a response that carries per-store + * topology information, retaining a newer cached sequence for each + * store. + * - Responses without per-store information update the legacy topology. + */ + private void updateCachedTopology(Result ret) { + if (ret.getStoreTopologies() != null) { + setPerStoreTopology(ret.getStoreTopologies()); + return; + } + + setTopology(ret.getTopology()); + } + + private void setPerStoreTopology(List storeTopos) { + + if (storeTopos == null) { + return; + } + + for (TopologyInfo topo : storeTopos) { + storeTopologies.compute( + topo.getStoreName(), (storeName, currentTopo) -> { + if (currentTopo == null || + currentTopo.getSeqNum() < topo.getSeqNum()) { + trace("New store topology: " + topo, 1); + return topo; + } + return currentTopo; + }); + } + } + private synchronized int getTopoSeqNum() { return (topology == null ? -1 : topology.getSeqNum()); } diff --git a/driver/src/main/java/oracle/nosql/driver/ops/PreparedStatement.java b/driver/src/main/java/oracle/nosql/driver/ops/PreparedStatement.java index fa6ced67..a27c35ff 100644 --- a/driver/src/main/java/oracle/nosql/driver/ops/PreparedStatement.java +++ b/driver/src/main/java/oracle/nosql/driver/ops/PreparedStatement.java @@ -95,6 +95,13 @@ public class PreparedStatement { */ private final ArrayList topTableNames; + /* + * The store names returned from a prepared query result. When present, + * this is branch-aligned with proxyStatements. Null means legacy plan + * metadata without per-store branch identity. + */ + private final ArrayList branchStoreNames; + /* * the operation code for the query. */ @@ -136,6 +143,8 @@ public class PreparedStatement { * @param operation operation code for the query * @param maxParallelism the maximum degree of parallelism possible for the * query + * @param branchStoreNames optional store names aligned with each proxy + * statement. Null means legacy plan metadata. * @hidden */ public PreparedStatement( @@ -150,7 +159,8 @@ public PreparedStatement( ArrayList namespaces, ArrayList tableNames, byte operation, - int maxParallelism) { + int maxParallelism, + ArrayList branchStoreNames) { if (proxyStatements == null || proxyStatements.isEmpty()) { throw new IllegalArgumentException( @@ -169,6 +179,7 @@ public PreparedStatement( this.topTableNames = tableNames; this.operation = operation; this.maxParallelism = maxParallelism; + this.branchStoreNames = branchStoreNames; } /** @@ -191,7 +202,8 @@ public PreparedStatement copyStatement() { namespaces, topTableNames, operation, - maxParallelism); + maxParallelism, + branchStoreNames); } /** @@ -330,6 +342,15 @@ public final byte[] getProxyStatement(int branch) { return proxyStatements.get(branch).clone(); } + /** + * Internal use only + * @return the number of proxy-side query branches in this statement + * @hidden + */ + public int getNumBranches() { + return proxyStatements.size(); + } + private static ArrayList copyProxyStatements( ArrayList source) { @@ -434,6 +455,18 @@ public String getTopTableName(int branch) { return topTableNames.get(branch); } + /** + * Internal use only + * @return store name from prepared statement if any + * @hidden + */ + public String getBranchStoreName(int branch) { + if (branchStoreNames == null) { + return null; + } + return branchStoreNames.get(branch); + } + /** * Internal use only * @return true if the query does writes diff --git a/driver/src/main/java/oracle/nosql/driver/ops/QueryRequest.java b/driver/src/main/java/oracle/nosql/driver/ops/QueryRequest.java index 37f59b11..397cdc85 100644 --- a/driver/src/main/java/oracle/nosql/driver/ops/QueryRequest.java +++ b/driver/src/main/java/oracle/nosql/driver/ops/QueryRequest.java @@ -207,6 +207,23 @@ public class QueryRequest extends DurableRequest implements AutoCloseable { public QueryRequest() { } + /** + * Updates this request's topology with the snapshot selected for an + * advanced query execution. This is called only after the + * {@link QueryDriver} has selected the base topologies. + * + * @param topoSeqNum the legacy topology sequence number, or {@code -1} + * if no legacy topology is available + * @param storeTopoSeqNums the per-store topology sequences + * @hidden + */ + public void setExecutionTopology(int topoSeqNum, + Map storeTopoSeqNums){ + + this.topoSeqNum = topoSeqNum; + this.storeTopoSeqNums = storeTopoSeqNums; + } + /** * Creates an internal QueryRequest out of the application-provided request. * @return a copy of the instance in a new object @@ -233,6 +250,7 @@ public QueryRequest copyInternal() { internalReq.isInternal = true; internalReq.driver = driver; internalReq.topoSeqNum = topoSeqNum; + internalReq.storeTopoSeqNums = storeTopoSeqNums; internalReq.inTestMode = inTestMode; internalReq.operationNumber = operationNumber; internalReq.numberOfOperations = numberOfOperations; @@ -1147,7 +1165,7 @@ public void validate() { } /* * Parallel queries have multiple requirements: - * o only for prepared queries + * o only for simple prepared queries * o if set, both number of operations and op number need to be set * o operation number must be <= number of operations * o number of operations must be <= the max @@ -1159,6 +1177,11 @@ public void validate() { throw new IllegalArgumentException( "Parallel queries are only allowed on prepared queries"); } + if (!isSimpleQuery()) { + throw new IllegalArgumentException( + "Parallel queries are only allowed on simple " + + "prepared queries"); + } /* check both non-zero and value of operation number */ if (getOperationNumber() > getNumberOfOperations() || getOperationNumber() <= 0) { diff --git a/driver/src/main/java/oracle/nosql/driver/ops/Request.java b/driver/src/main/java/oracle/nosql/driver/ops/Request.java index 4f1c2951..3acca4cd 100644 --- a/driver/src/main/java/oracle/nosql/driver/ops/Request.java +++ b/driver/src/main/java/oracle/nosql/driver/ops/Request.java @@ -9,6 +9,8 @@ import static oracle.nosql.driver.util.HttpHeaderValidation.validateHttpHeaderValue; +import java.util.Map; + import oracle.nosql.driver.NoSQLHandleConfig; import oracle.nosql.driver.RateLimiter; import oracle.nosql.driver.ops.serde.Serializer; @@ -85,6 +87,14 @@ public abstract class Request { protected int topoSeqNum = -1; + /** + * @hidden + * Per-store topology sequences serialized as request header h.sts. The + * client initializes this map before sending a request. An empty map + * advertises per-store topology support with no cached store topology. + */ + protected Map storeTopoSeqNums; + /** * @hidden * This is only required by Java SDK for internal cross-region request, not @@ -535,6 +545,27 @@ public void setTopoSeqNum(int n) { } } + /** + * internal use only + * @return the per-store topology sequences for request + * @hidden + */ + public Map getStoreTopoSeqNums() { + return storeTopoSeqNums; + } + + /** + * internal use only + * @param topoSeqNums the per-store topology sequences + * @hidden + */ + public void setStoreTopoSeqNums(Map topoSeqNums) { + if (storeTopoSeqNums != null) { + return; + } + storeTopoSeqNums = topoSeqNums; + } + /** * Returns the type name of the request. This is used for stats. * diff --git a/driver/src/main/java/oracle/nosql/driver/ops/Result.java b/driver/src/main/java/oracle/nosql/driver/ops/Result.java index ba7527b1..ee51ca6d 100644 --- a/driver/src/main/java/oracle/nosql/driver/ops/Result.java +++ b/driver/src/main/java/oracle/nosql/driver/ops/Result.java @@ -7,6 +7,8 @@ package oracle.nosql.driver.ops; +import java.util.List; + import oracle.nosql.driver.query.TopologyInfo; /** @@ -45,8 +47,12 @@ public class Result { */ private RetryStats retryStats; + /* Topology returned by the legacy single-topology protocol. */ private TopologyInfo topology; + /* Per-store topology entries returned by the per-store protocol. */ + private List storeTopologies; + protected Result() {} /** @@ -181,6 +187,24 @@ public void setTopology(TopologyInfo ti) { topology = ti; } + /** + * internal use only + * @return per-store topology information + * @hidden + */ + public List getStoreTopologies() { + return storeTopologies; + } + + /** + * internal use only + * @param topologies per-store topology information + * @hidden + */ + public void setStoreTopologies(List topologies) { + storeTopologies = topologies; + } + /** * Returns the server protocol serial version or 0 if not available. * This is a new feature not supported in older servers. diff --git a/driver/src/main/java/oracle/nosql/driver/ops/serde/BinaryProtocol.java b/driver/src/main/java/oracle/nosql/driver/ops/serde/BinaryProtocol.java index 0ee8627a..3f2a0ed2 100644 --- a/driver/src/main/java/oracle/nosql/driver/ops/serde/BinaryProtocol.java +++ b/driver/src/main/java/oracle/nosql/driver/ops/serde/BinaryProtocol.java @@ -349,7 +349,7 @@ public static TopologyInfo readTopologyInfo(ByteInputStream in) int[] shardIds = SerializationUtil.readPackedIntArray(in); - return new TopologyInfo(seqNum, shardIds); + return new TopologyInfo(null /* storeName */, seqNum, shardIds); } /* diff --git a/driver/src/main/java/oracle/nosql/driver/ops/serde/PrepareRequestSerializer.java b/driver/src/main/java/oracle/nosql/driver/ops/serde/PrepareRequestSerializer.java index 27b533a0..f90f086a 100644 --- a/driver/src/main/java/oracle/nosql/driver/ops/serde/PrepareRequestSerializer.java +++ b/driver/src/main/java/oracle/nosql/driver/ops/serde/PrepareRequestSerializer.java @@ -172,7 +172,8 @@ static PreparedStatement deserializeInternal( namespaces, tableNames, operation, - 0); /* no parallelism available */ + 0, /* no parallelism available */ + null); /* branchStoreNames */ result.setPreparedStatement(prep); result.setTopology(ti); diff --git a/driver/src/main/java/oracle/nosql/driver/ops/serde/nson/NsonProtocol.java b/driver/src/main/java/oracle/nosql/driver/ops/serde/nson/NsonProtocol.java index 35d0f35d..91431570 100644 --- a/driver/src/main/java/oracle/nosql/driver/ops/serde/nson/NsonProtocol.java +++ b/driver/src/main/java/oracle/nosql/driver/ops/serde/nson/NsonProtocol.java @@ -76,6 +76,7 @@ public class NsonProtocol { public static String PREPARED_STATEMENT = "ps"; public static String QUERY = "q"; public static String QUERY_BRANCHES = "qb"; + public static String QUERY_BRANCH_STORES = "qbs"; public static String QUERY_NAME = "qn"; public static String QUERY_OPERATION_NUM = "on"; public static String QUERY_VERSION = "qv"; @@ -89,9 +90,11 @@ public class NsonProtocol { public static String RETURN_ROW = "rr"; public static String SERVER_MEMORY_CONSUMPTION = "sm"; public static String SHARD_ID = "si"; + public static String STORE_TOPO_SEQ_NUMS = "sts"; public static String START = "sr"; public static String STATEMENT = "st"; public static String STORAGE_THROTTLE_COUNT = "sl"; + public static String STORE_ID = "sid"; public static String SYSTEM = "sy"; public static String TABLES = "tb"; public static String TABLE_DDL = "td"; @@ -138,6 +141,7 @@ public class NsonProtocol { public static String SHARD_IDS = "sa"; public static String SUCCESS = "ss"; public static String TOPOLOGY_INFO = "tp"; + public static String STORE_TOPOLOGY_INFO = "stp"; public static String WM_FAILURE = "wf"; public static String WM_FAIL_INDEX = "wi"; public static String WM_FAIL_RESULT = "wr"; @@ -253,6 +257,7 @@ public class NsonProtocol { {PREPARED_STATEMENT,"PREPARED_STATEMENT"}, {QUERY,"QUERY"}, {QUERY_BRANCHES,"QUERY_BRANCHES"}, + {QUERY_BRANCH_STORES,"QUERY_BRANCH_STORES"}, {QUERY_NAME,"QUERY_NAME"}, {QUERY_OPERATION_NUM,"QUERY_OPERATION_NUM"}, {QUERY_VERSION,"QUERY_VERSION"}, @@ -266,9 +271,11 @@ public class NsonProtocol { {RETURN_ROW,"RETURN_ROW"}, {SERVER_MEMORY_CONSUMPTION,"SERVER_MEMORY_CONSUMPTION"}, {SHARD_ID,"SHARD_ID"}, + {STORE_TOPO_SEQ_NUMS,"STORE_TOPO_SEQ_NUMS"}, {START,"START"}, {STATEMENT,"STATEMENT"}, {STORAGE_THROTTLE_COUNT,"STORAGE_THROTTLE_COUNT"}, + {STORE_ID,"STORE_ID"}, {SYSTEM,"SYSTEM"}, {TABLES,"TABLES"}, {TABLE_DDL,"TABLE_DDL"}, @@ -311,6 +318,7 @@ public class NsonProtocol { {SHARD_IDS,"SHARD_IDS"}, {SUCCESS,"SUCCESS"}, {TOPOLOGY_INFO,"TOPOLOGY_INFO"}, + {STORE_TOPOLOGY_INFO,"STORE_TOPOLOGY_INFO"}, {WM_FAILURE,"WM_FAILURE"}, {WM_FAIL_INDEX,"WM_FAIL_INDEX"}, {WM_FAIL_RESULT,"WM_FAIL_RESULT"}, diff --git a/driver/src/main/java/oracle/nosql/driver/ops/serde/nson/NsonSerializerFactory.java b/driver/src/main/java/oracle/nosql/driver/ops/serde/nson/NsonSerializerFactory.java index 951c400b..7275b59b 100644 --- a/driver/src/main/java/oracle/nosql/driver/ops/serde/nson/NsonSerializerFactory.java +++ b/driver/src/main/java/oracle/nosql/driver/ops/serde/nson/NsonSerializerFactory.java @@ -111,6 +111,13 @@ import oracle.nosql.driver.values.MapValue; public class NsonSerializerFactory implements SerializerFactory { + + /* Maximum number of stores. */ + private static final int MAX_STORES = 10_000; + + /* Maximum number of query branches. */ + private static final int MAX_QUERY_BRANCHES = 10_000; + static private NsonSerializerFactory factory = new NsonSerializerFactory(); /* return the singleton */ @@ -502,6 +509,8 @@ public Result deserialize(Request request, readRow(in, result); } else if (name.equals(TOPOLOGY_INFO)) { readTopologyInfo(in, result); + } else if (name.equals(STORE_TOPOLOGY_INFO)) { + readStoreTopoInfo(in, result); } else { skipUnknownField(walker, name); } @@ -579,6 +588,8 @@ public Result deserialize(Request request, readReturnInfo(in, result); } else if (name.equals(TOPOLOGY_INFO)) { readTopologyInfo(in, result); + } else if (name.equals(STORE_TOPOLOGY_INFO)) { + readStoreTopoInfo(in, result); } else { skipUnknownField(walker, name); } @@ -673,6 +684,8 @@ public Result deserialize(Request request, result.setContinuationKey(Nson.readNsonBinary(in)); } else if (name.equals(TOPOLOGY_INFO)) { readTopologyInfo(in, result); + } else if (name.equals(STORE_TOPOLOGY_INFO)) { + readStoreTopoInfo(in, result); } else { skipUnknownField(walker, name); } @@ -758,6 +771,8 @@ public Result deserialize(Request request, result.setGeneratedValue(Nson.readFieldValue(in)); } else if (name.equals(TOPOLOGY_INFO)) { readTopologyInfo(in, result); + } else if (name.equals(STORE_TOPOLOGY_INFO)) { + readStoreTopoInfo(in, result); } else { skipUnknownField(walker, name); } @@ -875,8 +890,8 @@ public void serialize(Request request, if (isPrepared) { QueryDriver driver = rq.getDriver(); int unionBranch = (driver != null ? driver.getUnionBranch() : 0); - byte[] proxyPlan = rq.getPreparedStatement(). - getProxyStatement(unionBranch); + PreparedStatement pstmt = rq.getPreparedStatement(); + byte[] proxyPlan = pstmt.getProxyStatement(unionBranch); writeMapField(ns, IS_PREPARED, isPrepared); writeMapField(ns, IS_SIMPLE_QUERY, rq.isSimpleQuery()); writeMapField(ns, PREPARED_QUERY, proxyPlan); @@ -892,6 +907,8 @@ public void serialize(Request request, } writeBindVariables(ns, out, rq.getPreparedStatement().getVariables()); + writeMapField(ns, STORE_ID, + pstmt.getBranchStoreName(unionBranch)); } else { writeMapField(ns, STATEMENT, rq.getStatement()); } @@ -1015,6 +1032,8 @@ private static void deserializePrepareOrQuery( ArrayList namespaces = new ArrayList<>(); ArrayList tableNames = new ArrayList<>(); ArrayList proxyPreparedQueries = new ArrayList<>(); + ArrayList branchStores = null; + int numBranches = -1; DriverPlanInfo dpi = null; String queryPlan = null; @@ -1054,8 +1073,7 @@ private static void deserializePrepareOrQuery( } else if (name.equals(QUERY_BRANCHES)) { readType(in, Nson.TYPE_ARRAY); in.readInt(); /* length of array in bytes */ - int numBranches = in.readInt(); /* number of array elements */ - + numBranches = readBranchCount(in, QUERY_BRANCHES); for (int i = 0; i < numBranches; ++i) { MapWalker walker2 = getMapWalker(in); while (walker2.hasNext()) { @@ -1075,6 +1093,9 @@ private static void deserializePrepareOrQuery( } } + } else if (name.equals(QUERY_BRANCH_STORES)) { + branchStores = readQueryStoreNames(in); + } else if (name.equals(DRIVER_QUERY_PLAN)) { dpi = getDriverPlanInfo(Nson.readNsonBinary(in), queryVersion); @@ -1097,6 +1118,9 @@ private static void deserializePrepareOrQuery( } else if (name.equals(QUERY_OPERATION)) { operation = (byte)Nson.readNsonInt(in); + } else if (name.equals(STORE_TOPOLOGY_INFO)) { + readStoreTopoInfo(in, (qres != null ? qres : pres)); + } else if (name.equals(TOPOLOGY_INFO)) { readTopologyInfo(in, (qres != null ? qres : pres)); @@ -1139,7 +1163,15 @@ private static void deserializePrepareOrQuery( /* QUERY_V3 and earlier return topo differently */ Result res = (qres != null) ? qres : pres; if (res.getTopology() == null && proxyTopoSeqNum >= 0) { - res.setTopology(new TopologyInfo(proxyTopoSeqNum, shardIds)); + res.setTopology(new TopologyInfo(null /* storeName */, + proxyTopoSeqNum, + shardIds)); + } + + if (branchStores != null && branchStores.size() != numBranches) { + throw new IllegalArgumentException( + "Invalid prepared query: QUERY_BRANCH_STORES count does " + + "not match QUERY_BRANCHES count"); } if (qres != null) { @@ -1171,7 +1203,8 @@ private static void deserializePrepareOrQuery( namespaces, tableNames, operation, - maxParallelism); + maxParallelism, + branchStores); if (pres != null) { pres.setPreparedStatement(prep); } else if (qreq != null) { @@ -1184,6 +1217,44 @@ private static void deserializePrepareOrQuery( } } + /* + * Reads the store name for each query branch. Entry i identifies the + * store for branch i, so duplicate store names are valid. + * + * QUERY_BRANCH_STORES: [string, ...] + */ + private static ArrayList readQueryStoreNames(ByteInputStream in) + throws IOException { + + readType(in, Nson.TYPE_ARRAY); + in.readInt(); /* length of array in bytes */ + + int numBranches = readBranchCount(in, QUERY_BRANCH_STORES); + + ArrayList storeNames = new ArrayList<>(numBranches); + for (int i = 0; i < numBranches; ++i) { + String storeName = Nson.readNsonString(in); + if (storeName == null || storeName.isBlank()) { + throw new IllegalArgumentException( + "Invalid prepared query: empty branch STORE_ID"); + } + storeNames.add(storeName); + } + return storeNames; + } + + private static int readBranchCount(ByteInputStream in, String field) + throws IOException { + + int numBranches = in.readInt(); + if (numBranches < 0 || numBranches > MAX_QUERY_BRANCHES) { + throw new IllegalArgumentException( + "Invalid prepared query: invalid " + + NsonProtocol.readable(field) + " count"); + } + return numBranches; + } + private static VirtualScan readVirtualScan(ByteInputStream in) throws IOException { @@ -1488,6 +1559,8 @@ public void serialize(Request request, } writeMapField(ns, OP_CODE, OpCode.WRITE_MULTIPLE.ordinal()); writeMapField(ns, TIMEOUT, rq.getTimeoutInternal()); + writeMapField(ns, TOPO_SEQ_NUM, rq.topoSeqNum()); + writeStoreTopoSeqs(ns, rq.getStoreTopoSeqNums()); endMap(ns, HEADER); /* @@ -1598,6 +1671,8 @@ public Result deserialize(Request request, } } else if (name.equals(TOPOLOGY_INFO)) { readTopologyInfo(in, result); + } else if (name.equals(STORE_TOPOLOGY_INFO)) { + readStoreTopoInfo(in, result); } else { skipUnknownField(walker, name); } @@ -2398,6 +2473,7 @@ private static void throwTypeMismatch(int expected, int found) { * version (int) * operation (int) * sequence number of cached topology, if available + * per-store-topo sequences, if avaiable * timeout (int) * tableName if available * it is helpful to have the tableName available as early as possible @@ -2415,6 +2491,7 @@ protected static void writeHeader(NsonSerializer ns, int op, Request rq) } writeMapField(ns, OP_CODE, op); writeMapField(ns, TOPO_SEQ_NUM, rq.topoSeqNum()); + writeStoreTopoSeqs(ns, rq.getStoreTopoSeqNums()); writeMapField(ns, TIMEOUT, rq.getTimeoutInternal()); if (rq.getPreferThrottling()) { writeMapField(ns, PREFER_THROTTLING, true); @@ -2424,6 +2501,38 @@ protected static void writeHeader(NsonSerializer ns, int op, Request rq) } } + /** + * Writes per-store topology sequence numbers in the request header. + * Each entry identifies a store and its cached topology sequence, it + * does not include shard IDs. + * + * STORE_TOPO_SEQ_NUMS: [ + * { + * STORE_ID: string, + * TOPO_SEQ_NUM: integer + * }, + * .. + * ] + */ + protected static void writeStoreTopoSeqs(NsonSerializer ns, + Map topoSeqs) + throws IOException { + + if (topoSeqs == null) { + return; + } + + startArray(ns, STORE_TOPO_SEQ_NUMS); + for (Map.Entry e : topoSeqs.entrySet()) { + ns.startMap(0); + writeMapField(ns, STORE_ID, e.getKey()); + writeMapField(ns, TOPO_SEQ_NUM, e.getValue()); + ns.endMap(0); + ns.incrSize(1); + } + endArray(ns, STORE_TOPO_SEQ_NUMS); + } + /** * Writes consistency * "consistency": { @@ -2776,13 +2885,14 @@ static void readConsumedCapacity(ByteInputStream in, } /* + * Read the legacy global topology information: + * * "topology_info" : { * "PROXY_TOPO_SEQNUM" : int * "SHARD_IDS" : [ int, ... ] * } */ - static void readTopologyInfo(ByteInputStream in, - Result result) + static void readTopologyInfo(ByteInputStream in, Result result) throws IOException { int proxyTopoSeqNum = -1; @@ -2803,11 +2913,95 @@ static void readTopologyInfo(ByteInputStream in, TopologyInfo ti = null; if (proxyTopoSeqNum >= 0) { - ti = new TopologyInfo(proxyTopoSeqNum, shardIds); + ti = new TopologyInfo(null /* storeName */, + proxyTopoSeqNum, + shardIds); result.setTopology(ti); } } + /* + * Reads the per-store topology array. + * + * "STORE_TOPOLOGY_INFO" : [ + * { + * "STORE_ID": string, + * "PROXY_TOPO_SEQNUM" : int, + * "SHARD_IDS" : [ int, ... ] + * }, + * ... + * ] + */ + protected static void readStoreTopoInfo(ByteInputStream in, + Result result) + throws IOException { + + readType(in, Nson.TYPE_ARRAY); + in.readInt(); /* length of array in bytes */ + int numTopos = in.readInt(); + if (numTopos < 0 || numTopos > MAX_STORES) { + throw new IllegalArgumentException( + "Invalid number of topology entries: " + numTopos); + } + + ArrayList storeTopos = new ArrayList<>(numTopos); + for (int i = 0; i < numTopos; ++i) { + TopologyInfo ti = readStoreTopoEntry(in); + if (ti != null) { + storeTopos.add(ti); + } + } + + result.setStoreTopologies(storeTopos); + } + + /* + * Reads a store topology entry: + * { + * "STORE_ID" : string, + * "PROXY_TOPO_SEQNUM" : int, + * "SHARD_IDS" : [ int, ... ] + * } + */ + private static TopologyInfo readStoreTopoEntry(ByteInputStream in) + throws IOException { + + String storeName = null; + int proxyTopoSeqNum = -1; + int[] shardIds = null; + MapWalker walker = new MapWalker(in); + + while (walker.hasNext()) { + walker.next(); + String name = walker.getCurrentName(); + if (name.equals(STORE_ID)) { + storeName = Nson.readNsonString(in); + } else if (name.equals(PROXY_TOPO_SEQNUM)) { + proxyTopoSeqNum = Nson.readNsonInt(in); + } else if (name.equals(SHARD_IDS)) { + shardIds = readNsonIntArray(in); + } else { + skipUnknownField(walker, name); + } + } + + if (storeName == null) { + throw new IllegalArgumentException( + "Per-store topology entry is missing store id"); + } + if (proxyTopoSeqNum < 0) { + throw new IllegalArgumentException( + "Per-store topology entry has invalid topoSeqNum: " + + proxyTopoSeqNum); + } + if (shardIds == null) { + throw new IllegalArgumentException( + "Per-store topology entry is missing shard IDs"); + } + + return new TopologyInfo(storeName, proxyTopoSeqNum, shardIds); + } + // TODO: move this to Nson static int[] readNsonIntArray(ByteInputStream bis) throws IOException { diff --git a/driver/src/main/java/oracle/nosql/driver/query/QueryDriver.java b/driver/src/main/java/oracle/nosql/driver/query/QueryDriver.java index 26b02720..1305ade1 100644 --- a/driver/src/main/java/oracle/nosql/driver/query/QueryDriver.java +++ b/driver/src/main/java/oracle/nosql/driver/query/QueryDriver.java @@ -8,9 +8,12 @@ package oracle.nosql.driver.query; import java.util.ArrayList; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.TimeoutException; import oracle.nosql.driver.NoSQLException; +import oracle.nosql.driver.PrepareQueryException; import oracle.nosql.driver.RequestTimeoutException; import oracle.nosql.driver.RetryableException; import oracle.nosql.driver.http.Client; @@ -57,6 +60,13 @@ public class QueryDriver { private RuntimeControlBlock theRCB; + /* + * The topology snapshots used by all requests during this execution. + * When per-store topology is enabled, the array contains one snapshot for + * each query branch. Otherwise, it contains the single legacy snapshot. + */ + private TopologyInfo[] theBaseTopos; + /* * The max number of results the app will receive per NoSQLHandle.query() * invocation @@ -85,6 +95,62 @@ QueryRequest getRequest() { return theRequest; } + TopologyInfo[] getBaseTopos() { + if (theBaseTopos == null) { + initializeBaseTopos(); + } + return theBaseTopos; + } + + /* + * Freezes the topologies used by this query execution and records their + * sequence numbers on the request. + */ + private void initializeBaseTopos() { + TopologyInfo legacyTopo = theClient.getTopology(); + Map storeTopos = theClient.getStoreTopoSnapshot(); + + PreparedStatement prep = theRequest.getPreparedStatement(); + String branchStoreName = prep.getBranchStoreName(0); + if (branchStoreName == null) { + if (legacyTopo == null) { + throw new PrepareQueryException( + "Missing legacy topology for prepared query. " + + "Prepare the query again."); + } + theBaseTopos = new TopologyInfo[] { legacyTopo }; + } else { + int numBranches = prep.getNumBranches(); + theBaseTopos = new TopologyInfo[numBranches]; + for (int branch = 0; branch < numBranches; ++branch) { + branchStoreName = prep.getBranchStoreName(branch); + TopologyInfo topo = storeTopos.get(branchStoreName); + if (topo == null) { + throw new PrepareQueryException( + "Missing topology for prepared query store " + + branchStoreName + ". Prepare the query again."); + } + theBaseTopos[branch] = topo; + } + } + + /* + * The initial query request is stamped with the cached topology before + * it is sent to the proxy. Its prepare response may then refresh the + * client cache. The driver plan above uses the refreshed cache, so + * update this request as well. Internal requests copy these fields; + * without this update, they could send the old topology while the + * driver plan executes with the refreshed topology. + */ + Map storeTopoSeqs = new HashMap<>(); + for (Map.Entry e : storeTopos.entrySet()) { + storeTopoSeqs.put(e.getKey(), e.getValue().getSeqNum()); + } + theRequest.setExecutionTopology( + (legacyTopo == null ? -1 : legacyTopo.getSeqNum()), + storeTopoSeqs); + } + public RuntimeControlBlock getRCB() { return theRCB; } diff --git a/driver/src/main/java/oracle/nosql/driver/query/RuntimeControlBlock.java b/driver/src/main/java/oracle/nosql/driver/query/RuntimeControlBlock.java index 51ad2212..2416dfeb 100644 --- a/driver/src/main/java/oracle/nosql/driver/query/RuntimeControlBlock.java +++ b/driver/src/main/java/oracle/nosql/driver/query/RuntimeControlBlock.java @@ -30,7 +30,12 @@ public class RuntimeControlBlock { */ private final QueryDriver theQueryDriver; - private final TopologyInfo theBaseTopo; + /* + * The topology snapshots used by all requests during this execution. + * When per-store topology is enabled, the array contains one snapshot for + * each query branch. Otherwise, it contains the single legacy snapshot. + */ + private final TopologyInfo[] theBaseTopos; /* * An array storing the values of the extenrnal variables set for the @@ -99,7 +104,7 @@ public RuntimeControlBlock( theIteratorStates = new PlanIterState[numIters]; theRegisters = new FieldValue[numRegs]; theExternalVars = externalVars; - theBaseTopo = getClient().getTopology(); + theBaseTopos = driver.getBaseTopos(); } public int getTraceLevel() { @@ -139,7 +144,10 @@ public QueryRequest getRequest() { } TopologyInfo getBaseTopo() { - return theBaseTopo; + if (theBaseTopos.length == 1) { + return theBaseTopos[0]; + } + return theBaseTopos[theUnionBranch]; } Consistency getConsistency() { @@ -203,10 +211,6 @@ void setUnionBranch(int b) { theUnionBranch = b; } - void incUnionBranch() { - ++theUnionBranch; - } - public void setState(int pos, PlanIterState state) { theIteratorStates[pos] = state; } diff --git a/driver/src/main/java/oracle/nosql/driver/query/TopologyInfo.java b/driver/src/main/java/oracle/nosql/driver/query/TopologyInfo.java index ab301a51..97d82699 100644 --- a/driver/src/main/java/oracle/nosql/driver/query/TopologyInfo.java +++ b/driver/src/main/java/oracle/nosql/driver/query/TopologyInfo.java @@ -8,19 +8,36 @@ package oracle.nosql.driver.query; import java.util.Arrays; +import java.util.Objects; public class TopologyInfo { - private int theSeqNum = -1; + private final String theStoreName; - private int[] theShardIds; + private final int theSeqNum; - public TopologyInfo(int seqNum, int[] shardIds) { - if (shardIds == null) { - throw new IllegalArgumentException("TopologyInfo shardIds must not be null"); + private final int[] theShardIds; + + public TopologyInfo(String storeName, int seqNum, int[] shardIds) { + if (storeName != null && storeName.isBlank()) { + throw new IllegalArgumentException( + "TopologyInfo storeName must not be empty string"); + } + if (seqNum < 0) { + throw new IllegalArgumentException( + "TopologyInfo seqNum must not be negative"); + } + if (shardIds == null || shardIds.length == 0) { + throw new IllegalArgumentException( + "TopologyInfo shardIds must not be null or empty"); } + theStoreName = storeName; theSeqNum = seqNum; - theShardIds = shardIds; + theShardIds = shardIds.clone(); + } + + public String getStoreName() { + return theStoreName; } public int getSeqNum() { @@ -38,19 +55,23 @@ int getShardId(int i) { int getLastShardId() { return theShardIds[theShardIds.length-1]; } - + int[] getShardIds() { - return theShardIds; + return theShardIds.clone(); } @Override public boolean equals(Object o) { + if (!(o instanceof TopologyInfo)) { + return false; + } TopologyInfo other = (TopologyInfo)o; if (this == other || - theSeqNum == other.theSeqNum || - Arrays.equals(theShardIds, other.theShardIds)) { + (Objects.equals(theStoreName, other.theStoreName) && + theSeqNum == other.theSeqNum && + Arrays.equals(theShardIds, other.theShardIds))) { return true; } @@ -59,13 +80,20 @@ public boolean equals(Object o) { @Override public int hashCode() { - return theSeqNum; + int code = 1; + code = 31 * code + (theStoreName != null ? theStoreName.hashCode() : 0); + code = 31 * code + theSeqNum; + code = 31 * code + Arrays.hashCode(theShardIds); + return code; } @Override public String toString() { StringBuilder sb = new StringBuilder(); + if (theStoreName != null) { + sb.append("storeName = ").append(theStoreName).append(" "); + } sb.append("seqNum = ").append(theSeqNum); sb.append(" shards ids = [ "); for (int sid : theShardIds) { diff --git a/driver/src/main/java/oracle/nosql/driver/query/UnionIter.java b/driver/src/main/java/oracle/nosql/driver/query/UnionIter.java index 537782b5..925ed63f 100644 --- a/driver/src/main/java/oracle/nosql/driver/query/UnionIter.java +++ b/driver/src/main/java/oracle/nosql/driver/query/UnionIter.java @@ -146,14 +146,17 @@ public void open(RuntimeControlBlock rcb) { if (theSortFields == null) { PlanIter branch = theBranches[state.theCurrentBranch]; + rcb.setUnionBranch(state.theCurrentBranch); branch.open(rcb); } else { for (int i = 0; i < theBranches.length; ++i) { PlanIter branch = theBranches[i]; + rcb.setUnionBranch(i); branch.open(rcb); SortedBranch sb = new SortedBranch(rcb, branch, i); state.theSortedBranches.add(sb); } + rcb.setUnionBranch(0); } } @@ -197,6 +200,7 @@ private boolean simpleNext(RuntimeControlBlock rcb) { while (state.theCurrentBranch < theBranches.length) { PlanIter branch = theBranches[state.theCurrentBranch]; + rcb.setUnionBranch(state.theCurrentBranch); boolean more = branch.next(rcb); @@ -215,7 +219,6 @@ private boolean simpleNext(RuntimeControlBlock rcb) { } ++state.theCurrentBranch; - rcb.incUnionBranch(); if (rcb.getTraceLevel() >= 3) { rcb.trace("UNION: moved to branch " + state.theCurrentBranch); @@ -223,6 +226,7 @@ private boolean simpleNext(RuntimeControlBlock rcb) { if (state.theCurrentBranch < theBranches.length) { branch = theBranches[state.theCurrentBranch]; + rcb.setUnionBranch(state.theCurrentBranch); branch.open(rcb); /* For simplicity, we don't want to allow the possibility of diff --git a/driver/src/test/java/oracle/nosql/driver/RequestTest.java b/driver/src/test/java/oracle/nosql/driver/RequestTest.java index 9990b3e0..712a56b0 100644 --- a/driver/src/test/java/oracle/nosql/driver/RequestTest.java +++ b/driver/src/test/java/oracle/nosql/driver/RequestTest.java @@ -113,7 +113,8 @@ public void testPreparedStatementDoesNotExposeMutableProxyBytes() { namespaces, tableNames, (byte)5, - 0); + 0, + null); proxyStatement[0] = 9; proxyStatements.set(0, new byte[] {9, 9, 9});