ascherbakoff commented on code in PR #6007:
URL: https://github.com/apache/ignite-3/pull/6007#discussion_r2157074103
##########
modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java:
##########
@@ -877,12 +886,148 @@ 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 -> {
+ List<E> unmapped = new ArrayList<>();
+ Map<Integer, List<E>> mapped = new HashMap<>();
+ for (E key : keys) {
+ ClientSchema schema = schemaFut.getNow(null);
+ @Nullable List<String> aff =
partitionsFut.getNow(null);
+ int hash = hashFunc.apply(schema, key);
+ Integer part = aff == null ? null : Math.abs(hash %
aff.size());
Review Comment:
Good catch. Simplified the splitting code.
##########
modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java:
##########
@@ -877,12 +886,148 @@ 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 -> {
+ List<E> unmapped = new ArrayList<>();
+ Map<Integer, List<E>> mapped = new HashMap<>();
+ for (E key : keys) {
+ ClientSchema schema = schemaFut.getNow(null);
+ @Nullable List<String> aff =
partitionsFut.getNow(null);
+ int hash = hashFunc.apply(schema, key);
+ Integer part = aff == null ? null : Math.abs(hash %
aff.size());
+ if (part == null) {
+ unmapped.add(key);
+ } else {
+ mapped.computeIfAbsent(part, k -> new
ArrayList<>()).add(key);
+ }
+ }
+
+ List<CompletableFuture<R>> res = new ArrayList<>();
+
+ if (!unmapped.isEmpty()) {
+ // Disable awareness for unmapped keys.
+ res.add(fun.apply(unmapped,
PartitionAwarenessProvider.NULL_PROVIDER));
+ }
+
+ 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 -> {
+ Batch<E> unmapped = new Batch<>();
+ Map<Integer, Batch<E>> mapped = new HashMap<>();
+ int idx = 0;
+ for (E key : keys) {
+ ClientSchema schema = schemaFut.getNow(null);
+ @Nullable List<String> aff =
partitionsFut.getNow(null);
+ int hash = hashFunc.apply(schema, key);
+ Integer part = aff == null ? null : Math.abs(hash %
aff.size());
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
