ascherbakoff commented on code in PR #6007:
URL: https://github.com/apache/ignite-3/pull/6007#discussion_r2158223983
##########
modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java:
##########
@@ -877,12 +877,135 @@ int tryGetPartitionCount() {
return partitionCount;
}
+ /**
+ * Batch with indexes.
+ *
+ * @param <E> Batch type element.
+ */
+ static class Batch<E> {
+ List<E> batch = new ArrayList<>();
+ List<Integer> originalIndices = new ArrayList<>();
+
+ void add(E entry, int origIdx) {
+ batch.add(entry);
+ originalIndices.add(origIdx);
+ }
+ }
+
+ private static <E> void reduceWithKeepOrder(List<E> agg, List<E> cur,
List<Integer> originalIndices) {
+ for (int i = 0; i < cur.size(); i++) {
+ E val = cur.get(i);
+ Integer orig = originalIndices.get(i);
+ agg.set(orig, val);
+ }
+ }
+
+ <R, E> CompletableFuture<R> split(
+ Transaction tx,
+ Collection<E> keys,
+ BiFunction<Collection<E>, PartitionAwarenessProvider,
CompletableFuture<R>> fun,
+ @Nullable R initialValue,
+ Reducer<R> reducer,
+ BiFunction<ClientSchema, E, Integer> hashFunc
+ ) {
+ assert tx != null;
+
+ CompletableFuture<ClientSchema> schemaFut = getSchema(latestSchemaVer);
+ CompletableFuture<List<String>> partitionsFut =
getPartitionAssignment();
+
+ return CompletableFuture.allOf(schemaFut, partitionsFut)
+ .thenCompose(v -> {
+ @Nullable List<String> aff = partitionsFut.getNow(null);
+ Map<Integer, List<E>> mapped = new HashMap<>();
+ List<CompletableFuture<R>> res = new ArrayList<>();
+
+ if (aff == null) {
+ return fun.apply(keys,
PartitionAwarenessProvider.NULL_PROVIDER);
+ }
+
+ for (E key : keys) {
+ ClientSchema schema = schemaFut.getNow(null);
+ int hash = hashFunc.apply(schema, key);
+ Integer part = Math.abs(hash % aff.size());
+ mapped.computeIfAbsent(part, k -> new
ArrayList<>()).add(key);
+ }
+
+ for (Entry<Integer, List<E>> entry : mapped.entrySet()) {
+ res.add(fun.apply(entry.getValue(),
PartitionAwarenessProvider.of(entry.getKey())));
+ }
+
+ return CompletableFuture.allOf(res.toArray(new
CompletableFuture[0])).thenApply(ignored -> {
+ R in = initialValue;
+
+ for (CompletableFuture<R> val : res) {
+ in = reducer.reduce(in, val.getNow(null));
+ }
+
+ return in;
+ });
+ });
+ }
+
+ <E> CompletableFuture<List<E>> split(
+ Transaction tx,
+ Collection<E> keys,
+ BiFunction<Collection<E>, PartitionAwarenessProvider,
CompletableFuture<List<E>>> fun,
+ BiFunction<ClientSchema, E, Integer> hashFunc
+ ) {
+ assert tx != null;
+
+ CompletableFuture<ClientSchema> schemaFut = getSchema(latestSchemaVer);
+ CompletableFuture<List<String>> partitionsFut =
getPartitionAssignment();
+
+ return CompletableFuture.allOf(schemaFut, partitionsFut)
+ .thenCompose(v -> {
+ @Nullable List<String> aff = partitionsFut.getNow(null);
+ Map<Integer, Batch<E>> mapped = new HashMap<>();
+
+ if (aff == null) {
+ return fun.apply(keys,
PartitionAwarenessProvider.NULL_PROVIDER);
+ }
+
+ int idx = 0;
+ for (E key : keys) {
+ ClientSchema schema = schemaFut.getNow(null);
Review Comment:
🆗
##########
modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java:
##########
@@ -877,12 +877,135 @@ int tryGetPartitionCount() {
return partitionCount;
}
+ /**
+ * Batch with indexes.
+ *
+ * @param <E> Batch type element.
+ */
+ static class Batch<E> {
+ List<E> batch = new ArrayList<>();
+ List<Integer> originalIndices = new ArrayList<>();
+
+ void add(E entry, int origIdx) {
+ batch.add(entry);
+ originalIndices.add(origIdx);
+ }
+ }
+
+ private static <E> void reduceWithKeepOrder(List<E> agg, List<E> cur,
List<Integer> originalIndices) {
+ for (int i = 0; i < cur.size(); i++) {
+ E val = cur.get(i);
+ Integer orig = originalIndices.get(i);
+ agg.set(orig, val);
+ }
+ }
+
+ <R, E> CompletableFuture<R> split(
+ Transaction tx,
+ Collection<E> keys,
+ BiFunction<Collection<E>, PartitionAwarenessProvider,
CompletableFuture<R>> fun,
+ @Nullable R initialValue,
+ Reducer<R> reducer,
+ BiFunction<ClientSchema, E, Integer> hashFunc
+ ) {
+ assert tx != null;
+
+ CompletableFuture<ClientSchema> schemaFut = getSchema(latestSchemaVer);
+ CompletableFuture<List<String>> partitionsFut =
getPartitionAssignment();
+
+ return CompletableFuture.allOf(schemaFut, partitionsFut)
+ .thenCompose(v -> {
+ @Nullable List<String> aff = partitionsFut.getNow(null);
+ Map<Integer, List<E>> mapped = new HashMap<>();
+ List<CompletableFuture<R>> res = new ArrayList<>();
Review Comment:
🆗
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
