This is an automated email from the ASF dual-hosted git repository.

amashenkov pushed a commit to branch ignite-16004
in repository https://gitbox.apache.org/repos/asf/ignite-3.git

commit 90c5c0a03b6c17755d33f4724d1664c513ace39f
Author: Andrew Mashenkov <[email protected]>
AuthorDate: Mon Dec 13 14:57:53 2021 +0300

    Implement bulk operations.
---
 .../ignite/internal/table/KeyValueViewImpl.java    | 288 +++++++++++----------
 .../ignite/internal/table/RecordViewImpl.java      | 109 ++++++--
 .../KeyValueViewOperationsSimpleSchemaTest.java    |   4 +-
 .../internal/table/KeyValueViewOperationsTest.java |  25 ++
 .../table/RecordBinaryViewOperationsTest.java      |   1 -
 .../internal/table/RecordViewOperationsTest.java   |  25 +-
 .../ignite/internal/table/TxAbstractTest.java      |   6 +-
 7 files changed, 287 insertions(+), 171 deletions(-)

diff --git 
a/modules/table/src/main/java/org/apache/ignite/internal/table/KeyValueViewImpl.java
 
b/modules/table/src/main/java/org/apache/ignite/internal/table/KeyValueViewImpl.java
index 37fd4ad..809b297 100644
--- 
a/modules/table/src/main/java/org/apache/ignite/internal/table/KeyValueViewImpl.java
+++ 
b/modules/table/src/main/java/org/apache/ignite/internal/table/KeyValueViewImpl.java
@@ -20,12 +20,13 @@ package org.apache.ignite.internal.table;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
-import java.util.stream.Collectors;
 import org.apache.ignite.internal.schema.BinaryRow;
 import org.apache.ignite.internal.schema.SchemaDescriptor;
 import org.apache.ignite.internal.schema.SchemaRegistry;
@@ -33,7 +34,6 @@ import 
org.apache.ignite.internal.schema.marshaller.KvMarshaller;
 import org.apache.ignite.internal.schema.marshaller.MarshallerException;
 import 
org.apache.ignite.internal.schema.marshaller.reflection.KvMarshallerImpl;
 import org.apache.ignite.internal.schema.row.Row;
-import org.apache.ignite.internal.util.Pair;
 import org.apache.ignite.lang.IgniteException;
 import org.apache.ignite.table.InvokeProcessor;
 import org.apache.ignite.table.KeyValueView;
@@ -68,130 +68,85 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         marshallerFactory = (schema) -> new KvMarshallerImpl<>(schema, 
keyMapper, valueMapper);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public V get(@NotNull K key) {
         return sync(getAsync(key));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull
     CompletableFuture<V> getAsync(@NotNull K key) {
         BinaryRow keyRow = marshal(Objects.requireNonNull(key), null);
 
-        return tbl.get(keyRow, tx)
-                       .thenApply(this::unmarshalValue); // row -> 
deserialized obj.
+        return tbl.get(keyRow, tx).thenApply(this::unmarshalValue);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public Map<K, V> getAll(@NotNull Collection<K> keys) {
         return sync(getAllAsync(keys));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull
     CompletableFuture<Map<K, V>> getAllAsync(@NotNull Collection<K> keys) {
         Objects.requireNonNull(keys);
 
-        List<BinaryRow> keyRows = new ArrayList<>(keys.size());
-
-        for (K keyRec : keys) {
-            final BinaryRow keyRow = marshal(keyRec, null);
-
-            keyRows.add(keyRow);
-        }
-
-        return tbl.getAll(keyRows, tx)
-                       .thenApply(ts -> ts.stream().filter(Objects::nonNull)
-                                                .map(this::unmarshalPair)
-                                                
.collect(Collectors.toMap(Pair::getFirst, Pair::getSecond)));
+        return tbl.getAll(marshal(keys), tx).thenApply(this::unmarshalPairs);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public boolean contains(@NotNull K key) {
         return get(key) != null;
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public CompletableFuture<Boolean> containsAsync(@NotNull K key) {
         return getAsync(key).thenApply(Objects::nonNull);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public void put(@NotNull K key, V val) {
         sync(putAsync(key, val));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull
     CompletableFuture<Void> putAsync(@NotNull K key, V val) {
         BinaryRow keyRow = marshal(Objects.requireNonNull(key), val);
 
-        return tbl.upsert(keyRow, tx).thenAccept(ignore -> {
-        });
+        return tbl.upsert(keyRow, tx);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public void putAll(@NotNull Map<K, V> pairs) {
         sync(putAllAsync(pairs));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull
     CompletableFuture<Void> putAllAsync(@NotNull Map<K, V> pairs) {
         Objects.requireNonNull(pairs);
 
-        List<BinaryRow> rows = new ArrayList<>(pairs.size());
-
-        for (Map.Entry<K, V> pair : pairs.entrySet()) {
-            final BinaryRow row = marshal(pair.getKey(), pair.getValue());
-
-            rows.add(row);
-        }
-
-        return tbl.upsertAll(rows, tx);
+        return tbl.upsertAll(marshal(pairs), tx);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public V getAndPut(@NotNull K key, V val) {
         return sync(getAndPutAsync(key, val));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull
     CompletableFuture<V> getAndPutAsync(@NotNull K key, V val) {
@@ -200,17 +155,13 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         return tbl.getAndUpsert(keyRow, tx).thenApply(this::unmarshalValue);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public boolean putIfAbsent(@NotNull K key, @NotNull V val) {
         return sync(putIfAbsentAsync(key, val));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull
     CompletableFuture<Boolean> putIfAbsentAsync(@NotNull K key, V val) {
@@ -219,25 +170,19 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         return tbl.insert(keyRow, tx);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public boolean remove(@NotNull K key) {
         return sync(removeAsync(key));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public boolean remove(@NotNull K key, @NotNull V val) {
         return sync(removeAsync(key, val));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull
     CompletableFuture<Boolean> removeAsync(@NotNull K key) {
@@ -246,9 +191,7 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         return tbl.delete(keyRow, tx);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull CompletableFuture<Boolean> removeAsync(@NotNull K key, 
@NotNull V val) {
         BinaryRow keyRow = marshal(Objects.requireNonNull(key), 
Objects.requireNonNull(val));
@@ -256,34 +199,29 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         return tbl.deleteExact(keyRow, tx);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public Collection<K> removeAll(@NotNull Collection<K> keys) {
         return sync(removeAllAsync(keys));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull
     CompletableFuture<Collection<K>> removeAllAsync(@NotNull Collection<K> 
keys) {
-        throw new UnsupportedOperationException("Not implemented yet.");
+        Objects.requireNonNull(keys);
+
+        return tbl.deleteAll(marshal(keys), tx).thenApply(this::unmarshalKeys);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+
+    /** {@inheritDoc} */
     @Override
     public V getAndRemove(@NotNull K key) {
         return sync(getAndRemoveAsync(key));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull
     CompletableFuture<V> getAndRemoveAsync(@NotNull K key) {
@@ -292,25 +230,19 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         return tbl.getAndDelete(keyRow, tx).thenApply(this::unmarshalValue);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public boolean replace(@NotNull K key, V val) {
         return sync(replaceAsync(key, val));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public boolean replace(@NotNull K key, @NotNull V oldVal, @NotNull V 
newVal) {
         return sync(replaceAsync(key, oldVal, newVal));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull CompletableFuture<Boolean> replaceAsync(@NotNull K key, 
@NotNull V val) {
         BinaryRow row = marshal(Objects.requireNonNull(key), 
Objects.requireNonNull(val));
@@ -318,9 +250,7 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         return tbl.replace(row, tx);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull CompletableFuture<Boolean> replaceAsync(@NotNull K key, 
@NotNull V oldVal, @NotNull V newVal) {
         Objects.requireNonNull(key);
@@ -331,17 +261,13 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         return tbl.replace(oldRow, newRow, tx);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public V getAndReplace(@NotNull K key, V val) {
         return sync(getAndReplaceAsync(key, val));
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull
     CompletableFuture<V> getAndReplaceAsync(@NotNull K key, V val) {
@@ -350,18 +276,14 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         return tbl.getAndReplace(row, tx).thenApply(this::unmarshalValue);
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public <R extends Serializable> R invoke(@NotNull K key, 
InvokeProcessor<K, V, R> proc,
             Serializable... args) {
         throw new UnsupportedOperationException("Not implemented yet.");
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull <R extends Serializable> CompletableFuture<R> invokeAsync(
             @NotNull K key,
@@ -371,9 +293,7 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         throw new UnsupportedOperationException("Not implemented yet.");
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public <R extends Serializable> Map<K, R> invokeAll(
             @NotNull Collection<K> keys,
@@ -383,9 +303,7 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         throw new UnsupportedOperationException("Not implemented yet.");
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public @NotNull <R extends Serializable> CompletableFuture<Map<K, R>> 
invokeAllAsync(
             @NotNull Collection<K> keys,
@@ -394,9 +312,7 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
         throw new UnsupportedOperationException("Not implemented yet.");
     }
 
-    /**
-     * {@inheritDoc}
-     */
+    /** {@inheritDoc} */
     @Override
     public KeyValueViewImpl<K, V> withTransaction(Transaction tx) {
         throw new UnsupportedOperationException("Not implemented yet.");
@@ -442,42 +358,130 @@ public class KeyValueViewImpl<K, V> extends 
AbstractTableView implements KeyValu
     }
 
     /**
-     * Unmarshal value object from given binary row.
+     * Marshal key-value pair to a row.
      *
-     * @param binaryRow Binary row.
-     * @return Value object.
+     * @param key Key object.
+     * @param val Value object.
+     * @return Binary row.
      */
-    private Pair<K, V> unmarshalPair(BinaryRow binaryRow) {
-        if (binaryRow == null) {
-            return null;
+    private BinaryRow marshal(@NotNull K key, V val) {
+        final KvMarshaller<K, V> marsh = 
marshaller(schemaReg.lastSchemaVersion());
+
+        try {
+            return marsh.marshal(key, val);
+        } catch (MarshallerException e) {
+            throw new IgniteException(e);
         }
+    }
 
-        Row row = schemaReg.resolve(binaryRow);
+    /**
+     * Marshal keys to a row.
+     *
+     * @param keys Key objects.
+     * @return Binary rows.
+     */
+    @NotNull
+    public Collection<BinaryRow> marshal(@NotNull Collection<K> keys) {
+        final KvMarshaller<K, V> marsh = 
marshaller(schemaReg.lastSchemaVersion());
 
-        KvMarshaller<K, V> marshaller = marshaller(row.schemaVersion());
+        List<BinaryRow> keyRows = new ArrayList<>(keys.size());
 
         try {
-            return new Pair<>(marshaller.unmarshalKey(row), 
marshaller.unmarshalValue(row));
+            for (K key : keys) {
+                final BinaryRow keyRow = 
marsh.marshal(Objects.requireNonNull(key), null);
+
+                keyRows.add(keyRow);
+            }
         } catch (MarshallerException e) {
             throw new IgniteException(e);
         }
+
+        return keyRows;
     }
 
     /**
-     * Marshal key-value pair to a row.
+     * Marshal key-value pairs.
      *
-     * @param key Key object.
-     * @param val Value object.
-     * @return Binary row.
+     * @param pairs Key-value map.
+     * @return Binary rows.
      */
-    private BinaryRow marshal(@NotNull K key, V val) {
+    @NotNull
+    public List<BinaryRow> marshal(@NotNull Map<K, V> pairs) {
         final KvMarshaller<K, V> marsh = 
marshaller(schemaReg.lastSchemaVersion());
 
+        List<BinaryRow> rows = new ArrayList<>(pairs.size());
+
         try {
-            return marsh.marshal(key, val);
+            for (Map.Entry<K, V> pair : pairs.entrySet()) {
+                final BinaryRow row = 
marsh.marshal(Objects.requireNonNull(pair.getKey()), pair.getValue());
+
+                rows.add(row);
+            }
+        } catch (MarshallerException e) {
+            throw new IgniteException(e);
+        }
+
+        return rows;
+    }
+
+    /**
+     * Marshal keys.
+     *
+     * @param rows Binary rows.
+     * @return Keys.
+     */
+    @NotNull
+    public Collection<K> unmarshalKeys(Collection<BinaryRow> rows) {
+        if (rows.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        final KvMarshaller<K, V> marsh = 
marshaller(schemaReg.lastSchemaVersion());
+
+        List<K> keys = new ArrayList<>(rows.size());
+
+        try {
+            for (BinaryRow row : rows) {
+                if (row != null) {
+                    keys.add(marsh.unmarshalKey(schemaReg.resolve(row)));
+                }
+            }
+
+            return keys;
         } catch (MarshallerException e) {
             throw new IgniteException(e);
         }
     }
 
+
+    /**
+     * Marshal key-value pairs.
+     *
+     * @param rows Binary rows.
+     * @return Key-value pairs.
+     */
+    @NotNull
+    public Map<K, V> unmarshalPairs(Collection<BinaryRow> rows) {
+        if (rows.isEmpty()) {
+            return Collections.emptyMap();
+        }
+
+        final KvMarshaller<K, V> marsh = 
marshaller(schemaReg.lastSchemaVersion());
+
+        Map<K, V> pairs = new HashMap<>(rows.size());
+
+        try {
+            for (BinaryRow row : rows) {
+                if (row != null) {
+                    Row row0 = schemaReg.resolve(row);
+
+                    pairs.put(marsh.unmarshalKey(row0), 
marsh.unmarshalValue(row0));
+                }
+            }
+
+            return pairs;
+        } catch (MarshallerException e) {
+            throw new IgniteException(e);
+        }
+    }
 }
diff --git 
a/modules/table/src/main/java/org/apache/ignite/internal/table/RecordViewImpl.java
 
b/modules/table/src/main/java/org/apache/ignite/internal/table/RecordViewImpl.java
index 3d74414..24e5456 100644
--- 
a/modules/table/src/main/java/org/apache/ignite/internal/table/RecordViewImpl.java
+++ 
b/modules/table/src/main/java/org/apache/ignite/internal/table/RecordViewImpl.java
@@ -18,7 +18,10 @@
 package org.apache.ignite.internal.table;
 
 import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
@@ -75,9 +78,7 @@ public class RecordViewImpl<R> extends AbstractTableView 
implements RecordView<R
 
         BinaryRow keyRow = marshalKey(keyRec);  // Convert to portable format 
to pass TX/storage layer.
 
-        return tbl.get(keyRow, tx)  // Load async.
-                .thenApply(this::wrap) // Binary -> schema-aware row
-                .thenApply(this::unmarshal); // Deserialize.
+        return tbl.get(keyRow, tx).thenApply(this::unmarshal);
     }
 
     /** {@inheritDoc} */
@@ -89,7 +90,9 @@ public class RecordViewImpl<R> extends AbstractTableView 
implements RecordView<R
     /** {@inheritDoc} */
     @Override
     public @NotNull CompletableFuture<Collection<R>> getAllAsync(@NotNull 
Collection<R> keyRecs) {
-        throw new UnsupportedOperationException("Not implemented yet.");
+        Objects.requireNonNull(keyRecs);
+
+        return tbl.getAll(marshalKeys(keyRecs), tx).thenApply(this::unmarshal);
     }
 
     /** {@inheritDoc} */
@@ -103,8 +106,7 @@ public class RecordViewImpl<R> extends AbstractTableView 
implements RecordView<R
     public @NotNull CompletableFuture<Void> upsertAsync(@NotNull R rec) {
         BinaryRow keyRow = marshal(Objects.requireNonNull(rec));
 
-        return tbl.upsert(keyRow, tx).thenAccept(ignore -> {
-        });
+        return tbl.upsert(keyRow, tx);
     }
 
     /** {@inheritDoc} */
@@ -116,7 +118,9 @@ public class RecordViewImpl<R> extends AbstractTableView 
implements RecordView<R
     /** {@inheritDoc} */
     @Override
     public @NotNull CompletableFuture<Void> upsertAllAsync(@NotNull 
Collection<R> recs) {
-        throw new UnsupportedOperationException("Not implemented yet.");
+        Objects.requireNonNull(recs);
+
+        return tbl.upsertAll(marshal(recs), tx);
     }
 
     /** {@inheritDoc} */
@@ -156,7 +160,9 @@ public class RecordViewImpl<R> extends AbstractTableView 
implements RecordView<R
     /** {@inheritDoc} */
     @Override
     public @NotNull CompletableFuture<Collection<R>> insertAllAsync(@NotNull 
Collection<R> recs) {
-        throw new UnsupportedOperationException("Not implemented yet.");
+        Objects.requireNonNull(recs);
+
+        return tbl.insertAll(marshal(recs), tx).thenApply(this::unmarshal);
     }
 
     /** {@inheritDoc} */
@@ -253,7 +259,9 @@ public class RecordViewImpl<R> extends AbstractTableView 
implements RecordView<R
     /** {@inheritDoc} */
     @Override
     public @NotNull CompletableFuture<Collection<R>> deleteAllAsync(@NotNull 
Collection<R> recs) {
-        throw new UnsupportedOperationException("Not implemented yet.");
+        Objects.requireNonNull(recs);
+
+        return tbl.deleteAll(marshal(recs), tx).thenApply(this::unmarshal);
     }
 
     /** {@inheritDoc} */
@@ -265,7 +273,9 @@ public class RecordViewImpl<R> extends AbstractTableView 
implements RecordView<R
     /** {@inheritDoc} */
     @Override
     public @NotNull CompletableFuture<Collection<R>> 
deleteAllExactAsync(@NotNull Collection<R> recs) {
-        throw new UnsupportedOperationException("Not implemented yet.");
+        Objects.requireNonNull(recs);
+
+        return tbl.deleteAllExact(marshal(recs), 
tx).thenApply(this::unmarshal);
     }
 
     /** {@inheritDoc} */
@@ -308,8 +318,7 @@ public class RecordViewImpl<R> extends AbstractTableView 
implements RecordView<R
     }
 
     /**
-    /**
-     * Returns marshaller.
+     * /** Returns marshaller.
      *
      * @param schemaVersion Schema version.
      * @return Marshaller.
@@ -381,17 +390,79 @@ public class RecordViewImpl<R> extends AbstractTableView 
implements RecordView<R
     }
 
     /**
-     * Returns schema-aware row.
+     * Marshal records.
      *
-     * @param row Binary row.
+     * @param recs Records collection.
+     * @return Binary rows collection.
      */
-    private Row wrap(BinaryRow row) {
-        if (row == null) {
-            return null;
+    private Collection<BinaryRow> marshal(@NotNull Collection<R> recs) {
+        final RecordMarshaller<R> marsh = 
marshaller(schemaReg.lastSchemaVersion());
+
+        List<BinaryRow> rows = new ArrayList<>(recs.size());
+
+        try {
+            for (R rec : recs) {
+                final BinaryRow row = 
marsh.marshal(Objects.requireNonNull(rec));
+
+                rows.add(row);
+            }
+
+            return rows;
+        } catch (MarshallerException e) {
+            throw new IgniteException(e);
         }
+    }
 
-        final SchemaDescriptor rowSchema = 
schemaReg.schema(row.schemaVersion()); // Get a schema for row.
+    /**
+     * Marshal key-records.
+     *
+     * @param recs Records collection.
+     * @return Binary rows collection.
+     */
+    private Collection<BinaryRow> marshalKeys(@NotNull Collection<R> recs) {
+        final RecordMarshaller<R> marsh = 
marshaller(schemaReg.lastSchemaVersion());
+
+        List<BinaryRow> rows = new ArrayList<>(recs.size());
+
+        try {
+            for (R rec : recs) {
+                final BinaryRow row = 
marsh.marshalKey(Objects.requireNonNull(rec));
+
+                rows.add(row);
+            }
+
+            return rows;
+        } catch (MarshallerException e) {
+            throw new IgniteException(e);
+        }
+    }
 
-        return new Row(rowSchema, row);
+    /**
+     * Unmarshal records.
+     *
+     * @param rows Row collection.
+     * @return Records collection.
+     */
+    @NotNull
+    public Collection<R> unmarshal(Collection<BinaryRow> rows) {
+        if (rows.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        final RecordMarshaller<R> marsh = 
marshaller(schemaReg.lastSchemaVersion());
+
+        List<R> recs = new ArrayList<>(rows.size());
+
+        try {
+            for (Row row : schemaReg.resolve(rows)) {
+                if (row != null) {
+                    recs.add(marsh.unmarshal(row));
+                }
+            }
+
+            return recs;
+        } catch (MarshallerException e) {
+            throw new IgniteException(e);
+        }
     }
 }
diff --git 
a/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsSimpleSchemaTest.java
 
b/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsSimpleSchemaTest.java
index 0e7d86c..ccb8aec 100644
--- 
a/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsSimpleSchemaTest.java
+++ 
b/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsSimpleSchemaTest.java
@@ -327,12 +327,12 @@ public class KeyValueViewOperationsSimpleSchemaTest {
 
         Map<Long, Long> res = kvView.getAll(List.of(1L, 2L, 3L));
 
+        assertEquals(2, res.size());
         assertEquals(11L, res.get(1L));
-        assertNull(res.get(22L));
         assertEquals(33L, res.get(3L));
+        assertNull(res.get(22L));
     }
 
-
     /**
      * Creates key-value view.
      *
diff --git 
a/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsTest.java
 
b/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsTest.java
index 463f255..67bd054 100644
--- 
a/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsTest.java
+++ 
b/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsTest.java
@@ -39,6 +39,8 @@ import static org.mockito.Answers.RETURNS_DEEP_STUBS;
 
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Random;
 import java.util.Set;
@@ -317,6 +319,29 @@ public class KeyValueViewOperationsTest {
         assertNotNull(tbl.get(key));
     }
 
+    @Test
+    public void getAll() {
+        KeyValueView<TestKeyObject, TestObjectWithAllTypes> kvView = kvView();
+
+        final TestKeyObject key1 = TestKeyObject.randomObject(rnd);
+        final TestKeyObject key2 = TestKeyObject.randomObject(rnd);
+        final TestKeyObject key3 = TestKeyObject.randomObject(rnd);
+        final TestObjectWithAllTypes val1 = 
TestObjectWithAllTypes.randomObject(rnd);
+        final TestObjectWithAllTypes val3 = 
TestObjectWithAllTypes.randomObject(rnd);
+
+        kvView.putAll(Map.of(
+                key1, val1,
+                key3, val3
+        ));
+
+        Map<TestKeyObject, TestObjectWithAllTypes> res = 
kvView.getAll(List.of(key1, key2, key3));
+
+        assertEquals(2, res.size());
+        assertEquals(val1, res.get(key1));
+        assertEquals(val3, res.get(key3));
+        assertNull(res.get(key2));
+    }
+
     /**
      * Creates key-value view.
      */
diff --git 
a/modules/table/src/test/java/org/apache/ignite/internal/table/RecordBinaryViewOperationsTest.java
 
b/modules/table/src/test/java/org/apache/ignite/internal/table/RecordBinaryViewOperationsTest.java
index b8fbc7c..a15ae25 100644
--- 
a/modules/table/src/test/java/org/apache/ignite/internal/table/RecordBinaryViewOperationsTest.java
+++ 
b/modules/table/src/test/java/org/apache/ignite/internal/table/RecordBinaryViewOperationsTest.java
@@ -360,7 +360,6 @@ public class RecordBinaryViewOperationsTest {
         assertTrue(res.contains(rec1));
     }
 
-
     /**
      * Check tuples equality.
      *
diff --git 
a/modules/table/src/test/java/org/apache/ignite/internal/table/RecordViewOperationsTest.java
 
b/modules/table/src/test/java/org/apache/ignite/internal/table/RecordViewOperationsTest.java
index fdbc6ea..32fd477 100644
--- 
a/modules/table/src/test/java/org/apache/ignite/internal/table/RecordViewOperationsTest.java
+++ 
b/modules/table/src/test/java/org/apache/ignite/internal/table/RecordViewOperationsTest.java
@@ -36,7 +36,9 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Answers.RETURNS_DEEP_STUBS;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Random;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -271,6 +273,25 @@ public class RecordViewOperationsTest {
         assertNull(tbl.get(key));
     }
 
+    @Test
+    public void getAll() {
+        final TestObjectWithAllTypes key1 = key(rnd);
+        final TestObjectWithAllTypes key2 = key(rnd);
+        final TestObjectWithAllTypes key3 = key(rnd);
+        final TestObjectWithAllTypes val1 = randomObject(rnd, key1);
+        final TestObjectWithAllTypes val3 = randomObject(rnd, key3);
+
+        RecordView<TestObjectWithAllTypes> tbl = recordView();
+
+        tbl.upsertAll(List.of(val1, val3));
+
+        Collection<TestObjectWithAllTypes> res = tbl.getAll(List.of(key1, 
/*key2,*/ key3));
+
+        assertEquals(2, res.size());
+        assertTrue(res.contains(val1));
+        assertTrue(res.contains(val3));
+    }
+
     /**
      * Creates RecordView.
      */
@@ -323,9 +344,9 @@ public class RecordViewOperationsTest {
 
         // Validate all types are tested.
         Set<NativeTypeSpec> testedTypes = Arrays.stream(valCols).map(c -> 
c.type().spec())
-                .collect(Collectors.toSet());
+                                                  .collect(Collectors.toSet());
         Set<NativeTypeSpec> missedTypes = 
Arrays.stream(NativeTypeSpec.values())
-                .filter(t -> 
!testedTypes.contains(t)).collect(Collectors.toSet());
+                                                  .filter(t -> 
!testedTypes.contains(t)).collect(Collectors.toSet());
 
         assertEquals(Collections.emptySet(), missedTypes);
 
diff --git 
a/modules/table/src/test/java/org/apache/ignite/internal/table/TxAbstractTest.java
 
b/modules/table/src/test/java/org/apache/ignite/internal/table/TxAbstractTest.java
index 977f2ad..ef6c008 100644
--- 
a/modules/table/src/test/java/org/apache/ignite/internal/table/TxAbstractTest.java
+++ 
b/modules/table/src/test/java/org/apache/ignite/internal/table/TxAbstractTest.java
@@ -541,11 +541,7 @@ public abstract class TxAbstractTest extends 
IgniteAbstractTest {
 
         Collection<Tuple> ret = accounts.recordView().getAll(keys);
 
-        assertEquals(2, ret.size());
-
-        for (Tuple tuple : ret) {
-            assertNull(tuple);
-        }
+        assertEquals(0, ret.size());
 
         accounts.recordView().upsert(makeValue(1, 100.));
         accounts.recordView().upsert(makeValue(2, 200.));

Reply via email to