You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pt...@apache.org on 2024/02/14 15:36:16 UTC
(ignite-3) branch main updated: IGNITE-21403 Add DataStreamer data removal API (#3208)
This is an automated email from the ASF dual-hosted git repository.
ptupitsyn pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 231593e94b IGNITE-21403 Add DataStreamer data removal API (#3208)
231593e94b is described below
commit 231593e94b4d20985b31445042f26391f0fa7d30
Author: Pavel Tupitsyn <pt...@apache.org>
AuthorDate: Wed Feb 14 17:36:10 2024 +0200
IGNITE-21403 Add DataStreamer data removal API (#3208)
Change `DataStreamer` API to allow data removal:
* `DataStreamerTarget` now accepts `Publisher<DataStreamerItem<T>>` instead of `Publisher<T>`
* `DataStreamerItem.of` for add/update
* `DataStreamerItem.removed` for delete
---
.../org/apache/ignite/table/DataStreamerItem.java | 75 +++++++++++++
...rTarget.java => DataStreamerOperationType.java} | 26 ++---
.../apache/ignite/table/DataStreamerTarget.java | 2 +-
.../table/ClientStreamerBatchSendRequest.java | 28 ++++-
.../internal/client/table/ClientDataStreamer.java | 3 +-
.../client/table/ClientKeyValueBinaryView.java | 5 +-
.../internal/client/table/ClientKeyValueView.java | 28 ++++-
.../client/table/ClientRecordBinaryView.java | 3 +-
.../client/table/ClientRecordSerializer.java | 16 ++-
.../internal/client/table/ClientRecordView.java | 3 +-
.../client/table/ClientTupleSerializer.java | 14 ++-
.../apache/ignite/client/ClientMetricsTest.java | 7 +-
.../org/apache/ignite/client/DataStreamerTest.java | 57 ++++++++--
.../ignite/client/PartitionAwarenessTest.java | 14 ++-
.../ignite/internal/streamer/StreamerBuffer.java | 28 +++--
.../internal/streamer/StreamerSubscriber.java | 19 ++--
.../internal/streamer/StreamerSubscriberTest.java | 8 +-
.../ignite/internal/streamer/SimplePublisher.java} | 39 ++++---
.../streamer/ItAbstractDataStreamerTest.java | 124 +++++++++++++++++----
.../ignite/internal/table/ItInternalTableTest.java | 38 +++++++
.../apache/ignite/internal/table/DataStreamer.java | 3 +-
.../internal/table/KeyValueBinaryViewImpl.java | 18 ++-
.../ignite/internal/table/KeyValueViewImpl.java | 20 +++-
.../internal/table/RecordBinaryViewImpl.java | 30 ++++-
.../ignite/internal/table/RecordViewImpl.java | 25 ++++-
25 files changed, 507 insertions(+), 126 deletions(-)
diff --git a/modules/api/src/main/java/org/apache/ignite/table/DataStreamerItem.java b/modules/api/src/main/java/org/apache/ignite/table/DataStreamerItem.java
new file mode 100644
index 0000000000..0d9123f35c
--- /dev/null
+++ b/modules/api/src/main/java/org/apache/ignite/table/DataStreamerItem.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.table;
+
+/**
+ * Data streamer item: data and operation type.
+ */
+public interface DataStreamerItem<T> {
+ /**
+ * Gets the data.
+ *
+ * @return Data.
+ */
+ T get();
+
+ /**
+ * Gets the operation type.
+ *
+ * @return Operation type.
+ */
+ DataStreamerOperationType operationType();
+
+ /**
+ * Creates a new data streamer item with the given data and {@link DataStreamerOperationType#PUT} operation type.
+ *
+ * @param item Data.
+ * @param <T> Data type.
+ *
+ * @return PUT data streamer item.
+ */
+ static <T> DataStreamerItem<T> of(T item) {
+ return instance(item, DataStreamerOperationType.PUT);
+ }
+
+ /**
+ * Creates a new data streamer item with the given data and {@link DataStreamerOperationType#REMOVE} operation type.
+ *
+ * @param item Data.
+ * @param <T> Data type.
+ *
+ * @return REMOVE data streamer item.
+ */
+ static <T> DataStreamerItem<T> removed(T item) {
+ return instance(item, DataStreamerOperationType.REMOVE);
+ }
+
+ private static <T> DataStreamerItem<T> instance(T item, DataStreamerOperationType operationType) {
+ return new DataStreamerItem<>() {
+ @Override
+ public T get() {
+ return item;
+ }
+
+ @Override
+ public DataStreamerOperationType operationType() {
+ return operationType;
+ }
+ };
+ }
+}
\ No newline at end of file
diff --git a/modules/api/src/main/java/org/apache/ignite/table/DataStreamerTarget.java b/modules/api/src/main/java/org/apache/ignite/table/DataStreamerOperationType.java
similarity index 55%
copy from modules/api/src/main/java/org/apache/ignite/table/DataStreamerTarget.java
copy to modules/api/src/main/java/org/apache/ignite/table/DataStreamerOperationType.java
index a6a1bd3768..bc796a72c1 100644
--- a/modules/api/src/main/java/org/apache/ignite/table/DataStreamerTarget.java
+++ b/modules/api/src/main/java/org/apache/ignite/table/DataStreamerOperationType.java
@@ -17,25 +17,13 @@
package org.apache.ignite.table;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Flow;
-import org.jetbrains.annotations.Nullable;
-
/**
- * Represents an entity that can be used as a target for streaming data.
- *
- * @param <T> Entry type.
+ * Data streamer operation type.
*/
-@SuppressWarnings("InterfaceMayBeAnnotatedFunctional")
-public interface DataStreamerTarget<T> {
- /**
- * Streams data into the underlying table.
- *
- * @param publisher Producer.
- * @param options Options (can be null).
- * @return Future that will be completed when the stream is finished.
- */
- CompletableFuture<Void> streamData(
- Flow.Publisher<T> publisher,
- @Nullable DataStreamerOptions options);
+public enum DataStreamerOperationType {
+ /** Put operation. */
+ PUT,
+
+ /** Remove operation. */
+ REMOVE
}
diff --git a/modules/api/src/main/java/org/apache/ignite/table/DataStreamerTarget.java b/modules/api/src/main/java/org/apache/ignite/table/DataStreamerTarget.java
index a6a1bd3768..3b9e91e8ba 100644
--- a/modules/api/src/main/java/org/apache/ignite/table/DataStreamerTarget.java
+++ b/modules/api/src/main/java/org/apache/ignite/table/DataStreamerTarget.java
@@ -36,6 +36,6 @@ public interface DataStreamerTarget<T> {
* @return Future that will be completed when the stream is finished.
*/
CompletableFuture<Void> streamData(
- Flow.Publisher<T> publisher,
+ Flow.Publisher<DataStreamerItem<T>> publisher,
@Nullable DataStreamerOptions options);
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientStreamerBatchSendRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientStreamerBatchSendRequest.java
index 56819f11d8..6624bb0e15 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientStreamerBatchSendRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientStreamerBatchSendRequest.java
@@ -17,15 +17,21 @@
package org.apache.ignite.client.handler.requests.table;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readSchema;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
-import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import java.util.ArrayList;
import java.util.BitSet;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.table.RecordBinaryViewImpl;
+import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.table.Tuple;
import org.apache.ignite.table.manager.IgniteTables;
+import org.jetbrains.annotations.Nullable;
/**
* Client streamer batch request.
@@ -47,7 +53,7 @@ public class ClientStreamerBatchSendRequest {
return readTableAsync(in, tables).thenCompose(table -> {
int partition = in.unpackInt();
BitSet deleted = in.unpackBitSetNullable();
- return readTuples(in, table, false).thenCompose(tuples -> {
+ return readTuples(in, table, deleted).thenCompose(tuples -> {
RecordBinaryViewImpl recordView = (RecordBinaryViewImpl) table.recordView();
return recordView.updateAll(partition, tuples, deleted)
@@ -55,4 +61,22 @@ public class ClientStreamerBatchSendRequest {
});
});
}
+
+ private static CompletableFuture<List<Tuple>> readTuples(
+ ClientMessageUnpacker unpacker,
+ TableViewInternal table,
+ @Nullable BitSet deleted) {
+ return readSchema(unpacker, table).thenApply(schema -> {
+ var rowCnt = unpacker.unpackInt();
+ var res = new ArrayList<Tuple>(rowCnt);
+
+ for (int i = 0; i < rowCnt; i++) {
+ var keyOnly = deleted != null && deleted.get(i);
+ res.add(readTuple(unpacker, keyOnly, schema));
+ }
+
+ return res;
+ });
+ }
+
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientDataStreamer.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientDataStreamer.java
index 3a1a9aa677..2df4d6a9cc 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientDataStreamer.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientDataStreamer.java
@@ -25,6 +25,7 @@ import org.apache.ignite.internal.streamer.StreamerBatchSender;
import org.apache.ignite.internal.streamer.StreamerOptions;
import org.apache.ignite.internal.streamer.StreamerPartitionAwarenessProvider;
import org.apache.ignite.internal.streamer.StreamerSubscriber;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
/**
@@ -33,7 +34,7 @@ import org.apache.ignite.table.DataStreamerOptions;
class ClientDataStreamer {
@SuppressWarnings("resource")
static <R> CompletableFuture<Void> streamData(
- Publisher<R> publisher,
+ Publisher<DataStreamerItem<R>> publisher,
DataStreamerOptions options,
StreamerBatchSender<R, Integer> batchSender,
StreamerPartitionAwarenessProvider<R, Integer> partitionAwarenessProvider,
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
index 70478e3c03..d4eac97012 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
@@ -39,6 +39,7 @@ import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.lang.NullableValue;
import org.apache.ignite.sql.ResultSetMetadata;
import org.apache.ignite.sql.SqlRow;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.Tuple;
@@ -445,7 +446,9 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> streamData(Publisher<Entry<Tuple, Tuple>> publisher, @Nullable DataStreamerOptions options) {
+ public CompletableFuture<Void> streamData(
+ Publisher<DataStreamerItem<Entry<Tuple, Tuple>>> publisher,
+ @Nullable DataStreamerOptions options) {
Objects.requireNonNull(publisher);
var provider = new KeyValueTupleStreamerPartitionAwarenessProvider(tbl);
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java
index c631eaeaaf..1416fe04bd 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java
@@ -55,6 +55,7 @@ import org.apache.ignite.lang.NullableValue;
import org.apache.ignite.lang.UnexpectedNullValueException;
import org.apache.ignite.sql.ResultSetMetadata;
import org.apache.ignite.sql.SqlRow;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.mapper.Mapper;
@@ -521,7 +522,7 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> streamData(Publisher<Entry<K, V>> publisher, @Nullable DataStreamerOptions options) {
+ public CompletableFuture<Void> streamData(Publisher<DataStreamerItem<Entry<K, V>>> publisher, @Nullable DataStreamerOptions options) {
Objects.requireNonNull(publisher);
var provider = new KeyValuePojoStreamerPartitionAwarenessProvider<K, V>(tbl, keySer.mapper());
@@ -538,8 +539,31 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
w.out().packInt(s.version());
w.out().packInt(items.size());
+ int i = 0;
+
+ Marshaller keyMarsh = s.getMarshaller(keySer.mapper(), TuplePart.KEY, false);
+ Marshaller valMarsh = s.getMarshaller(valSer.mapper(), TuplePart.VAL, false);
+ var noValueSet = new BitSet();
+
for (Entry<K, V> e : items) {
- writeKeyValueRaw(s, w, e.getKey(), e.getValue());
+ boolean del = deleted != null && deleted.get(i++);
+ int colCount = del ? s.keyColumnCount() : s.columns().length;
+
+ noValueSet.clear();
+ var builder = new BinaryTupleBuilder(colCount);
+ ClientMarshallerWriter writer = new ClientMarshallerWriter(builder, noValueSet);
+
+ try {
+ keyMarsh.writeObject(e.getKey(), writer);
+
+ if (!del) {
+ valMarsh.writeObject(e.getValue(), writer);
+ }
+ } catch (MarshallerException ex) {
+ throw new IgniteException(INTERNAL_ERR, ex.getMessage(), ex);
+ }
+
+ w.out().packBinaryTuple(builder, noValueSet);
}
},
r -> null,
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java
index 4b88f9b783..c99a6fca79 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java
@@ -30,6 +30,7 @@ import java.util.concurrent.Flow.Publisher;
import org.apache.ignite.client.RetryLimitPolicy;
import org.apache.ignite.internal.client.proto.ClientOp;
import org.apache.ignite.internal.streamer.StreamerBatchSender;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.Tuple;
@@ -358,7 +359,7 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> streamData(Publisher<Tuple> publisher, @Nullable DataStreamerOptions options) {
+ public CompletableFuture<Void> streamData(Publisher<DataStreamerItem<Tuple>> publisher, @Nullable DataStreamerOptions options) {
Objects.requireNonNull(publisher);
var provider = new TupleStreamerPartitionAwarenessProvider(tbl);
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java
index c3f0417c2c..c3420b428d 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java
@@ -182,10 +182,22 @@ public class ClientRecordSerializer<R> {
w.packInt(recs.size());
Marshaller marshaller = schema.getMarshaller(mapper, TuplePart.KEY_AND_VAL);
- int columnCount = columnCount(schema, TuplePart.KEY_AND_VAL);
+ Marshaller keyMarshaller = deleted == null || deleted.cardinality() == 0
+ ? null
+ : schema.getMarshaller(mapper, TuplePart.KEY);
+
+ int columnCount = schema.columns().length;
+ int keyColumnCount = schema.keyColumnCount();
+
+ int i = 0;
for (R rec : recs) {
- writeRecRaw(rec, w, marshaller, columnCount);
+ boolean del = deleted != null && deleted.get(i++);
+ int colCount = del ? keyColumnCount : columnCount;
+ Marshaller marsh = del ? keyMarshaller : marshaller;
+
+ //noinspection DataFlowIssue (reviewed).
+ writeRecRaw(rec, w, marsh, colCount);
}
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java
index 2b352be99f..2746a906d9 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.streamer.StreamerBatchSender;
import org.apache.ignite.internal.table.criteria.SqlRowProjection;
import org.apache.ignite.sql.ResultSetMetadata;
import org.apache.ignite.sql.SqlRow;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.mapper.Mapper;
@@ -363,7 +364,7 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> streamData(Publisher<R> publisher, @Nullable DataStreamerOptions options) {
+ public CompletableFuture<Void> streamData(Publisher<DataStreamerItem<R>> publisher, @Nullable DataStreamerOptions options) {
Objects.requireNonNull(publisher);
var provider = new PojoStreamerPartitionAwarenessProvider<>(tbl, ser.mapper());
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java
index 81f8f9006a..13a5bb5ebc 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java
@@ -260,8 +260,16 @@ public class ClientTupleSerializer {
w.packInt(schema.version());
w.packInt(pairs.size());
+ int i = 0;
+
for (Map.Entry<Tuple, Tuple> pair : pairs) {
- writeKvTuple(null, pair.getKey(), pair.getValue(), schema, out, true);
+ boolean del = deleted != null && deleted.get(i++);
+
+ if (del) {
+ writeTuple(null, pair.getKey(), schema, out, true, true);
+ } else {
+ writeKvTuple(null, pair.getKey(), pair.getValue(), schema, out, true);
+ }
}
}
@@ -314,8 +322,10 @@ public class ClientTupleSerializer {
w.packInt(schema.version());
w.packInt(tuples.size());
+ int i = 0;
for (var tuple : tuples) {
- writeTuple(null, tuple, schema, out, false, true);
+ boolean keyOnly = deleted != null && deleted.get(i++);
+ writeTuple(null, tuple, schema, out, keyOnly, true);
}
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/ClientMetricsTest.java b/modules/client/src/test/java/org/apache/ignite/client/ClientMetricsTest.java
index 0a4ae8c6ae..77f366c449 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/ClientMetricsTest.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/ClientMetricsTest.java
@@ -42,6 +42,7 @@ import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.lang.ErrorGroups.Sql;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.Table;
import org.apache.ignite.table.Tuple;
import org.junit.jupiter.api.AfterEach;
@@ -244,11 +245,11 @@ public class ClientMetricsTest extends BaseIgniteAbstractTest {
Table table = oneColumnTable();
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Tuple>(ForkJoinPool.commonPool(), 1)) {
+ try (var publisher = new SubmissionPublisher<DataStreamerItem<Tuple>>(ForkJoinPool.commonPool(), 1)) {
streamerFut = table.recordView().streamData(publisher, null);
- publisher.submit(Tuple.create().set("ID", "1"));
- publisher.submit(Tuple.create().set("ID", "2"));
+ publisher.submit(DataStreamerItem.of(Tuple.create().set("ID", "1")));
+ publisher.submit(DataStreamerItem.of(Tuple.create().set("ID", "2")));
assertTrue(IgniteTestUtils.waitForCondition(() -> metrics().streamerItemsQueued() == 2, 1000));
assertEquals(0, metrics().streamerItemsSent());
diff --git a/modules/client/src/test/java/org/apache/ignite/client/DataStreamerTest.java b/modules/client/src/test/java/org/apache/ignite/client/DataStreamerTest.java
index 1eaf563d7d..e21a9130a4 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/DataStreamerTest.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/DataStreamerTest.java
@@ -26,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -37,7 +38,9 @@ import java.util.function.Function;
import org.apache.ignite.client.IgniteClient.Builder;
import org.apache.ignite.client.fakes.FakeIgnite;
import org.apache.ignite.client.fakes.FakeIgniteTables;
+import org.apache.ignite.internal.streamer.SimplePublisher;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.RecordView;
@@ -68,7 +71,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SimplePublisher<Tuple>()) {
var options = DataStreamerOptions.builder().pageSize(batchSize).build();
streamerFut = view.streamData(publisher, options);
@@ -90,7 +93,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
RecordView<PersonPojo> view = defaultTable().recordView(PersonPojo.class);
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<PersonPojo>()) {
+ try (var publisher = new SimplePublisher<PersonPojo>()) {
streamerFut = view.streamData(publisher, null);
publisher.submit(new PersonPojo(1L, "foo"));
@@ -106,7 +109,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
KeyValueView<Tuple, Tuple> view = defaultTable().keyValueView();
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Map.Entry<Tuple, Tuple>>()) {
+ try (var publisher = new SimplePublisher<Entry<Tuple, Tuple>>()) {
streamerFut = view.streamData(publisher, null);
publisher.submit(Map.entry(tupleKey(1L), tupleVal("foo")));
@@ -122,7 +125,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
KeyValueView<Long, PersonValPojo> view = defaultTable().keyValueView(Mapper.of(Long.class), Mapper.of(PersonValPojo.class));
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Map.Entry<Long, PersonValPojo>>()) {
+ try (var publisher = new SimplePublisher<Entry<Long, PersonValPojo>>()) {
streamerFut = view.streamData(publisher, null);
publisher.submit(Map.entry(1L, new PersonValPojo("foo")));
@@ -137,7 +140,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
public void testAutoFlushByTimer() throws InterruptedException {
RecordView<Tuple> view = this.defaultTable().recordView();
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SimplePublisher<Tuple>()) {
var options = DataStreamerOptions.builder().autoFlushFrequency(100).build();
view.streamData(publisher, options);
@@ -150,7 +153,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
public void testAutoFlushDisabled() throws InterruptedException {
RecordView<Tuple> view = this.defaultTable().recordView();
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SimplePublisher<Tuple>()) {
var options = DataStreamerOptions.builder().autoFlushFrequency(-1).build();
view.streamData(publisher, options);
@@ -170,7 +173,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
RecordView<Tuple> view = defaultTableView(ignite2, client2);
var bufferSize = 2;
- try (var publisher = new SubmissionPublisher<Tuple>(ForkJoinPool.commonPool(), bufferSize)) {
+ try (var publisher = new SubmissionPublisher<DataStreamerItem<Tuple>>(ForkJoinPool.commonPool(), bufferSize)) {
var options = DataStreamerOptions.builder()
.pageSize(bufferSize)
.perNodeParallelOperations(1)
@@ -181,7 +184,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
// Stream 20 items (5 per partition) while buffer capacity is 2 to trigger back pressure.
var submitFut = CompletableFuture.runAsync(() -> {
for (long i = 0; i < 20; i++) {
- publisher.submit(tuple(i, "foo_" + i));
+ publisher.submit(DataStreamerItem.of(tuple(i, "foo_" + i)));
}
});
@@ -208,7 +211,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
RecordView<Tuple> view = defaultTableView(ignite2, client2);
CompletableFuture<Void> streamFut;
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SimplePublisher<Tuple>()) {
var options = DataStreamerOptions.builder().pageSize(2).build();
streamFut = view.streamData(publisher, options);
@@ -240,7 +243,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
RecordView<Tuple> view = defaultTableView(ignite2, client2);
CompletableFuture<Void> streamFut;
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SimplePublisher<Tuple>()) {
var options = DataStreamerOptions.builder().pageSize(2).retryLimit(3).build();
streamFut = view.streamData(publisher, options);
@@ -271,7 +274,7 @@ public class DataStreamerTest extends AbstractClientTableTest {
RecordView<Tuple> view = defaultTableView(ignite2, client2);
CompletableFuture<Void> streamFut;
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SimplePublisher<Tuple>()) {
var options = DataStreamerOptions.builder().retryLimit(0).build();
streamFut = view.streamData(publisher, options);
publisher.submit(tuple(1L, "foo"));
@@ -281,6 +284,38 @@ public class DataStreamerTest extends AbstractClientTableTest {
logger.assertLogContains("Failed to refresh schemas and partition assignment");
}
+ @Test
+ public void testAddUpdateRemove() {
+ RecordView<Tuple> view = defaultTable().recordView();
+ view.delete(null, tupleKey(1L));
+ view.upsert(null, tuple(2L, "foo"));
+ view.upsert(null, tuple(3L, "bar"));
+
+ CompletableFuture<Void> streamerFut;
+
+ try (var publisher = new SubmissionPublisher<DataStreamerItem<Tuple>>()) {
+ streamerFut = view.streamData(publisher, null);
+
+ // Add.
+ publisher.submit(DataStreamerItem.of(tuple(1L, "foo")));
+
+ // Update.
+ publisher.submit(DataStreamerItem.of(tuple(2L, "bar2")));
+
+ // Remove.
+ publisher.submit(DataStreamerItem.removed(tupleKey(3L)));
+ publisher.submit(DataStreamerItem.removed(tuple(4L, "x")));
+ }
+
+ streamerFut.orTimeout(1, TimeUnit.SECONDS).join();
+
+ assertNotNull(view.get(null, tupleKey(1L)));
+ assertNotNull(view.get(null, tupleKey(2L)));
+ assertNull(view.get(null, tupleKey(3L)));
+
+ assertEquals("bar2", view.get(null, tupleKey(2L)).stringValue("name"));
+ }
+
private static RecordView<Tuple> defaultTableView(FakeIgnite server, IgniteClient client) {
((FakeIgniteTables) server.tables()).createTable(DEFAULT_TABLE);
diff --git a/modules/client/src/test/java/org/apache/ignite/client/PartitionAwarenessTest.java b/modules/client/src/test/java/org/apache/ignite/client/PartitionAwarenessTest.java
index db07abf211..e97d621d6a 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/PartitionAwarenessTest.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/PartitionAwarenessTest.java
@@ -41,9 +41,11 @@ import org.apache.ignite.compute.IgniteCompute;
import org.apache.ignite.internal.client.ReliableChannel;
import org.apache.ignite.internal.client.tx.ClientTransaction;
import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.streamer.SimplePublisher;
import org.apache.ignite.internal.table.TableViewInternal;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.RecordView;
@@ -465,7 +467,7 @@ public class PartitionAwarenessTest extends AbstractClientTest {
Consumer<Tuple> stream = t -> {
CompletableFuture<Void> fut;
- try (SubmissionPublisher<Tuple> publisher = new SubmissionPublisher<>()) {
+ try (SimplePublisher<Tuple> publisher = new SimplePublisher<>()) {
fut = recordView.streamData(publisher, null);
publisher.submit(t);
}
@@ -486,7 +488,7 @@ public class PartitionAwarenessTest extends AbstractClientTest {
Consumer<PersonPojo> stream = t -> {
CompletableFuture<Void> fut;
- try (SubmissionPublisher<PersonPojo> publisher = new SubmissionPublisher<>()) {
+ try (SimplePublisher<PersonPojo> publisher = new SimplePublisher<>()) {
fut = pojoView.streamData(publisher, null);
publisher.submit(t);
}
@@ -507,7 +509,7 @@ public class PartitionAwarenessTest extends AbstractClientTest {
Consumer<Tuple> stream = t -> {
CompletableFuture<Void> fut;
- try (SubmissionPublisher<Entry<Tuple, Tuple>> publisher = new SubmissionPublisher<>()) {
+ try (SimplePublisher<Entry<Tuple, Tuple>> publisher = new SimplePublisher<>()) {
fut = recordView.streamData(publisher, null);
publisher.submit(Map.entry(t, Tuple.create()));
}
@@ -528,7 +530,7 @@ public class PartitionAwarenessTest extends AbstractClientTest {
Consumer<Long> stream = t -> {
CompletableFuture<Void> fut;
- try (SubmissionPublisher<Entry<Long, String>> publisher = new SubmissionPublisher<>()) {
+ try (SimplePublisher<Entry<Long, String>> publisher = new SimplePublisher<>()) {
fut = kvView.streamData(publisher, null);
publisher.submit(Map.entry(t, t.toString()));
}
@@ -553,13 +555,13 @@ public class PartitionAwarenessTest extends AbstractClientTest {
CompletableFuture<Void> fut;
RecordView<Tuple> recordView = defaultTable().recordView();
- try (SubmissionPublisher<Tuple> publisher = new SubmissionPublisher<>()) {
+ try (SubmissionPublisher<DataStreamerItem<Tuple>> publisher = new SubmissionPublisher<>()) {
fut = recordView.streamData(publisher, options);
Consumer<Long> submit = id -> {
try {
lastOpServerName = null;
- publisher.submit(Tuple.create().set("ID", id));
+ publisher.submit(DataStreamerItem.of(Tuple.create().set("ID", id)));
assertTrue(IgniteTestUtils.waitForCondition(() -> lastOpServerName != null, 1000));
} catch (InterruptedException e) {
throw new RuntimeException(e);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/streamer/StreamerBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/streamer/StreamerBuffer.java
index 4289caf63b..bff2dbfdda 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/streamer/StreamerBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/streamer/StreamerBuffer.java
@@ -18,23 +18,29 @@
package org.apache.ignite.internal.streamer;
import java.util.ArrayList;
+import java.util.BitSet;
import java.util.List;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
+import org.apache.ignite.table.DataStreamerItem;
+import org.apache.ignite.table.DataStreamerOperationType;
class StreamerBuffer<T> {
private final int capacity;
- private final Consumer<List<T>> flusher;
+ private final BiConsumer<List<T>, BitSet> flusher;
/** Primary buffer. Won't grow over capacity. */
private List<T> buf;
+ private BitSet deleted;
+
private boolean closed;
- StreamerBuffer(int capacity, Consumer<List<T>> flusher) {
+ StreamerBuffer(int capacity, BiConsumer<List<T>, BitSet> flusher) {
this.capacity = capacity;
this.flusher = flusher;
buf = new ArrayList<>(capacity);
+ deleted = new BitSet(capacity);
}
/**
@@ -42,16 +48,21 @@ class StreamerBuffer<T> {
*
* @param item Item.
*/
- synchronized void add(T item) {
+ synchronized void add(DataStreamerItem<T> item) {
if (closed) {
throw new IllegalStateException("Streamer is closed, can't add items.");
}
- buf.add(item);
+ buf.add(item.get());
+
+ if (item.operationType() == DataStreamerOperationType.REMOVE) {
+ deleted.set(buf.size() - 1);
+ }
if (buf.size() >= capacity) {
- flusher.accept(buf);
+ flusher.accept(buf, deleted);
buf = new ArrayList<>(capacity);
+ deleted = new BitSet(capacity);
}
}
@@ -63,7 +74,7 @@ class StreamerBuffer<T> {
closed = true;
if (!buf.isEmpty()) {
- flusher.accept(buf);
+ flusher.accept(buf, deleted);
}
}
@@ -72,7 +83,8 @@ class StreamerBuffer<T> {
return;
}
- flusher.accept(buf);
+ flusher.accept(buf, deleted);
buf = new ArrayList<>(capacity);
+ deleted = new BitSet(capacity);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/streamer/StreamerSubscriber.java b/modules/core/src/main/java/org/apache/ignite/internal/streamer/StreamerSubscriber.java
index f48a7c6240..d0f36b28e0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/streamer/StreamerSubscriber.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/streamer/StreamerSubscriber.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.streamer;
import static org.apache.ignite.internal.util.IgniteUtils.copyStateTo;
+import java.util.BitSet;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -34,6 +35,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.table.DataStreamerItem;
import org.jetbrains.annotations.Nullable;
/**
@@ -42,7 +44,7 @@ import org.jetbrains.annotations.Nullable;
* @param <T> Item type.
* @param <P> Partition type.
*/
-public class StreamerSubscriber<T, P> implements Subscriber<T> {
+public class StreamerSubscriber<T, P> implements Subscriber<DataStreamerItem<T>> {
private final StreamerBatchSender<T, P> batchSender;
private final StreamerPartitionAwarenessProvider<T, P> partitionAwarenessProvider;
@@ -120,14 +122,14 @@ public class StreamerSubscriber<T, P> implements Subscriber<T> {
/** {@inheritDoc} */
@Override
- public void onNext(T item) {
+ public void onNext(DataStreamerItem<T> item) {
pendingItemCount.decrementAndGet();
- P partition = partitionAwarenessProvider.partition(item);
+ P partition = partitionAwarenessProvider.partition(item.get());
StreamerBuffer<T> buf = buffers.computeIfAbsent(
partition,
- p -> new StreamerBuffer<>(options.pageSize(), items -> enlistBatch(p, items)));
+ p -> new StreamerBuffer<>(options.pageSize(), (items, deleted) -> enlistBatch(p, items, deleted)));
buf.add(item);
this.metrics.streamerItemsQueuedAdd(1);
@@ -156,7 +158,7 @@ public class StreamerSubscriber<T, P> implements Subscriber<T> {
return completionFut;
}
- private void enlistBatch(P partition, Collection<T> batch) {
+ private void enlistBatch(P partition, Collection<T> batch, BitSet deleted) {
int batchSize = batch.size();
assert batchSize > 0 : "Batch size must be positive.";
assert partition != null : "Partition must not be null.";
@@ -167,15 +169,14 @@ public class StreamerSubscriber<T, P> implements Subscriber<T> {
pendingRequests.compute(
partition,
// Chain existing futures to preserve request order.
- (part, fut) -> fut == null ? sendBatch(part, batch) : fut.thenCompose(v -> sendBatch(part, batch))
+ (part, fut) -> fut == null ? sendBatch(part, batch, deleted) : fut.thenCompose(v -> sendBatch(part, batch, deleted))
);
}
- private CompletableFuture<Void> sendBatch(P partition, Collection<T> batch) {
+ private CompletableFuture<Void> sendBatch(P partition, Collection<T> batch, BitSet deleted) {
// If a connection fails, the batch goes to default connection thanks to built-it retry mechanism.
try {
- // TODO IGNITE-21403 propagate deleted set from the API.
- return batchSender.sendAsync(partition, batch, null).whenComplete((res, err) -> {
+ return batchSender.sendAsync(partition, batch, deleted).whenComplete((res, err) -> {
if (err != null) {
// Retry is handled by the sender (RetryPolicy in ReliableChannel on the client, sendWithRetry on the server).
// If we get here, then retries are exhausted and we should fail the streamer.
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/streamer/StreamerSubscriberTest.java b/modules/core/src/test/java/org/apache/ignite/internal/streamer/StreamerSubscriberTest.java
index f06b3db7c3..11d56001c0 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/streamer/StreamerSubscriberTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/streamer/StreamerSubscriberTest.java
@@ -29,6 +29,7 @@ import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.LongFunction;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.table.DataStreamerItem;
import org.junit.jupiter.api.Test;
class StreamerSubscriberTest extends BaseIgniteAbstractTest {
@@ -89,7 +90,7 @@ class StreamerSubscriberTest extends BaseIgniteAbstractTest {
/**
* Publisher that generates a {@code limit} number of items and dispatches them in the same thread.
*/
- private static class LimitedPublisher<T> implements Publisher<T> {
+ private static class LimitedPublisher<T> implements Publisher<DataStreamerItem<T>> {
private final long limit;
private final LongFunction<T> generator;
@@ -100,7 +101,7 @@ class StreamerSubscriberTest extends BaseIgniteAbstractTest {
}
@Override
- public void subscribe(Subscriber<? super T> subscriber) {
+ public void subscribe(Subscriber<? super DataStreamerItem<T>> subscriber) {
subscriber.onSubscribe(new Subscription() {
private long produced = 0;
@@ -112,7 +113,8 @@ class StreamerSubscriberTest extends BaseIgniteAbstractTest {
} else if (produced < limit) {
for (int i = 0; i < Math.min(n, limit - produced); i++) {
produced++;
- subscriber.onNext(generator.apply(produced));
+ T item = generator.apply(produced);
+ subscriber.onNext(DataStreamerItem.of(item));
}
}
}
diff --git a/modules/api/src/main/java/org/apache/ignite/table/DataStreamerTarget.java b/modules/core/src/testFixtures/java/org/apache/ignite/internal/streamer/SimplePublisher.java
similarity index 51%
copy from modules/api/src/main/java/org/apache/ignite/table/DataStreamerTarget.java
copy to modules/core/src/testFixtures/java/org/apache/ignite/internal/streamer/SimplePublisher.java
index a6a1bd3768..85af80d478 100644
--- a/modules/api/src/main/java/org/apache/ignite/table/DataStreamerTarget.java
+++ b/modules/core/src/testFixtures/java/org/apache/ignite/internal/streamer/SimplePublisher.java
@@ -15,27 +15,32 @@
* limitations under the License.
*/
-package org.apache.ignite.table;
+package org.apache.ignite.internal.streamer;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Flow;
-import org.jetbrains.annotations.Nullable;
+import java.util.concurrent.Flow.Subscriber;
+import java.util.concurrent.SubmissionPublisher;
+import org.apache.ignite.table.DataStreamerItem;
/**
- * Represents an entity that can be used as a target for streaming data.
+ * Simple publisher that wraps items into {@link DataStreamerItem}.
*
- * @param <T> Entry type.
+ * @param <T> Type of the item.
*/
-@SuppressWarnings("InterfaceMayBeAnnotatedFunctional")
-public interface DataStreamerTarget<T> {
- /**
- * Streams data into the underlying table.
- *
- * @param publisher Producer.
- * @param options Options (can be null).
- * @return Future that will be completed when the stream is finished.
- */
- CompletableFuture<Void> streamData(
- Flow.Publisher<T> publisher,
- @Nullable DataStreamerOptions options);
+public class SimplePublisher<T> implements Flow.Publisher<DataStreamerItem<T>>, AutoCloseable {
+ private final SubmissionPublisher<DataStreamerItem<T>> publisher = new SubmissionPublisher<>();
+
+ @Override
+ public void subscribe(Subscriber<? super DataStreamerItem<T>> subscriber) {
+ publisher.subscribe(subscriber);
+ }
+
+ public void submit(T item) {
+ publisher.submit(DataStreamerItem.of((item)));
+ }
+
+ @Override
+ public void close() {
+ publisher.close();
+ }
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/streamer/ItAbstractDataStreamerTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/streamer/ItAbstractDataStreamerTest.java
index a07df6a851..51bce71074 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/streamer/ItAbstractDataStreamerTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/streamer/ItAbstractDataStreamerTest.java
@@ -27,14 +27,18 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import org.apache.ignite.Ignite;
import org.apache.ignite.internal.ClusterPerClassIntegrationTest;
import org.apache.ignite.sql.Session;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.RecordView;
@@ -46,11 +50,13 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
/**
* Common test logic for data streamer - client and server APIs.
*/
+@SuppressWarnings("DataFlowIssue")
public abstract class ItAbstractDataStreamerTest extends ClusterPerClassIntegrationTest {
public static final String TABLE_NAME = "test_table";
@@ -70,15 +76,18 @@ public abstract class ItAbstractDataStreamerTest extends ClusterPerClassIntegrat
@ValueSource(ints = {1, 2, 3})
public void testBasicStreamingRecordBinaryView(int batchSize) {
RecordView<Tuple> view = defaultTable().recordView();
+ view.upsert(null, tuple(2, "_"));
+ view.upsert(null, tuple(3, "baz"));
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SubmissionPublisher<DataStreamerItem<Tuple>>()) {
var options = DataStreamerOptions.builder().pageSize(batchSize).build();
streamerFut = view.streamData(publisher, options);
- publisher.submit(tuple(1, "foo"));
- publisher.submit(tuple(2, "bar"));
+ publisher.submit(DataStreamerItem.of(tuple(1, "foo")));
+ publisher.submit(DataStreamerItem.of(tuple(2, "bar")));
+ publisher.submit(DataStreamerItem.removed(tupleKey(3)));
}
streamerFut.orTimeout(1, TimeUnit.SECONDS).join();
@@ -93,49 +102,70 @@ public abstract class ItAbstractDataStreamerTest extends ClusterPerClassIntegrat
@Test
public void testBasicStreamingRecordPojoView() {
RecordView<PersonPojo> view = defaultTable().recordView(PersonPojo.class);
+ view.upsert(null, new PersonPojo(2, "_"));
+ view.upsert(null, new PersonPojo(3, "baz"));
+
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<PersonPojo>()) {
+ try (var publisher = new SubmissionPublisher<DataStreamerItem<PersonPojo>>()) {
streamerFut = view.streamData(publisher, null);
- publisher.submit(new PersonPojo(1, "foo"));
- publisher.submit(new PersonPojo(2, "bar"));
+ publisher.submit(DataStreamerItem.of(new PersonPojo(1, "foo")));
+ publisher.submit(DataStreamerItem.of(new PersonPojo(2, "bar")));
+ publisher.submit(DataStreamerItem.removed(new PersonPojo(3)));
}
streamerFut.orTimeout(1, TimeUnit.SECONDS).join();
+
+ assertEquals("foo", view.get(null, new PersonPojo(1)).name);
assertEquals("bar", view.get(null, new PersonPojo(2)).name);
+ assertNull(view.get(null, new PersonPojo(3)));
}
@Test
public void testBasicStreamingKvBinaryView() {
KeyValueView<Tuple, Tuple> view = defaultTable().keyValueView();
+ view.put(null, tupleKey(2), Tuple.create().set("name", "_"));
+ view.put(null, tupleKey(3), Tuple.create().set("name", "baz"));
+
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Map.Entry<Tuple, Tuple>>()) {
+ try (var publisher = new SubmissionPublisher<DataStreamerItem<Map.Entry<Tuple, Tuple>>>()) {
streamerFut = view.streamData(publisher, null);
- publisher.submit(Map.entry(tupleKey(1), Tuple.create().set("name", "foo")));
- publisher.submit(Map.entry(tupleKey(2), Tuple.create().set("name", "bar")));
+ publisher.submit(DataStreamerItem.of(Map.entry(tupleKey(1), Tuple.create().set("name", "foo"))));
+ publisher.submit(DataStreamerItem.of(Map.entry(tupleKey(2), Tuple.create().set("name", "bar"))));
+ publisher.submit(DataStreamerItem.removed(Map.entry(tupleKey(3), Tuple.create())));
}
streamerFut.orTimeout(1, TimeUnit.SECONDS).join();
+
+ assertEquals("foo", view.get(null, tupleKey(1)).stringValue("name"));
assertEquals("bar", view.get(null, tupleKey(2)).stringValue("name"));
+ assertNull(view.get(null, tupleKey(3)));
}
@Test
public void testBasicStreamingKvPojoView() {
KeyValueView<Integer, PersonValPojo> view = defaultTable().keyValueView(Mapper.of(Integer.class), Mapper.of(PersonValPojo.class));
+ view.put(null, 2, new PersonValPojo("_"));
+ view.put(null, 3, new PersonValPojo("baz"));
+
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Map.Entry<Integer, PersonValPojo>>()) {
+ try (var publisher = new SubmissionPublisher<DataStreamerItem<Map.Entry<Integer, PersonValPojo>>>()) {
streamerFut = view.streamData(publisher, null);
- publisher.submit(Map.entry(1, new PersonValPojo("foo")));
- publisher.submit(Map.entry(2, new PersonValPojo("bar")));
+ publisher.submit(DataStreamerItem.of(Map.entry(1, new PersonValPojo("foo"))));
+ publisher.submit(DataStreamerItem.of(Map.entry(2, new PersonValPojo("bar"))));
+ publisher.submit(DataStreamerItem.removed(Map.entry(3, new PersonValPojo("_"))));
}
streamerFut.orTimeout(1, TimeUnit.SECONDS).join();
+
+ assertEquals("foo", view.get(null, 1).name);
assertEquals("bar", view.get(null, 2).name);
+ assertNull(view.get(null, 3));
}
@Test
@@ -143,7 +173,7 @@ public abstract class ItAbstractDataStreamerTest extends ClusterPerClassIntegrat
RecordView<Tuple> view = this.defaultTable().recordView();
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SimplePublisher<Tuple>()) {
var options = DataStreamerOptions.builder().autoFlushFrequency(100).build();
streamerFut = view.streamData(publisher, options);
@@ -159,7 +189,7 @@ public abstract class ItAbstractDataStreamerTest extends ClusterPerClassIntegrat
RecordView<Tuple> view = this.defaultTable().recordView();
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SimplePublisher<Tuple>()) {
var options = DataStreamerOptions.builder().autoFlushFrequency(-1).build();
streamerFut = view.streamData(publisher, options);
@@ -176,7 +206,7 @@ public abstract class ItAbstractDataStreamerTest extends ClusterPerClassIntegrat
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SimplePublisher<Tuple>()) {
var options = DataStreamerOptions.builder().build();
streamerFut = view.streamData(publisher, options);
@@ -189,26 +219,74 @@ public abstract class ItAbstractDataStreamerTest extends ClusterPerClassIntegrat
assertEquals("Missed key column: ID", ex.getCause().getMessage());
}
+ @SuppressWarnings("Convert2MethodRef")
@Test
public void testManyItems() {
+ int count = 5_000;
+
RecordView<Tuple> view = defaultTable().recordView();
+ view.upsertAll(null, IntStream.range(0, count).mapToObj(i -> tuple(i, "old-" + i)).collect(Collectors.toList()));
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SubmissionPublisher<DataStreamerItem<Tuple>>()) {
var options = DataStreamerOptions.builder().pageSize(33).build();
streamerFut = view.streamData(publisher, options);
- for (int i = 0; i < 10_000; i++) {
- publisher.submit(tuple(i, "x-" + i));
+ for (int i = 0; i < count; i++) {
+ DataStreamerItem<Tuple> item = i % 2 == 0
+ ? DataStreamerItem.of(tuple(i, "new-" + i))
+ : DataStreamerItem.removed(tupleKey(i));
+
+ publisher.submit(item);
}
}
streamerFut.orTimeout(30, TimeUnit.SECONDS).join();
- assertNotNull(view.get(null, tupleKey(1)));
- assertNotNull(view.get(null, tupleKey(9999)));
- assertNull(view.get(null, tupleKey(10_000)));
+ List<Tuple> res = view.getAll(null, IntStream.range(0, count).mapToObj(i -> tupleKey(i)).collect(Collectors.toList()));
+
+ for (int i = 0; i < count; i++) {
+ Tuple tuple = res.get(i);
+
+ if (i % 2 == 0) {
+ assertEquals("new-" + i, tuple.stringValue("name"));
+ } else {
+ assertNull(tuple);
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource("1000, true")
+ // TODO IGNITE-21521 Wrong update order in DataStreamer for a new key
+ // @CsvSource({"100, false", "100, true", "1000, false", "1000, true"})
+ public void testSameItemMultipleUpdatesOrder(int pageSize, boolean existingKey) {
+ int id = pageSize + (existingKey ? 1 : 2);
+ RecordView<Tuple> view = defaultTable().recordView();
+
+ if (existingKey) {
+ view.upsert(null, tuple(id, "old"));
+ } else {
+ view.delete(null, tupleKey(id));
+ }
+
+ CompletableFuture<Void> streamerFut;
+
+ try (var publisher = new SubmissionPublisher<DataStreamerItem<Tuple>>()) {
+ DataStreamerOptions options = DataStreamerOptions.builder().pageSize(pageSize).build();
+ streamerFut = view.streamData(publisher, options);
+
+ for (int i = 0; i < 100; i++) {
+ publisher.submit(DataStreamerItem.of(tuple(id, "foo-" + i)));
+ publisher.submit(DataStreamerItem.removed(tupleKey(id)));
+ publisher.submit(DataStreamerItem.of(tuple(id, "bar-" + i)));
+ }
+ }
+
+ streamerFut.orTimeout(id, TimeUnit.SECONDS).join();
+
+ assertEquals("bar-99", view.get(null, tupleKey(id)).stringValue("name"));
}
@SuppressWarnings("resource")
@@ -222,7 +300,7 @@ public abstract class ItAbstractDataStreamerTest extends ClusterPerClassIntegrat
CompletableFuture<Void> streamerFut;
- try (var publisher = new SubmissionPublisher<Tuple>()) {
+ try (var publisher = new SimplePublisher<Tuple>()) {
var options = DataStreamerOptions.builder().pageSize(1).build();
streamerFut = view.streamData(publisher, options);
@@ -267,6 +345,7 @@ public abstract class ItAbstractDataStreamerTest extends ClusterPerClassIntegrat
.set("id", id);
}
+ @SuppressWarnings("unused")
private static class PersonPojo {
int id;
String name;
@@ -287,6 +366,7 @@ public abstract class ItAbstractDataStreamerTest extends ClusterPerClassIntegrat
}
}
+ @SuppressWarnings("unused")
private static class PersonValPojo {
String name;
Double salary;
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java
index b6cb3a5a38..50c196acc8 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java
@@ -74,6 +74,8 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
/**
* Tests for the internal table API.
@@ -474,6 +476,42 @@ public class ItInternalTableTest extends BaseIgniteAbstractTest {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = true)
+ // TODO IGNITE-21521 Wrong update order in DataStreamer for a new key
+ // @ValueSource(booleans = {true, false})
+ public void updateAllOrderTest(boolean existingKey) {
+ RecordView<Tuple> view = table.recordView();
+ InternalTable internalTable = ((TableViewInternal) table).internalTable();
+ List<BinaryRowEx> rows = new ArrayList<>();
+
+ int count = 100;
+ int lastId = count - 1;
+ long id = existingKey ? 1 : 12345;
+ BitSet deleted = new BitSet(count);
+
+ if (existingKey) {
+ view.upsert(null, Tuple.create().set("key", id).set("valInt", 1).set("valStr", "val1"));
+ }
+
+ for (int i = 0; i < count; i++) {
+ if (i % 2 == 0) {
+ rows.add(createKeyRow(id));
+ deleted.set(i);
+ } else {
+ rows.add(createKeyValueRow(id, i, "row-" + i));
+ }
+ }
+
+ int partitionId = internalTable.partitionId(rows.get(0));
+
+ internalTable.updateAll(rows, deleted, partitionId).join();
+
+ Tuple res = view.get(null, Tuple.create().set("key", id));
+ assertEquals(lastId, res.intValue("valInt"));
+ assertEquals("row-" + lastId, res.stringValue("valStr"));
+ }
+
@Test
public void updateAllWithDeleteTest() {
InternalTable internalTable = ((TableViewInternal) table).internalTable();
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/DataStreamer.java b/modules/table/src/main/java/org/apache/ignite/internal/table/DataStreamer.java
index 5af85ba60b..916768125f 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/DataStreamer.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/DataStreamer.java
@@ -25,6 +25,7 @@ import org.apache.ignite.internal.streamer.StreamerBatchSender;
import org.apache.ignite.internal.streamer.StreamerOptions;
import org.apache.ignite.internal.streamer.StreamerPartitionAwarenessProvider;
import org.apache.ignite.internal.streamer.StreamerSubscriber;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.jetbrains.annotations.Nullable;
@@ -32,7 +33,7 @@ class DataStreamer {
private static final IgniteLogger LOG = Loggers.forClass(DataStreamer.class);
static <R> CompletableFuture<Void> streamData(
- Publisher<R> publisher,
+ Publisher<DataStreamerItem<R>> publisher,
@Nullable DataStreamerOptions options,
StreamerBatchSender<R, Integer> batchSender,
StreamerPartitionAwarenessProvider<R, Integer> partitionAwarenessProvider) {
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/KeyValueBinaryViewImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/KeyValueBinaryViewImpl.java
index 3f5aee330c..ff6a58ccaa 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/KeyValueBinaryViewImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/KeyValueBinaryViewImpl.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.table;
import java.util.ArrayList;
+import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -46,6 +47,7 @@ import org.apache.ignite.lang.NullableValue;
import org.apache.ignite.sql.IgniteSql;
import org.apache.ignite.sql.ResultSetMetadata;
import org.apache.ignite.sql.SqlRow;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.Tuple;
@@ -212,7 +214,7 @@ public class KeyValueBinaryViewImpl extends AbstractTableView<Entry<Tuple, Tuple
}
return withSchemaSync(tx, (schemaVersion) -> {
- return tbl.upsertAll(marshalPairs(pairs.entrySet(), schemaVersion), (InternalTransaction) tx);
+ return tbl.upsertAll(marshalPairs(pairs.entrySet(), schemaVersion, null), (InternalTransaction) tx);
});
}
@@ -547,24 +549,30 @@ public class KeyValueBinaryViewImpl extends AbstractTableView<Entry<Tuple, Tuple
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> streamData(Publisher<Entry<Tuple, Tuple>> publisher, @Nullable DataStreamerOptions options) {
+ public CompletableFuture<Void> streamData(
+ Publisher<DataStreamerItem<Entry<Tuple, Tuple>>> publisher,
+ @Nullable DataStreamerOptions options) {
Objects.requireNonNull(publisher);
var partitioner = new KeyValueTupleStreamerPartitionAwarenessProvider(rowConverter.registry(), tbl.partitions());
StreamerBatchSender<Entry<Tuple, Tuple>, Integer> batchSender = (partitionId, items, deleted) ->
withSchemaSync(
null,
- schemaVersion -> this.tbl.updateAll(marshalPairs(items, schemaVersion), deleted, partitionId));
+ schemaVersion -> this.tbl.updateAll(marshalPairs(items, schemaVersion, deleted), deleted, partitionId));
return DataStreamer.streamData(publisher, options, batchSender, partitioner);
}
- private List<BinaryRowEx> marshalPairs(Collection<Entry<Tuple, Tuple>> pairs, int schemaVersion) {
+ private List<BinaryRowEx> marshalPairs(Collection<Entry<Tuple, Tuple>> pairs, int schemaVersion, @Nullable BitSet deleted) {
List<BinaryRowEx> rows = new ArrayList<>(pairs.size());
for (Entry<Tuple, Tuple> pair : pairs) {
- Row row = marshal(Objects.requireNonNull(pair.getKey()), Objects.requireNonNull(pair.getValue()), schemaVersion);
+ boolean isDeleted = deleted != null && deleted.get(rows.size());
+ Tuple key = Objects.requireNonNull(pair.getKey());
+ Tuple val = isDeleted ? null : Objects.requireNonNull(pair.getValue());
+
+ Row row = marshal(key, val, schemaVersion);
rows.add(row);
}
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 c98deebfa6..8cc00ffa70 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
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.table;
import java.util.ArrayList;
+import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -52,6 +53,7 @@ import org.apache.ignite.lang.UnexpectedNullValueException;
import org.apache.ignite.sql.IgniteSql;
import org.apache.ignite.sql.ResultSetMetadata;
import org.apache.ignite.sql.SqlRow;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.mapper.Mapper;
@@ -235,7 +237,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView<Entry<K, V>> imple
}
return withSchemaSync(tx, (schemaVersion) -> {
- Collection<BinaryRowEx> rows = marshalPairs(pairs.entrySet(), schemaVersion);
+ Collection<BinaryRowEx> rows = marshalPairs(pairs.entrySet(), schemaVersion, null);
return tbl.upsertAll(rows, (InternalTransaction) tx);
});
@@ -551,7 +553,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView<Entry<K, V>> imple
* @param schemaVersion Schema version to use when marshalling.
* @return Binary rows.
*/
- private List<BinaryRowEx> marshalPairs(Collection<Entry<K, V>> pairs, int schemaVersion) {
+ private List<BinaryRowEx> marshalPairs(Collection<Entry<K, V>> pairs, int schemaVersion, @Nullable BitSet deleted) {
if (pairs.isEmpty()) {
return Collections.emptyList();
}
@@ -562,7 +564,15 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView<Entry<K, V>> imple
try {
for (Map.Entry<K, V> pair : pairs) {
- rows.add(marsh.marshal(Objects.requireNonNull(pair.getKey()), pair.getValue()));
+ boolean isDeleted = deleted != null && deleted.get(rows.size());
+
+ K key = Objects.requireNonNull(pair.getKey());
+
+ Row row = isDeleted
+ ? marsh.marshal(key)
+ : marsh.marshal(key, pair.getValue());
+
+ rows.add(row);
}
} catch (MarshallerException e) {
throw new org.apache.ignite.lang.MarshallerException(e);
@@ -676,7 +686,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView<Entry<K, V>> imple
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> streamData(Publisher<Entry<K, V>> publisher, @Nullable DataStreamerOptions options) {
+ public CompletableFuture<Void> streamData(Publisher<DataStreamerItem<Entry<K, V>>> publisher, @Nullable DataStreamerOptions options) {
Objects.requireNonNull(publisher);
// Taking latest schema version for marshaller here because it's only used to calculate colocation hash, and colocation
@@ -690,7 +700,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView<Entry<K, V>> imple
StreamerBatchSender<Entry<K, V>, Integer> batchSender = (partitionId, items, deleted) ->
withSchemaSync(
null,
- schemaVersion -> this.tbl.updateAll(marshalPairs(items, schemaVersion), deleted, partitionId));
+ schemaVersion -> this.tbl.updateAll(marshalPairs(items, schemaVersion, deleted), deleted, partitionId));
return DataStreamer.streamData(publisher, options, batchSender, partitioner);
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/RecordBinaryViewImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/RecordBinaryViewImpl.java
index 714e8ba2e2..5efd34ec76 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/RecordBinaryViewImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/RecordBinaryViewImpl.java
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.lang.MarshallerException;
import org.apache.ignite.sql.IgniteSql;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.Tuple;
@@ -356,6 +357,19 @@ public class RecordBinaryViewImpl extends AbstractTableView<Tuple> implements Re
private Row marshal(Tuple tuple, int schemaVersion, boolean keyOnly) throws IgniteException {
TupleMarshaller marshaller = marshaller(schemaVersion);
+ return marshal(tuple, marshaller, keyOnly);
+ }
+
+ /**
+ * Marshal a tuple to a row.
+ *
+ * @param tuple The tuple.
+ * @param marshaller Marshaller.
+ * @param keyOnly Marshal key part only if {@code true}, otherwise marshal both, key and value parts.
+ * @return Row.
+ * @throws IgniteException If failed to marshal tuple.
+ */
+ private static Row marshal(Tuple tuple, TupleMarshaller marshaller, boolean keyOnly) {
try {
if (keyOnly) {
return marshaller.marshalKey(tuple);
@@ -434,9 +448,21 @@ public class RecordBinaryViewImpl extends AbstractTableView<Tuple> implements Re
return mapped;
}
+ private Collection<BinaryRowEx> mapToBinary(Collection<Tuple> rows, int schemaVersion, @Nullable BitSet deleted) {
+ Collection<BinaryRowEx> mapped = new ArrayList<>(rows.size());
+ TupleMarshaller marshaller = marshaller(schemaVersion);
+
+ for (Tuple row : rows) {
+ boolean key = deleted != null && deleted.get(mapped.size());
+ mapped.add(marshal(row, marshaller, key));
+ }
+
+ return mapped;
+ }
+
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> streamData(Publisher<Tuple> publisher, @Nullable DataStreamerOptions options) {
+ public CompletableFuture<Void> streamData(Publisher<DataStreamerItem<Tuple>> publisher, @Nullable DataStreamerOptions options) {
Objects.requireNonNull(publisher);
var partitioner = new TupleStreamerPartitionAwarenessProvider(rowConverter.registry(), tbl.partitions());
@@ -455,6 +481,6 @@ public class RecordBinaryViewImpl extends AbstractTableView<Tuple> implements Re
*/
public CompletableFuture<Void> updateAll(int partitionId, Collection<Tuple> rows, @Nullable BitSet deleted) {
return withSchemaSync(null,
- schemaVersion -> this.tbl.updateAll(mapToBinary(rows, schemaVersion, false), deleted, partitionId));
+ schemaVersion -> this.tbl.updateAll(mapToBinary(rows, schemaVersion, deleted), deleted, partitionId));
}
}
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 82049cc55e..c533cfb6a7 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,6 +18,7 @@
package org.apache.ignite.internal.table;
import java.util.ArrayList;
+import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -46,6 +47,7 @@ import org.apache.ignite.lang.MarshallerException;
import org.apache.ignite.sql.IgniteSql;
import org.apache.ignite.sql.ResultSetMetadata;
import org.apache.ignite.sql.SqlRow;
+import org.apache.ignite.table.DataStreamerItem;
import org.apache.ignite.table.DataStreamerOptions;
import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.mapper.Mapper;
@@ -420,6 +422,25 @@ public class RecordViewImpl<R> extends AbstractTableView<R> implements RecordVie
}
}
+ private Collection<BinaryRowEx> marshal(Collection<R> recs, int schemaVersion, @Nullable BitSet deleted) {
+ try {
+ RecordMarshaller<R> marsh = marshaller(schemaVersion);
+
+ List<BinaryRowEx> rows = new ArrayList<>(recs.size());
+
+ for (R rec : recs) {
+ boolean isDeleted = deleted != null && deleted.get(rows.size());
+ Row row = isDeleted ? marsh.marshalKey(rec) : marsh.marshal(rec);
+
+ rows.add(row);
+ }
+
+ return rows;
+ } catch (Exception e) {
+ throw new MarshallerException(e);
+ }
+ }
+
/**
* Marshals given key record to a row.
*
@@ -519,7 +540,7 @@ public class RecordViewImpl<R> extends AbstractTableView<R> implements RecordVie
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> streamData(Publisher<R> publisher, @Nullable DataStreamerOptions options) {
+ public CompletableFuture<Void> streamData(Publisher<DataStreamerItem<R>> publisher, @Nullable DataStreamerOptions options) {
Objects.requireNonNull(publisher);
// Taking latest schema version for marshaller here because it's only used to calculate colocation hash, and colocation
@@ -533,7 +554,7 @@ public class RecordViewImpl<R> extends AbstractTableView<R> implements RecordVie
StreamerBatchSender<R, Integer> batchSender = (partitionId, items, deleted) ->
withSchemaSync(
null,
- schemaVersion -> this.tbl.updateAll(marshal(items, schemaVersion), deleted, partitionId));
+ schemaVersion -> this.tbl.updateAll(marshal(items, schemaVersion, deleted), deleted, partitionId));
return DataStreamer.streamData(publisher, options, batchSender, partitioner);
}