You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tl...@apache.org on 2022/04/22 14:15:01 UTC
[ignite-3] branch main updated: IGNITE-16596 Use colocation key for affinity calculation (#728)
This is an automated email from the ASF dual-hosted git repository.
tledkov 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 6d048ef88 IGNITE-16596 Use colocation key for affinity calculation (#728)
6d048ef88 is described below
commit 6d048ef886b3481f72669538a8ea489780430f5d
Author: Taras Ledkov <tl...@gridgain.com>
AuthorDate: Fri Apr 22 17:14:56 2022 +0300
IGNITE-16596 Use colocation key for affinity calculation (#728)
---
.../ignite/client/fakes/FakeInternalTable.java | 33 +-
.../internal/table/ItPublicApiColocationTest.java | 300 ++++++++++++++++++
.../ignite/internal/schema/BinaryRowEx.java} | 35 +--
.../internal/schema/marshaller/KvMarshaller.java | 5 +-
.../schema/marshaller/RecordMarshaller.java | 5 +-
.../marshaller/asm/AsmMarshallerGenerator.java | 20 +-
.../asm/ObjectMarshallerCodeGenerator.java | 9 +-
.../marshaller/reflection/KvMarshallerImpl.java | 9 +-
.../reflection/RecordMarshallerImpl.java | 9 +-
.../org/apache/ignite/internal/schema/row/Row.java | 31 +-
.../ignite/internal/util/ColocationUtils.java | 19 --
.../internal/sql/engine/exec/rel/ModifyNode.java | 9 +-
.../sql/engine/schema/IgniteTableImpl.java | 6 +-
.../internal/sql/engine/schema/ModifyRow.java | 8 +-
.../ignite/internal/table/ItColocationTest.java | 350 +++++++++++++++++++++
.../ignite/internal/table/InternalTable.java | 33 +-
.../internal/table/KeyValueBinaryViewImpl.java | 13 +-
.../ignite/internal/table/KeyValueViewImpl.java | 53 ++--
.../internal/table/RecordBinaryViewImpl.java | 5 +-
.../ignite/internal/table/RecordViewImpl.java | 45 +--
.../apache/ignite/internal/table/TableImpl.java | 4 +-
.../distributed/storage/InternalTableImpl.java | 45 +--
.../table/ColocationHashCalculationTest.java | 5 +-
.../table/impl/DummyInternalTableImpl.java | 5 +-
24 files changed, 853 insertions(+), 203 deletions(-)
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
index 64d098234..77e7afad7 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
@@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Flow.Publisher;
import javax.naming.OperationNotSupportedException;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.storage.engine.TableStorage;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.tx.InternalTransaction;
@@ -85,13 +86,13 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<BinaryRow> get(BinaryRow keyRow, @Nullable InternalTransaction tx) {
+ public CompletableFuture<BinaryRow> get(BinaryRowEx keyRow, @Nullable InternalTransaction tx) {
return CompletableFuture.completedFuture(data.get(keyRow.keySlice()));
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRow> keyRows,
+ public CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRowEx> keyRows,
@Nullable InternalTransaction tx) {
var res = new ArrayList<BinaryRow>();
@@ -108,7 +109,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> upsert(BinaryRow row, @Nullable InternalTransaction tx) {
+ public CompletableFuture<Void> upsert(BinaryRowEx row, @Nullable InternalTransaction tx) {
data.put(row.keySlice(), row);
return CompletableFuture.completedFuture(null);
@@ -116,7 +117,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> upsertAll(Collection<BinaryRow> rows, @Nullable InternalTransaction tx) {
+ public CompletableFuture<Void> upsertAll(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx) {
for (var row : rows) {
upsert(row, tx);
}
@@ -126,7 +127,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<BinaryRow> getAndUpsert(BinaryRow row,
+ public CompletableFuture<BinaryRow> getAndUpsert(BinaryRowEx row,
@Nullable InternalTransaction tx) {
var res = get(row, tx);
@@ -137,7 +138,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> insert(BinaryRow row, @Nullable InternalTransaction tx) {
+ public CompletableFuture<Boolean> insert(BinaryRowEx row, @Nullable InternalTransaction tx) {
var old = get(row, tx).getNow(null);
if (old == null) {
@@ -151,7 +152,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRow> rows, @Nullable InternalTransaction tx) {
+ public CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx) {
var skipped = new ArrayList<BinaryRow>();
for (var row : rows) {
@@ -165,7 +166,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> replace(BinaryRow row, @Nullable InternalTransaction tx) {
+ public CompletableFuture<Boolean> replace(BinaryRowEx row, @Nullable InternalTransaction tx) {
var old = get(row, tx).getNow(null);
if (old == null) {
@@ -177,7 +178,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> replace(BinaryRow oldRow, BinaryRow newRow, @Nullable InternalTransaction tx) {
+ public CompletableFuture<Boolean> replace(BinaryRowEx oldRow, BinaryRowEx newRow, @Nullable InternalTransaction tx) {
var old = get(oldRow, tx).getNow(null);
if (old == null || !old.valueSlice().equals(oldRow.valueSlice())) {
@@ -189,7 +190,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<BinaryRow> getAndReplace(BinaryRow row,
+ public CompletableFuture<BinaryRow> getAndReplace(BinaryRowEx row,
@Nullable InternalTransaction tx) {
var old = get(row, tx);
@@ -198,7 +199,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> delete(BinaryRow keyRow, @Nullable InternalTransaction tx) {
+ public CompletableFuture<Boolean> delete(BinaryRowEx keyRow, @Nullable InternalTransaction tx) {
var old = get(keyRow, tx).getNow(null);
if (old != null) {
@@ -210,7 +211,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> deleteExact(BinaryRow oldRow, @Nullable InternalTransaction tx) {
+ public CompletableFuture<Boolean> deleteExact(BinaryRowEx oldRow, @Nullable InternalTransaction tx) {
var old = get(oldRow, tx).getNow(null);
if (old != null && old.valueSlice().equals(oldRow.valueSlice())) {
@@ -223,7 +224,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<BinaryRow> getAndDelete(BinaryRow row,
+ public CompletableFuture<BinaryRow> getAndDelete(BinaryRowEx row,
@Nullable InternalTransaction tx) {
var old = get(row, tx).getNow(null);
@@ -236,7 +237,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRow> rows, @Nullable InternalTransaction tx) {
+ public CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx) {
var skipped = new ArrayList<BinaryRow>();
for (var row : rows) {
@@ -250,7 +251,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Collection<BinaryRow>> deleteAllExact(Collection<BinaryRow> rows, @Nullable InternalTransaction tx) {
+ public CompletableFuture<Collection<BinaryRow>> deleteAllExact(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx) {
var skipped = new ArrayList<BinaryRow>();
for (var row : rows) {
@@ -282,7 +283,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public int partition(BinaryRow keyRow) {
+ public int partition(BinaryRowEx keyRow) {
return 0;
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItPublicApiColocationTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItPublicApiColocationTest.java
new file mode 100644
index 000000000..1f02dc822
--- /dev/null
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItPublicApiColocationTest.java
@@ -0,0 +1,300 @@
+/*
+ * 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.internal.table;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.params.ParameterizedTest.ARGUMENTS_PLACEHOLDER;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Flow.Subscriber;
+import java.util.concurrent.Flow.Subscription;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.NativeTypeSpec;
+import org.apache.ignite.internal.sql.engine.AbstractBasicIntegrationTest;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.table.Table;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Disabled;
+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.Arguments;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.EnumSource.Mode;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Tests for the data colocation.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+public class ItPublicApiColocationTest extends AbstractBasicIntegrationTest {
+ /** Rows count ot test. */
+ private static final int ROWS = 10;
+
+ /**
+ * Excluded native types.
+ * TODO: https://issues.apache.org/jira/browse/IGNITE-16711 - supports DECIMAL
+ */
+ private static final Set<NativeTypeSpec> EXCLUDED_TYPES = Stream.of(
+ NativeTypeSpec.UUID,
+ NativeTypeSpec.BITMASK,
+ NativeTypeSpec.DECIMAL,
+ NativeTypeSpec.NUMBER,
+ NativeTypeSpec.TIMESTAMP,
+ NativeTypeSpec.BYTES)
+ .collect(Collectors.toSet());
+
+ /**
+ * Clear tables after each test.
+ *
+ * @param testInfo Test information oject.
+ * @throws Exception If failed.
+ */
+ @AfterEach
+ @Override
+ public void tearDown(TestInfo testInfo) throws Exception {
+ for (Table t : CLUSTER_NODES.get(0).tables().tables()) {
+ sql("DROP TABLE " + t.name());
+ }
+
+ super.tearDownBase(testInfo);
+ }
+
+ /**
+ * Check colocation by one column PK and explicit colocation key for all types.
+ * TODO: https://issues.apache.org/jira/browse/IGNITE-16711 - supports DECIMAL
+ */
+ @ParameterizedTest(name = "type=" + ARGUMENTS_PLACEHOLDER)
+ @EnumSource(
+ value = NativeTypeSpec.class,
+ names = {"INT8", "UUID", "BITMASK", "DECIMAL", "NUMBER", "TIMESTAMP", "BYTES"},
+ mode = Mode.EXCLUDE
+ )
+ // @EnumSource(value = NativeTypeSpec.class, names = {"BYTES", "TIME", "DATETIME"}, mode = Mode.INCLUDE)
+ public void colocationOneColumn(NativeTypeSpec type) throws ExecutionException, InterruptedException {
+ sql(String.format("create table test0(id %s primary key, v INTEGER)", sqlTypeName(type)));
+ sql(String.format("create table test1(id0 integer, id1 %s, v INTEGER, primary key(id0, id1)) colocate by(id1)", sqlTypeName(type)));
+
+ for (int i = 0; i < ROWS; ++i) {
+ sql("insert into test0 values(?, ?)", generateValueByType(i, type), 0);
+ sql("insert into test1 values(?, ?, ?)", i, generateValueByType(i, type), 0);
+ }
+
+ int parts = ((TableImpl) CLUSTER_NODES.get(0).tables().table("public.test0")).internalTable().partitions();
+ TableImpl tbl0 = (TableImpl) CLUSTER_NODES.get(0).tables().table("public.test0");
+ TableImpl tbl1 = (TableImpl) CLUSTER_NODES.get(0).tables().table("public.test1");
+
+ for (int i = 0; i < parts; ++i) {
+ List<Tuple> r0 = getAll(tbl0, i);
+
+ Set<Object> ids0 = r0.stream().map(t -> t.value("id")).collect(Collectors.toSet());
+ List<Tuple> r1 = getAll(tbl1, i);
+
+ r1.forEach(t -> assertTrue(ids0.remove(t.value("id1"))));
+
+ assertTrue(ids0.isEmpty());
+ }
+ }
+
+ /**
+ * Check colocation by one column for all types.
+ * TODO: https://issues.apache.org/jira/browse/IGNITE-16711 - supports DECIMAL
+ */
+ @Disabled("https://issues.apache.org/jira/browse/IGNITE-16760")
+ @ParameterizedTest(name = "types=" + ARGUMENTS_PLACEHOLDER)
+ @MethodSource("twoColumnsParameters")
+ public void colocationTwoColumns(NativeTypeSpec t0, NativeTypeSpec t1) throws ExecutionException, InterruptedException {
+ sql(String.format("create table test0(id0 %s, id1 %s, v INTEGER, primary key(id0, id1))", sqlTypeName(t0), sqlTypeName(t1)));
+
+ sql(String.format(
+ "create table test1(id integer, id0 %s, id1 %s, v INTEGER, primary key(id, id0, id1)) colocate by(id0, id1)",
+ sqlTypeName(t0),
+ sqlTypeName(t1)
+ ));
+
+ for (int i = 0; i < ROWS; ++i) {
+ sql("insert into test0 values(?, ?, ?)", generateValueByType(i, t0), generateValueByType(i, t1), 0);
+ sql("insert into test1 values(?, ?, ?, ?)", i, generateValueByType(i, t0), generateValueByType(i, t1), 0);
+ }
+
+ int parts = ((TableImpl) CLUSTER_NODES.get(0).tables().table("public.test0")).internalTable().partitions();
+ TableImpl tbl0 = (TableImpl) CLUSTER_NODES.get(0).tables().table("public.test0");
+ TableImpl tbl1 = (TableImpl) CLUSTER_NODES.get(0).tables().table("public.test1");
+
+ Function<Tuple, Tuple> tupleColocationExtract = (t) -> {
+ Tuple ret = Tuple.create();
+ ret.set("id0", t.value("id0"));
+ ret.set("id1", t.value("id1"));
+ return ret;
+ };
+
+ for (int i = 0; i < parts; ++i) {
+ List<Tuple> r0 = getAll(tbl0, i);
+
+ Set<Tuple> ids0 = r0.stream().map(tupleColocationExtract).collect(Collectors.toSet());
+
+ List<Tuple> r1 = getAll(tbl1, i);
+
+ r1.forEach(t -> assertTrue(ids0.remove(tupleColocationExtract.apply(t))));
+
+ assertTrue(ids0.isEmpty());
+ }
+ }
+
+ private static Stream<Arguments> twoColumnsParameters() {
+ List<Arguments> args = new ArrayList<>();
+
+ for (NativeTypeSpec t0 : NativeTypeSpec.values()) {
+ for (NativeTypeSpec t1 : NativeTypeSpec.values()) {
+ if (!EXCLUDED_TYPES.contains(t0) && !EXCLUDED_TYPES.contains(t1)) {
+ args.add(Arguments.of(t0, t1));
+ }
+ }
+ }
+
+ return args.stream();
+ }
+
+ private static List<Tuple> getAll(TableImpl tbl, int part) throws ExecutionException, InterruptedException {
+ List<Tuple> res = new ArrayList<>();
+ CompletableFuture<Void> f = new CompletableFuture<>();
+
+ tbl.internalTable().scan(part, null).subscribe(new Subscriber<>() {
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ subscription.request(ROWS);
+ // subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(BinaryRow item) {
+ res.add(TableRow.tuple(tbl.schemaView().resolve(item)));
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ f.completeExceptionally(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ f.complete(null);
+ }
+ });
+
+ f.get();
+
+ return res;
+ }
+
+ private static Object generateValueByType(int i, NativeTypeSpec type) {
+ switch (type) {
+ case INT8:
+ return (byte) i;
+ case INT16:
+ return (short) i;
+ case INT32:
+ return i;
+ case INT64:
+ return (long) i;
+ case FLOAT:
+ return (float) i + ((float) i / 1000);
+ case DOUBLE:
+ return (double) i + ((double) i / 1000);
+ case DECIMAL:
+ return BigDecimal.valueOf((double) i + ((double) i / 1000));
+ case UUID:
+ return new UUID(i, i);
+ case STRING:
+ return "str_" + i;
+ case BYTES:
+ return new byte[]{(byte) i, (byte) (i + 1), (byte) (i + 2)};
+ case BITMASK:
+ return new byte[]{(byte) i};
+ case NUMBER:
+ return BigInteger.valueOf(i);
+ case DATE:
+ return LocalDate.of(2022, 01, 01).plusDays(i);
+ case TIME:
+ return LocalTime.of(0, 00, 00).plusSeconds(i);
+ case DATETIME:
+ return LocalDateTime.of(
+ (LocalDate) generateValueByType(i, NativeTypeSpec.DATE),
+ (LocalTime) generateValueByType(i, NativeTypeSpec.TIME)
+ );
+ case TIMESTAMP:
+ return Instant.from((LocalDateTime) generateValueByType(i, NativeTypeSpec.DATETIME));
+ default:
+ throw new IllegalStateException("Unexpected type: " + type);
+ }
+ }
+
+ private static String sqlTypeName(NativeTypeSpec type) {
+ switch (type) {
+ case INT8:
+ return "tinyint";
+ case INT16:
+ return "smallint";
+ case INT32:
+ return "integer";
+ case INT64:
+ return "bigint";
+ case FLOAT:
+ return "real";
+ case DOUBLE:
+ return "double";
+ case DECIMAL:
+ return "decimal";
+ case UUID:
+ return "uuid";
+ case STRING:
+ return "varchar";
+ case BYTES:
+ return "varbinary";
+ case BITMASK:
+ return "bitmap";
+ case NUMBER:
+ return "number";
+ case DATE:
+ return "date";
+ case TIME:
+ return "time";
+ case DATETIME:
+ return "timestamp";
+ case TIMESTAMP:
+ return "timestamp_tz";
+ default:
+ throw new IllegalStateException("Unexpected type: " + type);
+ }
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/ModifyRow.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/BinaryRowEx.java
similarity index 57%
copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/ModifyRow.java
copy to modules/schema/src/main/java/org/apache/ignite/internal/schema/BinaryRowEx.java
index e2734a557..031e0b677 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/ModifyRow.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/BinaryRowEx.java
@@ -15,31 +15,14 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.sql.engine.schema;
+package org.apache.ignite.internal.schema;
-import org.apache.ignite.internal.schema.BinaryRow;
-
-/** POJO to store row and modify operation. */
-public class ModifyRow {
- /** Operation state representation. */
- public enum Operation {
- INSERT_ROW, UPDATE_ROW, DELETE_ROW
- }
-
- private final BinaryRow row;
-
- private final Operation op;
-
- public ModifyRow(BinaryRow row, Operation op) {
- this.row = row;
- this.op = op;
- }
-
- public BinaryRow getRow() {
- return row;
- }
-
- public Operation getOp() {
- return op;
- }
+/**
+ * Extended binary row interface.
+ */
+public interface BinaryRowEx extends BinaryRow {
+ /**
+ * Return hash of the colocation columns' values. Used to calculate destination partition.
+ */
+ int colocationHash();
}
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/KvMarshaller.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/KvMarshaller.java
index 1e6ded936..7f09db768 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/KvMarshaller.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/KvMarshaller.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.schema.marshaller;
-import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.row.Row;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -42,7 +41,7 @@ public interface KvMarshaller<K, V> {
* @return Table row with columns from given key-value pair.
* @throws MarshallerException If failed to marshal key and/or value.
*/
- BinaryRow marshal(@NotNull K key) throws MarshallerException;
+ Row marshal(@NotNull K key) throws MarshallerException;
/**
* Marshal given key and value objects to a table row.
@@ -52,7 +51,7 @@ public interface KvMarshaller<K, V> {
* @return Table row with columns from given key-value pair.
* @throws MarshallerException If failed to marshal key and/or value.
*/
- BinaryRow marshal(@NotNull K key, V val) throws MarshallerException;
+ Row marshal(@NotNull K key, V val) throws MarshallerException;
/**
* Unmarshal given row to a key object.
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/RecordMarshaller.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/RecordMarshaller.java
index 9cdf57f10..fc7ead9d6 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/RecordMarshaller.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/RecordMarshaller.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.schema.marshaller;
-import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.row.Row;
import org.jetbrains.annotations.NotNull;
@@ -39,7 +38,7 @@ public interface RecordMarshaller<R> {
* @return Table row with columns set from given object.
* @throws MarshallerException If failed to marshal record.
*/
- BinaryRow marshal(@NotNull R rec) throws MarshallerException;
+ Row marshal(@NotNull R rec) throws MarshallerException;
/**
* Marshals key part of given record to a row.
@@ -48,7 +47,7 @@ public interface RecordMarshaller<R> {
* @return Table row with key columns set from given object.
* @throws MarshallerException If failed to marshal record.
*/
- BinaryRow marshalKey(@NotNull R keyRec) throws MarshallerException;
+ Row marshalKey(@NotNull R keyRec) throws MarshallerException;
/**
* Unmarshal given row to a record object.
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/asm/AsmMarshallerGenerator.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/asm/AsmMarshallerGenerator.java
index 23ea7abe9..0ce4ee754 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/asm/AsmMarshallerGenerator.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/asm/AsmMarshallerGenerator.java
@@ -329,7 +329,7 @@ public class AsmMarshallerGenerator implements MarshallerFactory {
final MethodDefinition methodDef = classDef.declareMethod(
EnumSet.of(Access.PUBLIC),
"marshal",
- ParameterizedType.type(BinaryRow.class),
+ ParameterizedType.type(Row.class),
Parameter.arg("key", Object.class),
Parameter.arg("val", Object.class)
).addException(MarshallerException.class);
@@ -352,19 +352,23 @@ public class AsmMarshallerGenerator implements MarshallerFactory {
final BytecodeBlock block = new BytecodeBlock();
block.append(
- keyMarsh.marshallObject(
- classDef.getType(),
- asm,
- methodDef.getScope().getVariable("key"))
- )
+ keyMarsh.marshallObject(
+ classDef.getType(),
+ asm,
+ methodDef.getScope().getVariable("key"))
+ )
.append(
valMarsh.marshallObject(
classDef.getType(),
asm,
methodDef.getScope().getVariable("val"))
)
- .append(BytecodeExpressions.newInstance(ByteBufferRow.class,
- asm.invoke("toBytes", byte[].class)))
+ .append(
+ BytecodeExpressions.newInstance(Row.class,
+ methodDef.getThis().getField("schema", SchemaDescriptor.class),
+ BytecodeExpressions.newInstance(ByteBufferRow.class,
+ asm.invoke("toBytes", byte[].class)).cast(BinaryRow.class))
+ )
.retObject();
final Variable ex = methodDef.getScope().createTempVariable(Throwable.class);
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/asm/ObjectMarshallerCodeGenerator.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/asm/ObjectMarshallerCodeGenerator.java
index 1e0093758..d8eba0fd7 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/asm/ObjectMarshallerCodeGenerator.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/asm/ObjectMarshallerCodeGenerator.java
@@ -98,9 +98,12 @@ class ObjectMarshallerCodeGenerator implements MarshallerCodeGenerator {
for (int i = 0; i < columns.length(); i++) {
final ColumnAccessCodeGenerator columnAccessor = columnAccessors[i];
- final BytecodeExpression fld = BytecodeExpressions.getStatic(marshallerClass, "FIELD_HANDLER_" + columnAccessor.columnIdx(),
- ParameterizedType.type(VarHandle.class))
- .invoke("get", columnAccessor.mappedType(), obj);
+ final BytecodeExpression fld = BytecodeExpressions.getStatic(
+ marshallerClass,
+ "FIELD_HANDLER_" + columnAccessor.columnIdx(),
+ ParameterizedType.type(VarHandle.class)
+ )
+ .invoke("get", columnAccessor.mappedType(), obj);
final BytecodeExpression marshallNonNulExpr = asm.invoke(
columnAccessor.writeMethodName(),
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/reflection/KvMarshallerImpl.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/reflection/KvMarshallerImpl.java
index 89ef99d17..db03c68a2 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/reflection/KvMarshallerImpl.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/reflection/KvMarshallerImpl.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.schema.marshaller.reflection;
import static org.apache.ignite.internal.schema.marshaller.MarshallerUtil.getValueSize;
-import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.ByteBufferRow;
import org.apache.ignite.internal.schema.Columns;
import org.apache.ignite.internal.schema.NativeType;
@@ -79,26 +78,26 @@ public class KvMarshallerImpl<K, V> implements KvMarshaller<K, V> {
/** {@inheritDoc} */
@Override
- public BinaryRow marshal(@NotNull K key) throws MarshallerException {
+ public Row marshal(@NotNull K key) throws MarshallerException {
assert keyClass.isInstance(key);
final RowAssembler asm = createAssembler(key, null);
keyMarsh.writeObject(key, asm);
- return new ByteBufferRow(asm.toBytes());
+ return new Row(schema, new ByteBufferRow(asm.toBytes()));
}
/** {@inheritDoc} */
@Override
- public BinaryRow marshal(@NotNull K key, V val) throws MarshallerException {
+ public Row marshal(@NotNull K key, V val) throws MarshallerException {
assert keyClass.isInstance(key);
assert val == null || valClass.isInstance(val);
final RowAssembler asm = createAssembler(key, val);
keyMarsh.writeObject(key, asm);
valMarsh.writeObject(val, asm);
- return new ByteBufferRow(asm.toBytes());
+ return new Row(schema, new ByteBufferRow(asm.toBytes()));
}
/** {@inheritDoc} */
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/reflection/RecordMarshallerImpl.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/reflection/RecordMarshallerImpl.java
index 2d38f91b5..fbbad91cb 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/reflection/RecordMarshallerImpl.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/marshaller/reflection/RecordMarshallerImpl.java
@@ -20,7 +20,6 @@ package org.apache.ignite.internal.schema.marshaller.reflection;
import static org.apache.ignite.internal.schema.marshaller.MarshallerUtil.getValueSize;
import java.util.Objects;
-import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.ByteBufferRow;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.Columns;
@@ -81,26 +80,26 @@ public class RecordMarshallerImpl<R> implements RecordMarshaller<R> {
/** {@inheritDoc} */
@Override
- public BinaryRow marshal(@NotNull R rec) throws MarshallerException {
+ public Row marshal(@NotNull R rec) throws MarshallerException {
assert recClass.isInstance(rec);
final RowAssembler asm = createAssembler(Objects.requireNonNull(rec), rec);
recMarsh.writeObject(rec, asm);
- return new ByteBufferRow(asm.toBytes());
+ return new Row(schema, new ByteBufferRow(asm.toBytes()));
}
/** {@inheritDoc} */
@Override
- public BinaryRow marshalKey(@NotNull R rec) throws MarshallerException {
+ public Row marshalKey(@NotNull R rec) throws MarshallerException {
assert recClass.isInstance(rec);
final RowAssembler asm = createAssembler(Objects.requireNonNull(rec), null);
keyMarsh.writeObject(rec, asm);
- return new ByteBufferRow(asm.toBytes());
+ return new Row(schema, new ByteBufferRow(asm.toBytes()));
}
/** {@inheritDoc} */
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/row/Row.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/row/Row.java
index 5ec7a7af9..ad6b54838 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/row/Row.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/row/Row.java
@@ -30,6 +30,7 @@ import java.time.LocalTime;
import java.util.BitSet;
import java.util.UUID;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.Columns;
import org.apache.ignite.internal.schema.DecimalNativeType;
@@ -38,6 +39,8 @@ import org.apache.ignite.internal.schema.NativeTypeSpec;
import org.apache.ignite.internal.schema.SchemaAware;
import org.apache.ignite.internal.schema.SchemaDescriptor;
import org.apache.ignite.internal.schema.TemporalNativeType;
+import org.apache.ignite.internal.util.ColocationUtils;
+import org.apache.ignite.internal.util.HashCalculator;
import org.jetbrains.annotations.NotNull;
/**
@@ -52,7 +55,7 @@ import org.jetbrains.annotations.NotNull;
*
* @see TemporalTypesHelper
*/
-public class Row implements BinaryRow, SchemaAware {
+public class Row implements BinaryRowEx, SchemaAware {
/**
* Null map offset.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
@@ -71,6 +74,9 @@ public class Row implements BinaryRow, SchemaAware {
/** Cached value slice byte buffer. */
private final ByteBuffer valueSlice;
+ /** Cached colocation hash value. */
+ private int colocationHash;
+
/**
* Constructor.
*
@@ -817,7 +823,7 @@ public class Row implements BinaryRow, SchemaAware {
}
/**
- * Utility method to extract the column offset from the {@link #findColumn(int, NativeTypeSpec)} result. The offset is calculated from
+ * Utility method to extract the column offset from the {@link #findColumn} result. The offset is calculated from
* the beginning of the row.
*
* @param offLen {@code findColumn} invocation result.
@@ -828,7 +834,7 @@ public class Row implements BinaryRow, SchemaAware {
}
/**
- * Utility method to extract the column length from the {@link #findColumn(int, NativeTypeSpec)} result for varlen columns.
+ * Utility method to extract the column length from the {@link #findColumn} result for varlen columns.
*
* @param offLen {@code findColumn} invocation result.
* @return Length of the column or {@code 0} if the column is fixed-length.
@@ -894,8 +900,27 @@ public class Row implements BinaryRow, SchemaAware {
}
}
+ /** {@inheritDoc} */
@Override
public byte[] bytes() {
return row.bytes();
}
+
+ /** {@inheritDoc} */
+ @Override
+ public int colocationHash() {
+ int h0 = colocationHash;
+
+ if (h0 == 0) {
+ HashCalculator hashCalc = new HashCalculator();
+
+ for (Column c : schema().colocationColumns()) {
+ ColocationUtils.append(hashCalc, value(c.schemaIndex()), c.type().spec());
+ }
+
+ colocationHash = h0 = hashCalc.hash();
+ }
+
+ return h0;
+ }
}
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/util/ColocationUtils.java b/modules/schema/src/main/java/org/apache/ignite/internal/util/ColocationUtils.java
index 31f4aae69..f67190e09 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/util/ColocationUtils.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/util/ColocationUtils.java
@@ -25,9 +25,7 @@ import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.BitSet;
import java.util.UUID;
-import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.NativeTypeSpec;
-import org.apache.ignite.internal.schema.row.Row;
/**
* Colocation hash utilities.
@@ -40,23 +38,6 @@ public class ColocationUtils {
// No-op.
}
- /**
- * Calculate colocation hash of the specified row.
- *
- * @param r Row to calculate row's colocation hash.
- * @return Row's colocation hash.
- */
- public static int colocationHash(Row r) {
- HashCalculator hashCalc = new HashCalculator();
-
- for (Column c : r.schema().colocationColumns()) {
- append(hashCalc, r.value(c.schemaIndex()), c.type().spec());
- }
-
- return hashCalc.hash();
- }
-
-
/**
* Append value to the hash calculation.
*
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java
index b992ccd28..8d500c74e 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java
@@ -28,6 +28,7 @@ import java.util.stream.Collectors;
import org.apache.calcite.rel.core.TableModify;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
import org.apache.ignite.internal.sql.engine.exec.RowHandler;
import org.apache.ignite.internal.sql.engine.schema.InternalIgniteTable;
@@ -191,8 +192,8 @@ public class ModifyNode<RowT> extends AbstractNode<RowT> implements SingleNode<R
}
/** Returns mapping of modifications per modification action. */
- private Map<ModifyRow.Operation, Collection<BinaryRow>> getOperationsPerAction(List<ModifyRow> rows) {
- Map<ModifyRow.Operation, Collection<BinaryRow>> store = new EnumMap<>(ModifyRow.Operation.class);
+ private Map<ModifyRow.Operation, Collection<BinaryRowEx>> getOperationsPerAction(List<ModifyRow> rows) {
+ Map<ModifyRow.Operation, Collection<BinaryRowEx>> store = new EnumMap<>(ModifyRow.Operation.class);
for (ModifyRow tuple : rows) {
store.computeIfAbsent(tuple.getOp(), k -> new ArrayList<>()).add(tuple.getRow());
@@ -209,10 +210,10 @@ public class ModifyNode<RowT> extends AbstractNode<RowT> implements SingleNode<R
List<ModifyRow> rows = this.rows;
this.rows = new ArrayList<>(MODIFY_BATCH_SIZE);
- Map<ModifyRow.Operation, Collection<BinaryRow>> operations = getOperationsPerAction(rows);
+ Map<ModifyRow.Operation, Collection<BinaryRowEx>> operations = getOperationsPerAction(rows);
// TODO: IGNITE-15087 Implement support for transactional SQL
- for (Map.Entry<ModifyRow.Operation, Collection<BinaryRow>> op : operations.entrySet()) {
+ for (Map.Entry<ModifyRow.Operation, Collection<BinaryRowEx>> op : operations.entrySet()) {
switch (op.getKey()) {
case INSERT_ROW:
Collection<BinaryRow> conflictKeys = tableView.insertAll(op.getValue(), null).join();
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
index fb1658b5a..428c6deeb 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
@@ -312,7 +312,7 @@ public class IgniteTableImpl extends AbstractTable implements InternalIgniteTabl
RowAssembler.writeValue(rowAssembler, colDesc.physicalType(), hnd.get(colDesc.logicalIndex(), row));
}
- return new ModifyRow(rowAssembler.build(), Operation.INSERT_ROW);
+ return new ModifyRow(new Row(schemaDescriptor, rowAssembler.build()), Operation.INSERT_ROW);
}
private <RowT> ModifyRow mergeTuple(RowT row, List<String> updateColList, ExecutionContext<RowT> ectx) {
@@ -375,7 +375,7 @@ public class IgniteTableImpl extends AbstractTable implements InternalIgniteTabl
RowAssembler.writeValue(rowAssembler, colDesc.physicalType(), val);
}
- return new ModifyRow(rowAssembler.build(), Operation.UPDATE_ROW);
+ return new ModifyRow(new Row(schemaDescriptor, rowAssembler.build()), Operation.UPDATE_ROW);
}
private <RowT> int countNotNullColumns(int start, int end, Object2IntMap<String> columnToIndex, int offset,
@@ -428,7 +428,7 @@ public class IgniteTableImpl extends AbstractTable implements InternalIgniteTabl
RowAssembler.writeValue(rowAssembler, colDesc.physicalType(), hnd.get(colDesc.logicalIndex(), row));
}
- return new ModifyRow(rowAssembler.build(), Operation.DELETE_ROW);
+ return new ModifyRow(new Row(schemaDescriptor, rowAssembler.build()), Operation.DELETE_ROW);
}
private ColocationGroup partitionedGroup() {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/ModifyRow.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/ModifyRow.java
index e2734a557..5c639a000 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/ModifyRow.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/ModifyRow.java
@@ -17,7 +17,7 @@
package org.apache.ignite.internal.sql.engine.schema;
-import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
/** POJO to store row and modify operation. */
public class ModifyRow {
@@ -26,16 +26,16 @@ public class ModifyRow {
INSERT_ROW, UPDATE_ROW, DELETE_ROW
}
- private final BinaryRow row;
+ private final BinaryRowEx row;
private final Operation op;
- public ModifyRow(BinaryRow row, Operation op) {
+ public ModifyRow(BinaryRowEx row, Operation op) {
this.row = row;
this.op = op;
}
- public BinaryRow getRow() {
+ public BinaryRowEx getRow() {
return row;
}
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
new file mode 100644
index 000000000..d3df36a1c
--- /dev/null
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
@@ -0,0 +1,350 @@
+/*
+ * 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.internal.table;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.params.ParameterizedTest.ARGUMENTS_PLACEHOLDER;
+import static org.mockito.Answers.RETURNS_DEEP_STUBS;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.when;
+
+import it.unimi.dsi.fastutil.ints.Int2IntMap;
+import it.unimi.dsi.fastutil.ints.Int2IntOpenHashMap;
+import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
+import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.ignite.internal.schema.BinaryRowEx;
+import org.apache.ignite.internal.schema.Column;
+import org.apache.ignite.internal.schema.NativeType;
+import org.apache.ignite.internal.schema.NativeTypeSpec;
+import org.apache.ignite.internal.schema.NativeTypes;
+import org.apache.ignite.internal.schema.SchemaDescriptor;
+import org.apache.ignite.internal.schema.SchemaRegistry;
+import org.apache.ignite.internal.schema.marshaller.TupleMarshallerException;
+import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
+import org.apache.ignite.internal.schema.row.Row;
+import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.table.distributed.command.InsertCommand;
+import org.apache.ignite.internal.table.distributed.command.MultiKeyCommand;
+import org.apache.ignite.internal.table.distributed.command.response.MultiRowsResponse;
+import org.apache.ignite.internal.table.distributed.storage.InternalTableImpl;
+import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
+import org.apache.ignite.internal.table.impl.DummySchemaManagerImpl;
+import org.apache.ignite.internal.tx.Timestamp;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.impl.HeapLockManager;
+import org.apache.ignite.internal.tx.impl.TxManagerImpl;
+import org.apache.ignite.internal.util.CollectionUtils;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.raft.client.Command;
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.service.RaftGroupService;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Mockito;
+
+/**
+ * Tests for data colocation.
+ */
+public class ItColocationTest {
+ /** Partitions count. */
+ private static final int PARTS = 32;
+
+ /** Keys count to check. */
+ private static final int KEYS = 100;
+
+ /** Dummy internal table for tests. */
+ private static final InternalTable INT_TABLE;
+
+ /** Map of the Raft commands are set by table operation. */
+ private static final Int2ObjectMap<Set<Command>> CMDS_MAP = new Int2ObjectOpenHashMap<>();
+
+ private SchemaDescriptor schema;
+
+ private SchemaRegistry schemaRegistry;
+
+ private TableImpl tbl;
+
+ private TupleMarshallerImpl marshaller;
+
+ static {
+ ClusterService clusterService = Mockito.mock(ClusterService.class, RETURNS_DEEP_STUBS);
+ when(clusterService.topologyService().localMember().address()).thenReturn(DummyInternalTableImpl.ADDR);
+
+ TxManager txManager = new TxManagerImpl(clusterService, new HeapLockManager()) {
+ @Override
+ public CompletableFuture<Void> finishRemote(NetworkAddress addr, Timestamp ts, boolean commit, Set<String> groups) {
+ return CompletableFuture.completedFuture(null);
+ }
+ };
+ txManager.start();
+
+ MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager);
+ when(clusterService.messagingService()).thenReturn(messagingService);
+
+ Int2ObjectMap<RaftGroupService> partRafts = new Int2ObjectOpenHashMap<>();
+
+ for (int i = 0; i < PARTS; ++i) {
+ RaftGroupService r = Mockito.mock(RaftGroupService.class);
+ when(r.leader()).thenReturn(Mockito.mock(Peer.class));
+
+ final int part = i;
+ doAnswer(invocation -> {
+ Command cmd = (Command) invocation.getArguments()[0];
+
+ CMDS_MAP.merge(part, new HashSet<>(Set.of(cmd)), (newSet, set) -> {
+ set.addAll(newSet);
+
+ return set;
+ });
+
+ if (cmd instanceof MultiKeyCommand) {
+ return CompletableFuture.completedFuture(new MultiRowsResponse(List.of()));
+ } else {
+ return CompletableFuture.completedFuture(true);
+ }
+ }).when(r).run(any());
+
+ partRafts.put(i, r);
+ }
+
+ INT_TABLE = new InternalTableImpl(
+ "PUBLIC.TEST",
+ UUID.randomUUID(),
+ partRafts,
+ PARTS,
+ null,
+ null,
+ txManager,
+ Mockito.mock(TableStorage.class)
+ );
+ }
+
+ @BeforeEach
+ public void beforeTest() {
+ CMDS_MAP.clear();
+ }
+
+ private static NativeType nativeType(NativeTypeSpec type) {
+ switch (type) {
+ case INT8:
+ return NativeTypes.INT8;
+ case INT16:
+ return NativeTypes.INT16;
+ case INT32:
+ return NativeTypes.INT32;
+ case INT64:
+ return NativeTypes.INT64;
+ case FLOAT:
+ return NativeTypes.FLOAT;
+ case DOUBLE:
+ return NativeTypes.DOUBLE;
+ case DECIMAL:
+ return NativeTypes.decimalOf(10, 3);
+ case UUID:
+ return NativeTypes.UUID;
+ case STRING:
+ return NativeTypes.STRING;
+ case BYTES:
+ return NativeTypes.BYTES;
+ case BITMASK:
+ return NativeTypes.bitmaskOf(16);
+ case NUMBER:
+ return NativeTypes.numberOf(10);
+ case DATE:
+ return NativeTypes.DATE;
+ case TIME:
+ return NativeTypes.time();
+ case DATETIME:
+ return NativeTypes.datetime();
+ case TIMESTAMP:
+ return NativeTypes.timestamp();
+ default:
+ throw new IllegalStateException("Unexpected type: " + type);
+ }
+ }
+
+ private static Object generateValueByType(int i, NativeTypeSpec type) {
+ switch (type) {
+ case INT8:
+ return (byte) i;
+ case INT16:
+ return (short) i;
+ case INT32:
+ return i;
+ case INT64:
+ return (long) i;
+ case FLOAT:
+ return (float) i + ((float) i / 1000);
+ case DOUBLE:
+ return (double) i + ((double) i / 1000);
+ case DECIMAL:
+ return BigDecimal.valueOf((double) i + ((double) i / 1000));
+ case UUID:
+ return new UUID(i, i);
+ case STRING:
+ return "str_" + i;
+ case BYTES:
+ return new byte[]{(byte) i, (byte) (i + 1), (byte) (i + 2)};
+ case BITMASK:
+ return BitSet.valueOf(new byte[]{(byte) i, (byte) (i + 1)});
+ case NUMBER:
+ return BigInteger.valueOf(i);
+ case DATE:
+ return LocalDate.of(2022, 01, 01).plusDays(i);
+ case TIME:
+ return LocalTime.of(0, 00, 00).plusSeconds(i);
+ case DATETIME:
+ return LocalDateTime.of(
+ (LocalDate) generateValueByType(i, NativeTypeSpec.DATE),
+ (LocalTime) generateValueByType(i, NativeTypeSpec.TIME)
+ );
+ case TIMESTAMP:
+ return ((LocalDateTime) generateValueByType(i, NativeTypeSpec.DATETIME))
+ .atZone(TimeZone.getDefault().toZoneId())
+ .toInstant();
+ default:
+ throw new IllegalStateException("Unexpected type: " + type);
+ }
+ }
+
+ private static Stream<Arguments> twoColumnsParameters() {
+ List<Arguments> args = new ArrayList<>();
+
+ for (NativeTypeSpec t0 : NativeTypeSpec.values()) {
+ for (NativeTypeSpec t1 : NativeTypeSpec.values()) {
+ args.add(Arguments.of(t0, t1));
+ }
+ }
+
+ return args.stream();
+ }
+
+ /**
+ * Check colocation by two columns for all types.
+ */
+ @ParameterizedTest(name = "types=" + ARGUMENTS_PLACEHOLDER)
+ @MethodSource("twoColumnsParameters")
+ public void colocationTwoColumnsInsert(NativeTypeSpec t0, NativeTypeSpec t1)
+ throws TupleMarshallerException {
+ init(t0, t1);
+
+ for (int i = 0; i < KEYS; ++i) {
+ CMDS_MAP.clear();
+
+ Tuple t = createTuple(i, t0, t1);
+
+ tbl.recordView().insert(null, t);
+
+ BinaryRowEx r = marshaller.marshal(t);
+
+ int part = INT_TABLE.partition(r);
+
+ assertThat(CollectionUtils.first(CMDS_MAP.get(part)), is(instanceOf(InsertCommand.class)));
+ }
+ }
+
+ /**
+ * Check colocation by two columns for all types.
+ */
+ @ParameterizedTest(name = "types=" + ARGUMENTS_PLACEHOLDER)
+ @MethodSource("twoColumnsParameters")
+ public void colocationTwoColumnsInsertAll(NativeTypeSpec t0, NativeTypeSpec t1)
+ throws TupleMarshallerException {
+ init(t0, t1);
+
+ tbl.recordView().insertAll(null, IntStream.range(0, KEYS).mapToObj(i -> createTuple(i, t0, t1)).collect(Collectors.toSet()));
+
+ Int2IntMap partsMap = new Int2IntOpenHashMap();
+
+ for (int i = 0; i < KEYS; ++i) {
+ Tuple t = createTuple(i, t0, t1);
+
+ BinaryRowEx r = marshaller.marshal(t);
+
+ int part = INT_TABLE.partition(r);
+
+ partsMap.merge(part, 1, (cnt, ignore) -> ++cnt);
+ }
+
+ assertEquals(partsMap.size(), CMDS_MAP.size());
+
+ CMDS_MAP.forEach((p, set) -> {
+ MultiKeyCommand cmd = (MultiKeyCommand) CollectionUtils.first(set);
+ assertEquals(partsMap.get(p), cmd.getRows().size(), () -> "part=" + p + ", set=" + set);
+
+ cmd.getRows().forEach(binRow -> {
+ Row r = new Row(schema, binRow);
+
+ assertEquals(INT_TABLE.partition(r), p);
+ });
+ });
+ }
+
+ private void init(NativeTypeSpec t0, NativeTypeSpec t1) {
+ schema = new SchemaDescriptor(1,
+ new Column[]{
+ new Column("ID", NativeTypes.INT64, false),
+ new Column("ID0", nativeType(t0), false),
+ new Column("ID1", nativeType(t1), false)
+ },
+ new String[]{"ID1", "ID0"},
+ new Column[]{
+ new Column("VAL", NativeTypes.INT64, true)
+ }
+ );
+
+ schemaRegistry = new DummySchemaManagerImpl(schema);
+
+ tbl = new TableImpl(INT_TABLE, schemaRegistry);
+
+ marshaller = new TupleMarshallerImpl(schemaRegistry);
+ }
+
+ private Tuple createTuple(int k, NativeTypeSpec t0, NativeTypeSpec t1) {
+ return Tuple.create()
+ .set("ID", 1L)
+ .set("ID0", generateValueByType(k, t0))
+ .set("ID1", generateValueByType(k, t1))
+ .set("VAL", 0L);
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java b/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
index 7ecc1bc24..8f11b8c03 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
@@ -23,6 +23,7 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Flow.Publisher;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.storage.engine.TableStorage;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.LockException;
@@ -64,7 +65,7 @@ public interface InternalTable extends AutoCloseable {
* @return Future representing pending completion of the operation.
* @throws LockException If a lock can't be acquired by some reason.
*/
- CompletableFuture<BinaryRow> get(BinaryRow keyRow, @Nullable InternalTransaction tx);
+ CompletableFuture<BinaryRow> get(BinaryRowEx keyRow, @Nullable InternalTransaction tx);
/**
* Asynchronously get rows from the table.
@@ -73,7 +74,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRow> keyRows, @Nullable InternalTransaction tx);
+ CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRowEx> keyRows, @Nullable InternalTransaction tx);
/**
* Asynchronously inserts a row into the table if does not exist or replaces the existed one.
@@ -82,7 +83,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Void> upsert(BinaryRow row, @Nullable InternalTransaction tx);
+ CompletableFuture<Void> upsert(BinaryRowEx row, @Nullable InternalTransaction tx);
/**
* Asynchronously inserts a row into the table if does not exist or replaces the existed one.
@@ -91,7 +92,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Void> upsertAll(Collection<BinaryRow> rows, @Nullable InternalTransaction tx);
+ CompletableFuture<Void> upsertAll(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx);
/**
* Asynchronously inserts a row into the table or replaces if exists and return replaced previous row.
@@ -100,7 +101,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<BinaryRow> getAndUpsert(BinaryRow row, @Nullable InternalTransaction tx);
+ CompletableFuture<BinaryRow> getAndUpsert(BinaryRowEx row, @Nullable InternalTransaction tx);
/**
* Asynchronously inserts a row into the table if not exists.
@@ -109,7 +110,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Boolean> insert(BinaryRow row, @Nullable InternalTransaction tx);
+ CompletableFuture<Boolean> insert(BinaryRowEx row, @Nullable InternalTransaction tx);
/**
* Asynchronously insert rows into the table which do not exist, skipping existed ones.
@@ -118,7 +119,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRow> rows, @Nullable InternalTransaction tx);
+ CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx);
/**
* Asynchronously replaces an existed row associated with the same key columns values as the given one has.
@@ -127,7 +128,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Boolean> replace(BinaryRow row, @Nullable InternalTransaction tx);
+ CompletableFuture<Boolean> replace(BinaryRowEx row, @Nullable InternalTransaction tx);
/**
* Asynchronously replaces an expected row in the table with the given new one.
@@ -137,7 +138,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Boolean> replace(BinaryRow oldRow, BinaryRow newRow, @Nullable InternalTransaction tx);
+ CompletableFuture<Boolean> replace(BinaryRowEx oldRow, BinaryRowEx newRow, @Nullable InternalTransaction tx);
/**
* Asynchronously gets an existed row associated with the same key columns values as the given one has, then replaces with the given
@@ -147,7 +148,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<BinaryRow> getAndReplace(BinaryRow row, @Nullable InternalTransaction tx);
+ CompletableFuture<BinaryRow> getAndReplace(BinaryRowEx row, @Nullable InternalTransaction tx);
/**
* Asynchronously deletes a row with the same key columns values as the given one from the table.
@@ -156,7 +157,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Boolean> delete(BinaryRow keyRow, @Nullable InternalTransaction tx);
+ CompletableFuture<Boolean> delete(BinaryRowEx keyRow, @Nullable InternalTransaction tx);
/**
* Asynchronously deletes given row from the table.
@@ -165,7 +166,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Boolean> deleteExact(BinaryRow oldRow, @Nullable InternalTransaction tx);
+ CompletableFuture<Boolean> deleteExact(BinaryRowEx oldRow, @Nullable InternalTransaction tx);
/**
* Asynchronously gets then deletes a row with the same key columns values from the table.
@@ -174,7 +175,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<BinaryRow> getAndDelete(BinaryRow row, @Nullable InternalTransaction tx);
+ CompletableFuture<BinaryRow> getAndDelete(BinaryRowEx row, @Nullable InternalTransaction tx);
/**
* Asynchronously remove rows with the same key columns values as the given one has from the table.
@@ -183,7 +184,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRow> rows, @Nullable InternalTransaction tx);
+ CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx);
/**
* Asynchronously remove given rows from the table.
@@ -192,7 +193,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Collection<BinaryRow>> deleteAllExact(Collection<BinaryRow> rows, @Nullable InternalTransaction tx);
+ CompletableFuture<Collection<BinaryRow>> deleteAllExact(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx);
/**
* Returns a partition for a key.
@@ -200,7 +201,7 @@ public interface InternalTable extends AutoCloseable {
* @param keyRow The key.
* @return The partition.
*/
- int partition(BinaryRow keyRow);
+ int partition(BinaryRowEx keyRow);
/**
* Scans given partition, providing {@link Publisher} that reactively notifies about partition rows.
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 8c6b51ce4..de010b3d1 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
@@ -26,6 +26,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.marshaller.TupleMarshallerException;
import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
@@ -108,7 +109,7 @@ public class KeyValueBinaryViewImpl extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Tuple> getOrDefaultAsync(@Nullable Transaction tx, @NotNull Tuple key, Tuple defaultValue) {
- BinaryRow keyRow = marshal(Objects.requireNonNull(key), null);
+ BinaryRowEx keyRow = marshal(Objects.requireNonNull(key), null);
return tbl.get(keyRow, (InternalTransaction) tx).thenApply(r -> IgniteUtils.nonNullOrElse(unmarshalValue(r), defaultValue));
}
@@ -122,7 +123,7 @@ public class KeyValueBinaryViewImpl extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Map<Tuple, Tuple>> getAllAsync(@Nullable Transaction tx, @NotNull Collection<Tuple> keys) {
- List<BinaryRow> keyRows = marshalKeys(Objects.requireNonNull(keys));
+ List<BinaryRowEx> keyRows = marshalKeys(Objects.requireNonNull(keys));
return tbl.getAll(keyRows, (InternalTransaction) tx).thenApply(this::unmarshalValue);
}
@@ -167,7 +168,7 @@ public class KeyValueBinaryViewImpl extends AbstractTableView implements KeyValu
public @NotNull CompletableFuture<Void> putAllAsync(@Nullable Transaction tx, @NotNull Map<@NotNull Tuple, @NotNull Tuple> pairs) {
Objects.requireNonNull(pairs);
- List<BinaryRow> rows = new ArrayList<>(pairs.size());
+ List<BinaryRowEx> rows = new ArrayList<>(pairs.size());
for (Map.Entry<Tuple, Tuple> pair : pairs.entrySet()) {
final Row row = marshal(Objects.requireNonNull(pair.getKey()), Objects.requireNonNull(pair.getValue()));
@@ -275,7 +276,7 @@ public class KeyValueBinaryViewImpl extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Collection<Tuple>> removeAllAsync(@Nullable Transaction tx, @NotNull Collection<Tuple> keys) {
- List<BinaryRow> keyRows = marshalKeys(Objects.requireNonNull(keys));
+ List<BinaryRowEx> keyRows = marshalKeys(Objects.requireNonNull(keys));
return tbl.deleteAll(keyRows, (InternalTransaction) tx)
.thenApply(this::unmarshalKeys);
@@ -492,8 +493,8 @@ public class KeyValueBinaryViewImpl extends AbstractTableView implements KeyValu
* @param keys Key tuples.
* @return Rows.
*/
- public List<BinaryRow> marshalKeys(@NotNull Collection<Tuple> keys) {
- List<BinaryRow> keyRows = new ArrayList<>(keys.size());
+ public List<BinaryRowEx> marshalKeys(@NotNull Collection<Tuple> keys) {
+ List<BinaryRowEx> keyRows = new ArrayList<>(keys.size());
for (Tuple keyRec : keys) {
final Row keyRow = marshal(Objects.requireNonNull(keyRec), null);
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 65d2afa7a..82db8bdd8 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
@@ -28,6 +28,7 @@ import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.schema.SchemaDescriptor;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.marshaller.KvMarshaller;
@@ -84,7 +85,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<V> getAsync(@Nullable Transaction tx, @NotNull K key) {
- BinaryRow keyRow = marshal(Objects.requireNonNull(key));
+ BinaryRowEx keyRow = marshal(Objects.requireNonNull(key));
return tbl.get(keyRow, (InternalTransaction) tx).thenApply(this::unmarshallValue);
}
@@ -98,7 +99,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<NullableValue<V>> getNullableAsync(@Nullable Transaction tx, @NotNull K key) {
- BinaryRow keyRow = marshal(Objects.requireNonNull(key));
+ BinaryRowEx keyRow = marshal(Objects.requireNonNull(key));
return tbl.get(keyRow, (InternalTransaction) tx).thenApply(r -> r == null ? null : NullableValue.of(unmarshalNullableValue(r)));
}
@@ -112,7 +113,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<V> getOrDefaultAsync(@Nullable Transaction tx, @NotNull K key, V defaultValue) {
- BinaryRow keyRow = marshal(Objects.requireNonNull(key));
+ BinaryRowEx keyRow = marshal(Objects.requireNonNull(key));
return tbl.get(keyRow, (InternalTransaction) tx).thenApply(r -> IgniteUtils.nonNullOrElse(unmarshalNullableValue(r), defaultValue));
}
@@ -126,7 +127,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Map<K, V>> getAllAsync(@Nullable Transaction tx, @NotNull Collection<K> keys) {
- Collection<BinaryRow> rows = marshal(Objects.requireNonNull(keys));
+ Collection<BinaryRowEx> rows = marshal(Objects.requireNonNull(keys));
return tbl.getAll(rows, (InternalTransaction) tx).thenApply(this::unmarshalPairs);
}
@@ -140,7 +141,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public CompletableFuture<Boolean> containsAsync(@Nullable Transaction tx, @NotNull K key) {
- BinaryRow keyRow = marshal(Objects.requireNonNull(key));
+ BinaryRowEx keyRow = marshal(Objects.requireNonNull(key));
return tbl.get(keyRow, (InternalTransaction) tx).thenApply(Objects::nonNull);
}
@@ -154,7 +155,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Void> putAsync(@Nullable Transaction tx, @NotNull K key, V val) {
- BinaryRow row = marshal(Objects.requireNonNull(key), val);
+ BinaryRowEx row = marshal(Objects.requireNonNull(key), val);
return tbl.upsert(row, (InternalTransaction) tx);
}
@@ -168,7 +169,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Void> putAllAsync(@Nullable Transaction tx, @NotNull Map<K, V> pairs) {
- Collection<BinaryRow> rows = marshal(Objects.requireNonNull(pairs));
+ Collection<BinaryRowEx> rows = marshal(Objects.requireNonNull(pairs));
return tbl.upsertAll(rows, (InternalTransaction) tx);
}
@@ -197,7 +198,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<NullableValue<V>> getNullableAndPutAsync(@Nullable Transaction tx, @NotNull K key, V val) {
- BinaryRow row = marshal(Objects.requireNonNull(key), val);
+ BinaryRowEx row = marshal(Objects.requireNonNull(key), val);
return tbl.getAndUpsert(row, (InternalTransaction) tx)
.thenApply(r -> r == null ? null : NullableValue.of(unmarshalNullableValue(r)));
@@ -212,7 +213,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Boolean> putIfAbsentAsync(@Nullable Transaction tx, @NotNull K key, V val) {
- BinaryRow row = marshal(Objects.requireNonNull(key), val);
+ BinaryRowEx row = marshal(Objects.requireNonNull(key), val);
return tbl.insert(row, (InternalTransaction) tx);
}
@@ -232,7 +233,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Boolean> removeAsync(@Nullable Transaction tx, @NotNull K key) {
- BinaryRow row = marshal(Objects.requireNonNull(key));
+ BinaryRowEx row = marshal(Objects.requireNonNull(key));
return tbl.delete(row, (InternalTransaction) tx);
}
@@ -240,7 +241,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Boolean> removeAsync(@Nullable Transaction tx, @NotNull K key, V val) {
- BinaryRow row = marshal(Objects.requireNonNull(key), val);
+ BinaryRowEx row = marshal(Objects.requireNonNull(key), val);
return tbl.deleteExact(row, (InternalTransaction) tx);
}
@@ -254,7 +255,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Collection<K>> removeAllAsync(@Nullable Transaction tx, @NotNull Collection<K> keys) {
- Collection<BinaryRow> rows = marshal(Objects.requireNonNull(keys));
+ Collection<BinaryRowEx> rows = marshal(Objects.requireNonNull(keys));
return tbl.deleteAll(rows, (InternalTransaction) tx).thenApply(this::unmarshalKeys);
}
@@ -268,7 +269,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<V> getAndRemoveAsync(@Nullable Transaction tx, @NotNull K key) {
- BinaryRow keyRow = marshal(Objects.requireNonNull(key));
+ BinaryRowEx keyRow = marshal(Objects.requireNonNull(key));
return tbl.getAndDelete(keyRow, (InternalTransaction) tx).thenApply(this::unmarshallValue);
}
@@ -282,7 +283,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<NullableValue<V>> getNullableAndRemoveAsync(@Nullable Transaction tx, @NotNull K key) {
- BinaryRow keyRow = marshal(Objects.requireNonNull(key));
+ BinaryRowEx keyRow = marshal(Objects.requireNonNull(key));
return tbl.getAndDelete(keyRow, (InternalTransaction) tx)
.thenApply(r -> r == null ? null : NullableValue.of(unmarshalNullableValue(r)));
@@ -303,7 +304,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull K key, V val) {
- BinaryRow row = marshal(Objects.requireNonNull(key), val);
+ BinaryRowEx row = marshal(Objects.requireNonNull(key), val);
return tbl.replace(row, (InternalTransaction) tx);
}
@@ -313,8 +314,8 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull K key, V oldVal, V newVal) {
Objects.requireNonNull(key);
- BinaryRow oldRow = marshal(key, oldVal);
- BinaryRow newRow = marshal(key, newVal);
+ BinaryRowEx oldRow = marshal(key, oldVal);
+ BinaryRowEx newRow = marshal(key, newVal);
return tbl.replace(oldRow, newRow, (InternalTransaction) tx);
}
@@ -343,7 +344,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<NullableValue<V>> getNullableAndReplaceAsync(@Nullable Transaction tx, @NotNull K key, V val) {
- BinaryRow row = marshal(Objects.requireNonNull(key), val);
+ BinaryRowEx row = marshal(Objects.requireNonNull(key), val);
return tbl.getAndReplace(row, (InternalTransaction) tx)
.thenApply(r -> r == null ? null : NullableValue.of(unmarshalNullableValue(r)));
@@ -416,7 +417,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
* @param key Key object.
* @return Binary row.
*/
- private BinaryRow marshal(@NotNull K key) {
+ private BinaryRowEx marshal(@NotNull K key) {
final KvMarshaller<K, V> marsh = marshaller(schemaReg.lastSchemaVersion());
try {
@@ -433,7 +434,7 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
* @param val Value object.
* @return Binary row.
*/
- private BinaryRow marshal(@NotNull K key, V val) {
+ private BinaryRowEx marshal(@NotNull K key, V val) {
final KvMarshaller<K, V> marsh = marshaller(schemaReg.lastSchemaVersion());
try {
@@ -450,14 +451,14 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
* @return Binary rows.
*/
@NotNull
- public Collection<BinaryRow> marshal(@NotNull Collection<K> keys) {
+ public Collection<BinaryRowEx> marshal(@NotNull Collection<K> keys) {
final KvMarshaller<K, V> marsh = marshaller(schemaReg.lastSchemaVersion());
- List<BinaryRow> keyRows = new ArrayList<>(keys.size());
+ List<BinaryRowEx> keyRows = new ArrayList<>(keys.size());
try {
for (K key : keys) {
- final BinaryRow keyRow = marsh.marshal(Objects.requireNonNull(key));
+ final BinaryRowEx keyRow = marsh.marshal(Objects.requireNonNull(key));
keyRows.add(keyRow);
}
@@ -475,14 +476,14 @@ public class KeyValueViewImpl<K, V> extends AbstractTableView implements KeyValu
* @return Binary rows.
*/
@NotNull
- public List<BinaryRow> marshal(@NotNull Map<K, V> pairs) {
+ public List<BinaryRowEx> marshal(@NotNull Map<K, V> pairs) {
final KvMarshaller<K, V> marsh = marshaller(schemaReg.lastSchemaVersion());
- List<BinaryRow> rows = new ArrayList<>(pairs.size());
+ List<BinaryRowEx> rows = new ArrayList<>(pairs.size());
try {
for (Map.Entry<K, V> pair : pairs.entrySet()) {
- final BinaryRow row = marsh.marshal(Objects.requireNonNull(pair.getKey()), pair.getValue());
+ final BinaryRowEx row = marsh.marshal(Objects.requireNonNull(pair.getKey()), pair.getValue());
rows.add(row);
}
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 9e37ca7ec..a9b5a3598 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
@@ -26,6 +26,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.marshaller.TupleMarshallerException;
import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
@@ -381,8 +382,8 @@ public class RecordBinaryViewImpl extends AbstractTableView implements RecordVie
* @param key {@code true} to marshal only a key.
* @return List of binary rows.
*/
- private Collection<BinaryRow> mapToBinary(Collection<Tuple> rows, boolean key) {
- Collection<BinaryRow> mapped = new ArrayList<>(rows.size());
+ private Collection<BinaryRowEx> mapToBinary(Collection<Tuple> rows, boolean key) {
+ Collection<BinaryRowEx> mapped = new ArrayList<>(rows.size());
for (Tuple row : rows) {
mapped.add(marshal(row, key));
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 a91016919..4ccce3540 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
@@ -27,6 +27,7 @@ import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.schema.SchemaDescriptor;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.marshaller.MarshallerException;
@@ -74,7 +75,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<R> getAsync(@Nullable Transaction tx, @NotNull R keyRec) {
- BinaryRow keyRow = marshalKey(Objects.requireNonNull(keyRec));
+ BinaryRowEx keyRow = marshalKey(Objects.requireNonNull(keyRec));
return tbl.get(keyRow, (InternalTransaction) tx).thenApply(this::unmarshal);
}
@@ -102,7 +103,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Void> upsertAsync(@Nullable Transaction tx, @NotNull R rec) {
- BinaryRow keyRow = marshal(Objects.requireNonNull(rec));
+ BinaryRowEx keyRow = marshal(Objects.requireNonNull(rec));
return tbl.upsert(keyRow, (InternalTransaction) tx);
}
@@ -130,7 +131,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<R> getAndUpsertAsync(@Nullable Transaction tx, @NotNull R rec) {
- BinaryRow keyRow = marshal(Objects.requireNonNull(rec));
+ BinaryRowEx keyRow = marshal(Objects.requireNonNull(rec));
return tbl.getAndUpsert(keyRow, (InternalTransaction) tx).thenApply(this::unmarshal);
}
@@ -144,7 +145,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Boolean> insertAsync(@Nullable Transaction tx, @NotNull R rec) {
- BinaryRow keyRow = marshal(Objects.requireNonNull(rec));
+ BinaryRowEx keyRow = marshal(Objects.requireNonNull(rec));
return tbl.insert(keyRow, (InternalTransaction) tx);
}
@@ -158,7 +159,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Collection<R>> insertAllAsync(@Nullable Transaction tx, @NotNull Collection<R> recs) {
- Collection<BinaryRow> rows = marshal(Objects.requireNonNull(recs));
+ Collection<BinaryRowEx> rows = marshal(Objects.requireNonNull(recs));
return tbl.insertAll(rows, (InternalTransaction) tx).thenApply(this::unmarshal);
}
@@ -178,7 +179,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull R rec) {
- BinaryRow newRow = marshal(Objects.requireNonNull(rec));
+ BinaryRowEx newRow = marshal(Objects.requireNonNull(rec));
return tbl.replace(newRow, (InternalTransaction) tx);
}
@@ -186,8 +187,8 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull R oldRec, @NotNull R newRec) {
- BinaryRow oldRow = marshal(Objects.requireNonNull(oldRec));
- BinaryRow newRow = marshal(Objects.requireNonNull(newRec));
+ BinaryRowEx oldRow = marshal(Objects.requireNonNull(oldRec));
+ BinaryRowEx newRow = marshal(Objects.requireNonNull(newRec));
return tbl.replace(oldRow, newRow, (InternalTransaction) tx);
}
@@ -201,7 +202,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<R> getAndReplaceAsync(@Nullable Transaction tx, @NotNull R rec) {
- BinaryRow row = marshal(Objects.requireNonNull(rec));
+ BinaryRowEx row = marshal(Objects.requireNonNull(rec));
return tbl.getAndReplace(row, (InternalTransaction) tx).thenApply(this::unmarshal);
}
@@ -215,7 +216,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Boolean> deleteAsync(@Nullable Transaction tx, @NotNull R keyRec) {
- BinaryRow row = marshalKey(Objects.requireNonNull(keyRec));
+ BinaryRowEx row = marshalKey(Objects.requireNonNull(keyRec));
return tbl.delete(row, (InternalTransaction) tx);
}
@@ -229,7 +230,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Boolean> deleteExactAsync(@Nullable Transaction tx, @NotNull R keyRec) {
- BinaryRow row = marshal(Objects.requireNonNull(keyRec));
+ BinaryRowEx row = marshal(Objects.requireNonNull(keyRec));
return tbl.deleteExact(row, (InternalTransaction) tx);
}
@@ -243,7 +244,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<R> getAndDeleteAsync(@Nullable Transaction tx, @NotNull R keyRec) {
- BinaryRow row = marshalKey(keyRec);
+ BinaryRowEx row = marshalKey(keyRec);
return tbl.getAndDelete(row, (InternalTransaction) tx).thenApply(this::unmarshal);
}
@@ -257,7 +258,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Collection<R>> deleteAllAsync(@Nullable Transaction tx, @NotNull Collection<R> keyRecs) {
- Collection<BinaryRow> rows = marshal(Objects.requireNonNull(keyRecs));
+ Collection<BinaryRowEx> rows = marshal(Objects.requireNonNull(keyRecs));
return tbl.deleteAll(rows, (InternalTransaction) tx).thenApply(this::unmarshal);
}
@@ -271,7 +272,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Collection<R>> deleteAllExactAsync(@Nullable Transaction tx, @NotNull Collection<R> keyRecs) {
- Collection<BinaryRow> rows = marshal(Objects.requireNonNull(keyRecs));
+ Collection<BinaryRowEx> rows = marshal(Objects.requireNonNull(keyRecs));
return tbl.deleteAllExact(rows, (InternalTransaction) tx).thenApply(this::unmarshal);
}
@@ -336,7 +337,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
* @param rec Record object.
* @return Binary row.
*/
- private BinaryRow marshal(@NotNull R rec) {
+ private BinaryRowEx marshal(@NotNull R rec) {
final RecordMarshaller<R> marsh = marshaller(schemaReg.lastSchemaVersion());
try {
@@ -352,14 +353,14 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
* @param recs Records collection.
* @return Binary rows collection.
*/
- private Collection<BinaryRow> marshal(@NotNull Collection<R> recs) {
+ private Collection<BinaryRowEx> marshal(@NotNull Collection<R> recs) {
final RecordMarshaller<R> marsh = marshaller(schemaReg.lastSchemaVersion());
- List<BinaryRow> rows = new ArrayList<>(recs.size());
+ List<BinaryRowEx> rows = new ArrayList<>(recs.size());
try {
for (R rec : recs) {
- final BinaryRow row = marsh.marshal(Objects.requireNonNull(rec));
+ final Row row = marsh.marshal(Objects.requireNonNull(rec));
rows.add(row);
}
@@ -376,7 +377,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
* @param rec Record key object.
* @return Binary row.
*/
- private BinaryRow marshalKey(@NotNull R rec) {
+ private BinaryRowEx marshalKey(@NotNull R rec) {
final RecordMarshaller<R> marsh = marshaller(schemaReg.lastSchemaVersion());
try {
@@ -392,14 +393,14 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
* @param recs Records collection.
* @return Binary rows collection.
*/
- private Collection<BinaryRow> marshalKeys(@NotNull Collection<R> recs) {
+ private Collection<BinaryRowEx> marshalKeys(@NotNull Collection<R> recs) {
final RecordMarshaller<R> marsh = marshaller(schemaReg.lastSchemaVersion());
- List<BinaryRow> rows = new ArrayList<>(recs.size());
+ List<BinaryRowEx> rows = new ArrayList<>(recs.size());
try {
for (R rec : recs) {
- final BinaryRow row = marsh.marshalKey(Objects.requireNonNull(rec));
+ final Row row = marsh.marshalKey(Objects.requireNonNull(rec));
rows.add(row);
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
index 3cbdcaed3..2fb907a6f 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
@@ -19,7 +19,7 @@ package org.apache.ignite.internal.table;
import java.util.Objects;
import java.util.UUID;
-import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.marshaller.MarshallerException;
import org.apache.ignite.internal.schema.marshaller.TupleMarshallerException;
@@ -137,7 +137,7 @@ public class TableImpl implements Table {
Objects.requireNonNull(key);
Objects.requireNonNull(keyMapper);
- BinaryRow keyRow;
+ BinaryRowEx keyRow;
var marshaller = new KvMarshallerImpl<>(schemaReg.schema(), keyMapper, keyMapper);
try {
keyRow = marshaller.marshal(key);
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
index 5e4bde4e9..c65eb0b46 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
@@ -38,6 +38,7 @@ import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.storage.engine.TableStorage;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.distributed.command.DeleteAllCommand;
@@ -179,7 +180,7 @@ public class InternalTableImpl implements InternalTable {
* @return The future.
*/
private <R, T> CompletableFuture<T> enlistInTx(
- Collection<BinaryRow> keyRows,
+ Collection<BinaryRowEx> keyRows,
InternalTransaction tx,
BiFunction<Collection<BinaryRow>, InternalTransaction, Command> op,
Function<CompletableFuture<R>[], CompletableFuture<T>> reducer
@@ -217,7 +218,7 @@ public class InternalTableImpl implements InternalTable {
* @return The future.
*/
private <R, T> CompletableFuture<T> enlistInTx(
- BinaryRow row,
+ BinaryRowEx row,
InternalTransaction tx,
Function<InternalTransaction, Command> op,
Function<R, T> trans
@@ -263,92 +264,92 @@ public class InternalTableImpl implements InternalTable {
/** {@inheritDoc} */
@Override
- public CompletableFuture<BinaryRow> get(BinaryRow keyRow, InternalTransaction tx) {
+ public CompletableFuture<BinaryRow> get(BinaryRowEx keyRow, InternalTransaction tx) {
return enlistInTx(keyRow, tx, tx0 -> new GetCommand(keyRow, tx0.timestamp()), SingleRowResponse::getValue);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRow> keyRows, InternalTransaction tx) {
+ public CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRowEx> keyRows, InternalTransaction tx) {
return enlistInTx(keyRows, tx, (rows0, tx0) -> new GetAllCommand(rows0, tx0.timestamp()), this::collectMultiRowsResponses);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> upsert(BinaryRow row, InternalTransaction tx) {
+ public CompletableFuture<Void> upsert(BinaryRowEx row, InternalTransaction tx) {
return enlistInTx(row, tx, tx0 -> new UpsertCommand(row, tx0.timestamp()), ignored -> null);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> upsertAll(Collection<BinaryRow> rows, InternalTransaction tx) {
+ public CompletableFuture<Void> upsertAll(Collection<BinaryRowEx> rows, InternalTransaction tx) {
return enlistInTx(rows, tx, (rows0, tx0) -> new UpsertAllCommand(rows0, tx0.timestamp()), CompletableFuture::allOf);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<BinaryRow> getAndUpsert(BinaryRow row, InternalTransaction tx) {
+ public CompletableFuture<BinaryRow> getAndUpsert(BinaryRowEx row, InternalTransaction tx) {
return enlistInTx(row, tx, tx0 -> new GetAndUpsertCommand(row, tx0.timestamp()), SingleRowResponse::getValue);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> insert(BinaryRow row, InternalTransaction tx) {
+ public CompletableFuture<Boolean> insert(BinaryRowEx row, InternalTransaction tx) {
return enlistInTx(row, tx, tx0 -> new InsertCommand(row, tx0.timestamp()), r -> (Boolean) r);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRow> rows, InternalTransaction tx) {
+ public CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRowEx> rows, InternalTransaction tx) {
return enlistInTx(rows, tx, (rows0, tx0) -> new InsertAllCommand(rows0, tx0.timestamp()), this::collectMultiRowsResponses);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> replace(BinaryRow row, InternalTransaction tx) {
+ public CompletableFuture<Boolean> replace(BinaryRowEx row, InternalTransaction tx) {
return enlistInTx(row, tx, tx0 -> new ReplaceIfExistCommand(row, tx0.timestamp()), r -> (Boolean) r);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> replace(BinaryRow oldRow, BinaryRow newRow, InternalTransaction tx) {
+ public CompletableFuture<Boolean> replace(BinaryRowEx oldRow, BinaryRowEx newRow, InternalTransaction tx) {
return enlistInTx(oldRow, tx, tx0 -> new ReplaceCommand(oldRow, newRow, tx0.timestamp()), r -> (Boolean) r);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<BinaryRow> getAndReplace(BinaryRow row, InternalTransaction tx) {
+ public CompletableFuture<BinaryRow> getAndReplace(BinaryRowEx row, InternalTransaction tx) {
return enlistInTx(row, tx, tx0 -> new GetAndReplaceCommand(row, tx0.timestamp()), SingleRowResponse::getValue);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> delete(BinaryRow keyRow, InternalTransaction tx) {
+ public CompletableFuture<Boolean> delete(BinaryRowEx keyRow, InternalTransaction tx) {
return enlistInTx(keyRow, tx, tx0 -> new DeleteCommand(keyRow, tx0.timestamp()), r -> (Boolean) r);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Boolean> deleteExact(BinaryRow oldRow, InternalTransaction tx) {
+ public CompletableFuture<Boolean> deleteExact(BinaryRowEx oldRow, InternalTransaction tx) {
return enlistInTx(oldRow, tx, tx0 -> new DeleteExactCommand(oldRow, tx0.timestamp()), r -> (Boolean) r);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<BinaryRow> getAndDelete(BinaryRow row, InternalTransaction tx) {
+ public CompletableFuture<BinaryRow> getAndDelete(BinaryRowEx row, InternalTransaction tx) {
return enlistInTx(row, tx, tx0 -> new GetAndDeleteCommand(row, tx0.timestamp()), SingleRowResponse::getValue);
}
/** {@inheritDoc} */
@Override
- public CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRow> rows, InternalTransaction tx) {
+ public CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRowEx> rows, InternalTransaction tx) {
return enlistInTx(rows, tx, (rows0, tx0) -> new DeleteAllCommand(rows0, tx0.timestamp()), this::collectMultiRowsResponses);
}
/** {@inheritDoc} */
@Override
public CompletableFuture<Collection<BinaryRow>> deleteAllExact(
- Collection<BinaryRow> rows,
+ Collection<BinaryRowEx> rows,
InternalTransaction tx
) {
return enlistInTx(rows, tx, (rows0, tx0) -> new DeleteExactAllCommand(rows0, tx0.timestamp()), this::collectMultiRowsResponses);
@@ -377,10 +378,10 @@ public class InternalTableImpl implements InternalTable {
* @param rows Rows.
* @return Partition -%gt; rows mapping.
*/
- private Int2ObjectOpenHashMap<List<BinaryRow>> mapRowsToPartitions(Collection<BinaryRow> rows) {
+ private Int2ObjectOpenHashMap<List<BinaryRow>> mapRowsToPartitions(Collection<BinaryRowEx> rows) {
Int2ObjectOpenHashMap<List<BinaryRow>> keyRowsByPartition = new Int2ObjectOpenHashMap<>();
- for (BinaryRow keyRow : rows) {
+ for (BinaryRowEx keyRow : rows) {
keyRowsByPartition.computeIfAbsent(partId(keyRow), k -> new ArrayList<>()).add(keyRow);
}
@@ -428,7 +429,7 @@ public class InternalTableImpl implements InternalTable {
/** {@inheritDoc} */
@TestOnly
@Override
- public int partition(BinaryRow keyRow) {
+ public int partition(BinaryRowEx keyRow) {
return partId(keyRow);
}
@@ -438,8 +439,8 @@ public class InternalTableImpl implements InternalTable {
* @param row Key row.
* @return partition id.
*/
- private int partId(BinaryRow row) {
- int partId = row.hash() % partitions;
+ private int partId(BinaryRowEx row) {
+ int partId = row.colocationHash() % partitions;
return (partId < 0) ? -partId : partId;
}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/ColocationHashCalculationTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/ColocationHashCalculationTest.java
index 38a4c7fa5..82f9ae2cd 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/ColocationHashCalculationTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/ColocationHashCalculationTest.java
@@ -36,7 +36,6 @@ import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
import org.apache.ignite.internal.schema.row.Row;
import org.apache.ignite.internal.schema.row.RowAssembler;
import org.apache.ignite.internal.table.impl.DummySchemaManagerImpl;
-import org.apache.ignite.internal.util.ColocationUtils;
import org.apache.ignite.internal.util.HashCalculator;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.table.Tuple;
@@ -103,7 +102,7 @@ public class ColocationHashCalculationTest {
new Column[]{new Column("val", INT32, true).copy(keyCols.length)});
Row r = generateRandomRow(rnd, schema);
- assertEquals(colocationHash(r), ColocationUtils.colocationHash(r));
+ assertEquals(colocationHash(r), r.colocationHash());
TupleMarshaller marshaller = new TupleMarshallerImpl(new DummySchemaManagerImpl(schema));
for (int i = 0; i < 10; ++i) {
@@ -114,7 +113,7 @@ public class ColocationHashCalculationTest {
r = new Row(schema, new ByteBufferRow(marshaller.marshal(t).bytes()));
- assertEquals(colocationHash(r), ColocationUtils.colocationHash(r));
+ assertEquals(colocationHash(r), r.colocationHash());
}
}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java b/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
index 3d6a0f14b..4e38735d8 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
@@ -30,6 +30,7 @@ import java.util.concurrent.Flow;
import java.util.function.BiFunction;
import javax.naming.OperationNotSupportedException;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.storage.engine.TableStorage;
import org.apache.ignite.internal.table.distributed.command.GetAllCommand;
import org.apache.ignite.internal.table.distributed.command.GetCommand;
@@ -154,7 +155,7 @@ public class DummyInternalTableImpl extends InternalTableImpl {
/** {@inheritDoc} */
@Override
- public CompletableFuture<BinaryRow> get(BinaryRow keyRow, InternalTransaction tx) {
+ public CompletableFuture<BinaryRow> get(BinaryRowEx keyRow, InternalTransaction tx) {
return super.get(keyRow, tx);
}
@@ -172,7 +173,7 @@ public class DummyInternalTableImpl extends InternalTableImpl {
/** {@inheritDoc} */
@Override
- public int partition(BinaryRow keyRow) {
+ public int partition(BinaryRowEx keyRow) {
return 0;
}
}