You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by vj...@apache.org on 2026/08/15 18:29:37 UTC
(phoenix) branch master updated: PHOENIX-7982 Data corruption in UPSERT SELECT with VARBINARY_ENCODED (#2599)
This is an automated email from the ASF dual-hosted git repository.
virajjasani pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/master by this push:
new 05153be193 PHOENIX-7982 Data corruption in UPSERT SELECT with VARBINARY_ENCODED (#2599)
05153be193 is described below
commit 05153be193b7392245b0816a001e92f912d3bffe
Author: Viraj Jasani <vj...@apache.org>
AuthorDate: Sat Aug 15 11:29:32 2026 -0700
PHOENIX-7982 Data corruption in UPSERT SELECT with VARBINARY_ENCODED (#2599)
---
.../org/apache/phoenix/compile/UpsertCompiler.java | 14 +-
.../phoenix/expression/CoerceExpression.java | 36 +--
.../org/apache/phoenix/schema/types/PBinary.java | 3 +
.../org/apache/phoenix/schema/types/PDataType.java | 6 +-
.../apache/phoenix/schema/types/PVarbinary.java | 5 +-
.../phoenix/end2end/VarBinaryEncoded1IT.java | 290 ++++++++++++++++++++-
.../VarBinaryEncodedIndexUpsertSelectIT.java | 140 ++++++++++
.../VarBinaryEncodedServerUpsertSelectIT.java | 118 +++++++++
.../VarBinaryEncodedUpsertSelectTestUtil.java | 173 ++++++++++++
.../java/org/apache/phoenix/query/BaseTest.java | 2 +-
.../apache/phoenix/schema/types/PDataTypeTest.java | 32 +++
11 files changed, 779 insertions(+), 40 deletions(-)
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
index b9dfbc33e5..106f5071e3 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
@@ -112,6 +112,7 @@ import org.apache.phoenix.schema.types.PSmallint;
import org.apache.phoenix.schema.types.PTimestamp;
import org.apache.phoenix.schema.types.PUnsignedLong;
import org.apache.phoenix.schema.types.PVarbinary;
+import org.apache.phoenix.schema.types.PVarbinaryEncoded;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.ExpressionUtil;
import org.apache.phoenix.util.IndexUtil;
@@ -271,9 +272,16 @@ public class UpsertCompiler {
throw new DataExceedsCapacityException(column.getDataType(), column.getMaxLength(),
column.getScale(), column.getName().getString());
}
- column.getDataType().coerceBytes(ptr, value, column.getDataType(), precision, scale,
- SortOrder.getDefault(), column.getMaxLength(), column.getScale(), column.getSortOrder(),
- table.rowKeyOrderOptimizable());
+ if (column.getDataType() == PVarbinaryEncoded.INSTANCE) {
+ // ptr holds the decoded value, because that is what the result set hands back for
+ // VARBINARY_ENCODED. coerceBytes() cannot restore the encoding here: it treats all
+ // binary types as byte comparable and so leaves ptr untouched (PHOENIX-7982).
+ ptr.set(column.getDataType().toBytes(value, column.getSortOrder()));
+ } else {
+ column.getDataType().coerceBytes(ptr, value, column.getDataType(), precision, scale,
+ SortOrder.getDefault(), column.getMaxLength(), column.getScale(),
+ column.getSortOrder(), table.rowKeyOrderOptimizable());
+ }
values[j] = ByteUtil.copyKeyBytesIfNecessary(ptr);
}
setValues(values, pkSlotIndexes, columnIndexes, table, mutation, statement,
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/expression/CoerceExpression.java b/phoenix-core-client/src/main/java/org/apache/phoenix/expression/CoerceExpression.java
index e3b7b60ed8..b8ed96a204 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/expression/CoerceExpression.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/expression/CoerceExpression.java
@@ -27,10 +27,7 @@ import org.apache.hadoop.io.WritableUtils;
import org.apache.phoenix.expression.visitor.ExpressionVisitor;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.tuple.Tuple;
-import org.apache.phoenix.schema.types.PBinary;
import org.apache.phoenix.schema.types.PDataType;
-import org.apache.phoenix.schema.types.PVarbinary;
-import org.apache.phoenix.schema.types.PVarbinaryEncoded;
import org.apache.phoenix.thirdparty.com.google.common.base.Preconditions;
import org.apache.phoenix.thirdparty.com.google.common.collect.ImmutableList;
@@ -164,35 +161,10 @@ public class CoerceExpression extends BaseSingleExpression {
@Override
public boolean evaluate(Tuple tuple, ImmutableBytesWritable ptr) {
- // For CoerceExpression evaluation, lhs is coerced to rhs literal expression. However,
- // in case of variable length binary literal expression, literal value by default
- // gets VARBINARY data type. If lhs expression is of type VARBINARY_ENCODED, we should
- // encode rhs literal value to VARBINARY_ENCODED type. This makes the eventual coerce
- // evaluation successful.
- if (
- getChild() instanceof LiteralExpression
- && (getChild().getDataType() == PVarbinary.INSTANCE
- || getChild().getDataType() == PBinary.INSTANCE)
- && getDataType() == PVarbinaryEncoded.INSTANCE
- ) {
- Expression expression;
- try {
- expression = LiteralExpression.newConstant(((LiteralExpression) getChild()).getValue(),
- PVarbinaryEncoded.INSTANCE);
- } catch (SQLException e) {
- throw new RuntimeException(e);
- }
- if (expression.evaluate(tuple, ptr)) {
- getDataType().coerceBytes(ptr, null, expression.getDataType(), expression.getMaxLength(),
- null, expression.getSortOrder(), maxLength, null, getSortOrder(), rowKeyOrderOptimizable);
- return true;
- }
- } else {
- if (getChild().evaluate(tuple, ptr)) {
- getDataType().coerceBytes(ptr, null, getChild().getDataType(), getChild().getMaxLength(),
- null, getChild().getSortOrder(), maxLength, null, getSortOrder(), rowKeyOrderOptimizable);
- return true;
- }
+ if (getChild().evaluate(tuple, ptr)) {
+ getDataType().coerceBytes(ptr, null, getChild().getDataType(), getChild().getMaxLength(),
+ null, getChild().getSortOrder(), maxLength, null, getSortOrder(), rowKeyOrderOptimizable);
+ return true;
}
return false;
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PBinary.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PBinary.java
index 31ffae9d90..14d36ada81 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PBinary.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PBinary.java
@@ -124,6 +124,9 @@ public class PBinary extends PBinaryBase {
@Override
public Object toObject(Object object, PDataType actualType) {
+ if (actualType == PVarbinaryEncoded.INSTANCE) {
+ return object;
+ }
return actualType.toBytes(object);
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PDataType.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PDataType.java
index 32170b0919..cf1bcb93fb 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PDataType.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PDataType.java
@@ -91,8 +91,10 @@ public abstract class PDataType<T> implements DataType<T>, Comparable<PDataType<
}
public boolean isBytesComparableWith(PDataType otherType) {
- return equalsAny(this, otherType, PVarbinary.INSTANCE, PBinary.INSTANCE,
- PVarbinaryEncoded.INSTANCE);
+ if (this == PVarbinaryEncoded.INSTANCE || otherType == PVarbinaryEncoded.INSTANCE) {
+ return this.equals(otherType);
+ }
+ return equalsAny(this, otherType, PVarbinary.INSTANCE, PBinary.INSTANCE);
}
/** Returns true if {@link PDataType} can be declared as primary key otherwise false. */
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PVarbinary.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PVarbinary.java
index e13c6444e8..8453e36abe 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PVarbinary.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/PVarbinary.java
@@ -88,6 +88,9 @@ public class PVarbinary extends PBinaryBase {
@Override
public Object toObject(Object object, PDataType actualType) {
+ if (actualType == PVarbinaryEncoded.INSTANCE) {
+ return object;
+ }
return actualType.toBytes(object);
}
@@ -121,7 +124,7 @@ public class PVarbinary extends PBinaryBase {
} else if (rhs == null) {
return 1;
}
- if (equalsAny(rhsType, this, PBinary.INSTANCE)) {
+ if (equalsAny(rhsType, this, PBinary.INSTANCE, PVarbinaryEncoded.INSTANCE)) {
return Bytes.compareTo((byte[]) lhs, (byte[]) rhs);
} else {
byte[] rhsBytes = rhsType.toBytes(rhs);
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncoded1IT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncoded1IT.java
index 88f6151778..220edee05c 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncoded1IT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncoded1IT.java
@@ -17,6 +17,13 @@
*/
package org.apache.phoenix.end2end;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.MULTI_PK_COLUMNS;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.assertRows;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.assertRowsAndRowKeys;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.createMultiColumnPkTable;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.createTable;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.upsertEncodedMultiPkRow;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.upsertEncodedRow;
import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
import java.sql.Connection;
@@ -76,7 +83,8 @@ public class VarBinaryEncoded1IT extends ParallelStatsDisabledIT {
}
@Parameterized.Parameters(
- name = "VarBinary1IT_columnEncoded={0}, transactionProvider={1}, mutable={2}")
+ name = "VarBinary1IT_columnEncoded={0}, transactionProvider={1}, mutable={2},"
+ + " isBindStatement={3}")
public static synchronized Collection<Object[]> data() {
return Arrays.asList(new Object[][] { { false, null, false, false },
{ false, "OMID", false, false }, { false, null, true, false }, { false, "OMID", true, false },
@@ -2268,4 +2276,284 @@ public class VarBinaryEncoded1IT extends ParallelStatsDisabledIT {
preparedStatement.executeUpdate();
}
+ @Test
+ public void testUpsertSelectVarBinaryEncodedRowKey() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String sourceTable = generateUniqueName();
+ final String targetTable = generateUniqueName();
+ byte[] pk1 = new byte[] { 1, 0, 2 };
+ byte[] col1 = new byte[] { 10, 20, 30 };
+ byte[] pk2 = new byte[] { 0, 0, 3, 0 };
+ byte[] col2 = new byte[] { 40, 50 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ createTable(conn, sourceTable, tableDDLOptions);
+ createTable(conn, targetTable, tableDDLOptions);
+
+ upsertEncodedRow(conn, sourceTable, pk1, col1, isBindStatement);
+ upsertEncodedRow(conn, sourceTable, pk2, col2, isBindStatement);
+ upsertEncodedRow(conn, targetTable, pk1, col1, isBindStatement);
+ upsertEncodedRow(conn, targetTable, pk2, col2, isBindStatement);
+ conn.commit();
+
+ byte[][] row2 = new byte[][] { pk2, col2 };
+ byte[][] row1 = new byte[][] { pk1, col1 };
+
+ assertRowsAndRowKeys(conn, sourceTable, row2, row1);
+ assertRowsAndRowKeys(conn, targetTable, row2, row1);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + targetTable + " (PK1, COL1) SELECT PK1, COL1 FROM " + sourceTable);
+ conn.commit();
+
+ assertRowsAndRowKeys(conn, targetTable, row2, row1);
+ }
+ }
+
+ @Test
+ public void testUpsertSelectVarBinaryEncodedColumnValue() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String sourceTable = generateUniqueName();
+ final String targetTable = generateUniqueName();
+ byte[] pk = new byte[] { 1, 2, 3 };
+ byte[] col = new byte[] { 0, -1, 5 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ createTable(conn, sourceTable, tableDDLOptions);
+ createTable(conn, targetTable, tableDDLOptions);
+
+ upsertEncodedRow(conn, sourceTable, pk, col, isBindStatement);
+
+ byte[][] row = new byte[][] { pk, col };
+ assertRowsAndRowKeys(conn, sourceTable, row);
+ conn.createStatement().execute(
+ "UPSERT INTO " + targetTable + " (PK1, COL1) SELECT PK1, COL1 FROM " + sourceTable);
+
+ assertRowsAndRowKeys(conn, targetTable, row);
+ }
+ }
+
+ @Test
+ public void testUpsertSelectVarBinaryIntoVarBinaryEncoded() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String sourceTable = generateUniqueName();
+ final String targetTable = generateUniqueName();
+
+ byte[] pk = new byte[] { 1, 2, 3 };
+ byte[] col = new byte[] { 0, -1, 5 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.createStatement()
+ .execute("CREATE TABLE " + sourceTable + " (PK1 VARBINARY NOT NULL, COL1 VARBINARY"
+ + " CONSTRAINT pk PRIMARY KEY(PK1)) " + tableDDLOptions);
+ createTable(conn, targetTable, tableDDLOptions);
+
+ try (PreparedStatement preparedStatement =
+ conn.prepareStatement("UPSERT INTO " + sourceTable + " (PK1, COL1) VALUES (?, ?)")) {
+ preparedStatement.setBytes(1, pk);
+ preparedStatement.setBytes(2, col);
+ preparedStatement.executeUpdate();
+ }
+ conn.commit();
+
+ assertRowsAndRowKeys(conn, sourceTable, new byte[][] { pk, col });
+ conn.createStatement().execute(
+ "UPSERT INTO " + targetTable + " (PK1, COL1) SELECT PK1, COL1 FROM " + sourceTable);
+ conn.commit();
+
+ assertRowsAndRowKeys(conn, targetTable, new byte[][] { pk, col });
+ }
+ }
+
+ @Test
+ public void testUpsertSelectVarBinaryEncodedIntoVarBinary() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String sourceTable = generateUniqueName();
+ final String targetTable = generateUniqueName();
+ byte[] pk = new byte[] { 1, 2, 3 };
+ byte[] col = new byte[] { 0, -1, 5 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ createTable(conn, sourceTable, tableDDLOptions);
+ conn.createStatement()
+ .execute("CREATE TABLE " + targetTable + " (PK1 VARBINARY NOT NULL, COL1 VARBINARY"
+ + " CONSTRAINT pk PRIMARY KEY(PK1)) " + tableDDLOptions);
+
+ upsertEncodedRow(conn, sourceTable, pk, col, isBindStatement);
+ conn.commit();
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + targetTable + " (PK1, COL1) SELECT PK1, COL1 FROM " + sourceTable);
+ conn.commit();
+
+ try (ResultSet resultSet =
+ conn.createStatement().executeQuery("SELECT PK1, COL1 FROM " + targetTable)) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertArrayEquals(pk, resultSet.getBytes(1));
+ Assert.assertArrayEquals(col, resultSet.getBytes(2));
+ Assert.assertFalse(resultSet.next());
+ }
+ }
+ }
+
+ @Test
+ public void testUpsertSelectVarBinaryEncodedSameTableOnClient() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String tableName = generateUniqueName();
+
+ byte[] pk1 = new byte[] { 1, 0, 2 };
+ byte[] col1 = new byte[] { 0, -1, 5 };
+ byte[] pk2 = new byte[] { 0, 0, 3, 0 };
+ byte[] col2 = new byte[] { 40, 50 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(false);
+ createTable(conn, tableName, tableDDLOptions);
+
+ upsertEncodedRow(conn, tableName, pk1, col1, isBindStatement);
+ upsertEncodedRow(conn, tableName, pk2, col2, isBindStatement);
+ conn.commit();
+
+ byte[][] row2 = new byte[][] { pk2, col2 };
+ byte[][] row1 = new byte[][] { pk1, col1 };
+ assertRowsAndRowKeys(conn, tableName, row2, row1);
+
+ conn.createStatement()
+ .execute("UPSERT INTO " + tableName + " (PK1, COL1) SELECT PK1, COL1 FROM " + tableName);
+ conn.commit();
+
+ assertRowsAndRowKeys(conn, tableName, row2, row1);
+ }
+ }
+
+ @Test
+ public void testUpsertSelectBinaryIntoVarBinaryEncoded() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String sourceTable = generateUniqueName();
+ final String targetTable = generateUniqueName();
+
+ byte[] pk = new byte[] { 1, 2, 3 };
+ byte[] col = new byte[] { 0, -1, 5 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.createStatement()
+ .execute("CREATE TABLE " + sourceTable + " (PK1 BINARY(3) NOT NULL, COL1 BINARY(3)"
+ + " CONSTRAINT pk PRIMARY KEY(PK1)) " + tableDDLOptions);
+ createTable(conn, targetTable, tableDDLOptions);
+
+ try (PreparedStatement preparedStatement =
+ conn.prepareStatement("UPSERT INTO " + sourceTable + " (PK1, COL1) VALUES (?, ?)")) {
+ preparedStatement.setBytes(1, pk);
+ preparedStatement.setBytes(2, col);
+ preparedStatement.executeUpdate();
+ }
+ conn.commit();
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + targetTable + " (PK1, COL1) SELECT PK1, COL1 FROM " + sourceTable);
+ conn.commit();
+
+ assertRowsAndRowKeys(conn, sourceTable, new byte[][] { pk, col });
+ assertRowsAndRowKeys(conn, targetTable, new byte[][] { pk, col });
+ }
+ }
+
+ @Test
+ public void testUpsertSelectVarBinaryEncodedIntoBinary() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String sourceTable = generateUniqueName();
+ final String targetTable = generateUniqueName();
+
+ byte[] pk = new byte[] { 1, 2, 3 };
+ byte[] col = new byte[] { 0, -1, 5 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ createTable(conn, sourceTable, tableDDLOptions);
+ conn.createStatement()
+ .execute("CREATE TABLE " + targetTable + " (PK1 BINARY(3) NOT NULL, COL1 BINARY(3)"
+ + " CONSTRAINT pk PRIMARY KEY(PK1)) " + tableDDLOptions);
+
+ upsertEncodedRow(conn, sourceTable, pk, col, isBindStatement);
+ conn.commit();
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + targetTable + " (PK1, COL1) SELECT PK1, COL1 FROM " + sourceTable);
+ conn.commit();
+
+ try (ResultSet resultSet =
+ conn.createStatement().executeQuery("SELECT PK1, COL1 FROM " + targetTable)) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertArrayEquals(pk, resultSet.getBytes(1));
+ Assert.assertArrayEquals(col, resultSet.getBytes(2));
+ Assert.assertFalse(resultSet.next());
+ }
+ }
+ }
+
+ @Test
+ public void testUpsertSelectVarBinaryEncodedMultiColumnPk() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String sourceTable = generateUniqueName();
+ final String targetTable = generateUniqueName();
+
+ byte[] pk1 = new byte[] { 1, 0, 2 };
+ byte[] pk2a = new byte[] { 9, 0, 1 };
+ byte[] col1a = new byte[] { 0, -1, 5 };
+ byte[] pk2b = new byte[] { 0, 3, 0, 0 };
+ byte[] col1b = new byte[] { 7, 0 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ createMultiColumnPkTable(conn, sourceTable, tableDDLOptions);
+ createMultiColumnPkTable(conn, targetTable, tableDDLOptions);
+
+ upsertEncodedMultiPkRow(conn, sourceTable, pk1, pk2a, col1a, isBindStatement);
+ upsertEncodedMultiPkRow(conn, sourceTable, pk1, pk2b, col1b, isBindStatement);
+ upsertEncodedMultiPkRow(conn, targetTable, pk1, pk2a, col1a, isBindStatement);
+ upsertEncodedMultiPkRow(conn, targetTable, pk1, pk2b, col1b, isBindStatement);
+ conn.commit();
+
+ byte[][] rowA = new byte[][] { pk1, pk2a, col1a };
+ byte[][] rowB = new byte[][] { pk1, pk2b, col1b };
+
+ assertRows(conn, sourceTable, MULTI_PK_COLUMNS, rowA, rowB);
+ assertRows(conn, targetTable, MULTI_PK_COLUMNS, rowA, rowB);
+
+ conn.createStatement().execute("UPSERT INTO " + targetTable
+ + " (PK1, PK2, COL1) SELECT PK1, PK2, COL1 FROM " + sourceTable);
+ conn.commit();
+
+ assertRows(conn, targetTable, MULTI_PK_COLUMNS, rowA, rowB);
+ }
+ }
+
+ @Test
+ public void testUpsertSelectVarBinaryEncodedSameTable() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String tableName = generateUniqueName();
+
+ byte[] pk1 = new byte[] { 1, 0, 2 };
+ byte[] col1 = new byte[] { 0, -1, 5 };
+ byte[] pk2 = new byte[] { 0, 0, 3, 0 };
+ byte[] col2 = new byte[] { 40, 50 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ createTable(conn, tableName, tableDDLOptions);
+
+ upsertEncodedRow(conn, tableName, pk1, col1, isBindStatement);
+ upsertEncodedRow(conn, tableName, pk2, col2, isBindStatement);
+ conn.commit();
+
+ byte[][] row2 = new byte[][] { pk2, col2 };
+ byte[][] row1 = new byte[][] { pk1, col1 };
+ assertRowsAndRowKeys(conn, tableName, row2, row1);
+
+ conn.setAutoCommit(true);
+ conn.createStatement()
+ .execute("UPSERT INTO " + tableName + " (PK1, COL1) SELECT PK1, COL1 FROM " + tableName);
+
+ assertRowsAndRowKeys(conn, tableName, row2, row1);
+ }
+ }
+
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncodedIndexUpsertSelectIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncodedIndexUpsertSelectIT.java
new file mode 100644
index 0000000000..4962db30bd
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncodedIndexUpsertSelectIT.java
@@ -0,0 +1,140 @@
+/*
+ * 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.phoenix.end2end;
+
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.SINGLE_PK_COLUMNS;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.assertRows;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.countHBaseRows;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.createTable;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.upsertEncodedRow;
+import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_RETRY_PAUSE_MS;
+import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_MUTATION_SERIALIZE;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+
+@Category(NeedsOwnMiniClusterTest.class)
+@RunWith(Parameterized.class)
+public class VarBinaryEncodedIndexUpsertSelectIT extends ParallelStatsDisabledIT {
+
+ private static final long EVENTUAL_CONSISTENCY_WAIT_MS = 12000;
+
+ private final boolean coveredIndex;
+ private final boolean eventual;
+
+ public VarBinaryEncodedIndexUpsertSelectIT(boolean coveredIndex, boolean eventual) {
+ this.coveredIndex = coveredIndex;
+ this.eventual = eventual;
+ }
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(5);
+ props.put(BaseScannerRegionObserverConstants.PHOENIX_MAX_LOOKBACK_AGE_CONF_KEY,
+ Integer.toString(60 * 60));
+ props.put(QueryServices.USE_STATS_FOR_PARALLELIZATION, Boolean.toString(false));
+ props.put(INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS, Integer.toString(200));
+ props.put(INDEX_CDC_CONSUMER_RETRY_PAUSE_MS, Integer.toString(5));
+ props.put(PHOENIX_INDEX_CDC_MUTATION_SERIALIZE, Boolean.FALSE.toString());
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ }
+
+ @Parameterized.Parameters(name = "VarBinaryIndexUpsertSelectIT_coveredIndex={0}, eventual={1}")
+ public static synchronized Collection<Object[]> data() {
+ List<Object[]> params = new ArrayList<>();
+ for (boolean coveredIndex : new boolean[] { false, true }) {
+ for (boolean eventual : new boolean[] { false, true }) {
+ params.add(new Object[] { coveredIndex, eventual });
+ }
+ }
+ return params;
+ }
+
+ private void waitForEventualConsistency() throws InterruptedException {
+ if (eventual) {
+ Thread.sleep(EVENTUAL_CONSISTENCY_WAIT_MS);
+ }
+ }
+
+ @Test
+ public void testUpsertSelectVarBinaryEncodedWithIndex() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String sourceTable = generateUniqueName();
+ final String targetTable = generateUniqueName();
+ final String indexName = generateUniqueName();
+
+ byte[] pk1 = new byte[] { 1, 0, 2 };
+ byte[] col1 = new byte[] { 0, -1, 5 };
+ byte[] pk2 = new byte[] { 0, 0, 3, 0 };
+ byte[] col2 = new byte[] { 7, 0 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ createTable(conn, sourceTable, "");
+ createTable(conn, targetTable, "");
+
+ String consistencyClause = eventual ? " CONSISTENCY = EVENTUAL" : " CONSISTENCY = STRONG";
+ if (this.coveredIndex) {
+ conn.createStatement().execute(
+ "CREATE INDEX " + indexName + " ON " + targetTable + " (COL1, COL2)" + consistencyClause);
+ } else {
+ conn.createStatement().execute("CREATE UNCOVERED INDEX " + indexName + " ON " + targetTable
+ + " (COL1, COL2)" + consistencyClause);
+ }
+
+ upsertEncodedRow(conn, sourceTable, pk1, col1, "TEXT1", true);
+ upsertEncodedRow(conn, sourceTable, pk2, col2, "TEXT2", true);
+ upsertEncodedRow(conn, targetTable, pk1, col1, "TEXT1", true);
+ upsertEncodedRow(conn, targetTable, pk2, col2, "TEXT2", true);
+ conn.commit();
+ waitForEventualConsistency();
+
+ byte[][] row2 = new byte[][] { pk2, col2 };
+ byte[][] row1 = new byte[][] { pk1, col1 };
+ Assert.assertEquals("index rows", 2, countHBaseRows(conn, indexName));
+ assertRows(conn, targetTable, SINGLE_PK_COLUMNS, row2, row1);
+
+ conn.createStatement().execute("UPSERT INTO " + targetTable
+ + " (PK1, COL1, COL2) SELECT PK1, COL1, COL2 FROM " + sourceTable);
+ conn.commit();
+ waitForEventualConsistency();
+
+ Assert.assertEquals("index rows", 2, countHBaseRows(conn, indexName));
+ assertRows(conn, targetTable, SINGLE_PK_COLUMNS, row2, row1);
+ }
+ }
+
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncodedServerUpsertSelectIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncodedServerUpsertSelectIT.java
new file mode 100644
index 0000000000..9cd931c668
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncodedServerUpsertSelectIT.java
@@ -0,0 +1,118 @@
+/*
+ * 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.phoenix.end2end;
+
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.SINGLE_PK_COLUMNS;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.assertRows;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.assertRowsAndRowKeys;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.createTable;
+import static org.apache.phoenix.end2end.VarBinaryEncodedUpsertSelectTestUtil.upsertEncodedRow;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+
+@Category(NeedsOwnMiniClusterTest.class)
+public class VarBinaryEncodedServerUpsertSelectIT extends ParallelStatsDisabledIT {
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(1);
+ props.put(QueryServices.ENABLE_SERVER_UPSERT_SELECT, Boolean.TRUE.toString());
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ }
+
+ @Test
+ public void testServerUpsertSelectVarBinaryEncoded() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String sourceTable = generateUniqueName();
+ final String targetTable = generateUniqueName();
+
+ byte[] pk1 = new byte[] { 1, 0, 2 };
+ byte[] col1 = new byte[] { 0, -1, 5 };
+ byte[] pk2 = new byte[] { 0, 0, 3, 0 };
+ byte[] col2 = new byte[] { 40, 50 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ createTable(conn, sourceTable, "");
+ createTable(conn, targetTable, "");
+
+ upsertEncodedRow(conn, sourceTable, pk1, col1);
+ upsertEncodedRow(conn, sourceTable, pk2, col2);
+ upsertEncodedRow(conn, targetTable, pk1, col1);
+ upsertEncodedRow(conn, targetTable, pk2, col2);
+
+ byte[][] row2 = new byte[][] { pk2, col2 };
+ byte[][] row1 = new byte[][] { pk1, col1 };
+ assertRowsAndRowKeys(conn, sourceTable, row2, row1);
+ assertRowsAndRowKeys(conn, targetTable, row2, row1);
+
+ String upsertSelect =
+ "UPSERT INTO " + targetTable + " (PK1, COL1) SELECT PK1, COL1 FROM " + sourceTable;
+ conn.createStatement().execute(upsertSelect);
+
+ assertRowsAndRowKeys(conn, targetTable, row2, row1);
+ }
+ }
+
+ @Test
+ public void testServerUpsertSelectVarBinaryEncodedIntoDescendingPk() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ final String sourceTable = generateUniqueName();
+ final String targetTable = generateUniqueName();
+
+ byte[] pk1 = new byte[] { 1, 0, 2 };
+ byte[] col1 = new byte[] { 0, -1, 5 };
+ byte[] pk2 = new byte[] { 0, 0, 3, 0 };
+ byte[] col2 = new byte[] { 40, 50 };
+
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ createTable(conn, sourceTable, "");
+ createTable(conn, targetTable, "", true);
+
+ upsertEncodedRow(conn, sourceTable, pk1, col1);
+ upsertEncodedRow(conn, sourceTable, pk2, col2);
+ upsertEncodedRow(conn, targetTable, pk1, col1);
+ upsertEncodedRow(conn, targetTable, pk2, col2);
+
+ byte[][] row1 = new byte[][] { pk1, col1 };
+ byte[][] row2 = new byte[][] { pk2, col2 };
+ assertRowsAndRowKeys(conn, sourceTable, row2, row1);
+ assertRows(conn, targetTable, SINGLE_PK_COLUMNS, row1, row2);
+
+ String upsertSelect =
+ "UPSERT INTO " + targetTable + " (PK1, COL1) SELECT PK1, COL1 FROM " + sourceTable;
+ conn.createStatement().execute(upsertSelect);
+
+ assertRows(conn, targetTable, SINGLE_PK_COLUMNS, row1, row2);
+ }
+ }
+
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncodedUpsertSelectTestUtil.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncodedUpsertSelectTestUtil.java
new file mode 100644
index 0000000000..a62a8b27d8
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncodedUpsertSelectTestUtil.java
@@ -0,0 +1,173 @@
+/*
+ * 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.phoenix.end2end;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.types.PVarbinary;
+import org.apache.phoenix.schema.types.PVarbinaryEncoded;
+import org.junit.Assert;
+
+final class VarBinaryEncodedUpsertSelectTestUtil {
+
+ static final String SINGLE_PK_COLUMNS = "PK1, COL1";
+
+ static final String MULTI_PK_COLUMNS = "PK1, PK2, COL1";
+
+ private VarBinaryEncodedUpsertSelectTestUtil() {
+ }
+
+ static void createTable(Connection conn, String tableName, String tableOptions)
+ throws SQLException {
+ createTable(conn, tableName, tableOptions, false);
+ }
+
+ static void createTable(Connection conn, String tableName, String tableOptions,
+ boolean descendingPk) throws SQLException {
+ conn.createStatement()
+ .execute("CREATE TABLE " + tableName + " (PK1 VARBINARY_ENCODED NOT NULL,"
+ + " COL1 VARBINARY_ENCODED, COL2 VARCHAR CONSTRAINT pk PRIMARY KEY(PK1"
+ + (descendingPk ? " DESC" : "") + ")) " + tableOptions);
+ }
+
+ static void createMultiColumnPkTable(Connection conn, String tableName, String tableOptions)
+ throws SQLException {
+ conn.createStatement()
+ .execute("CREATE TABLE " + tableName + " (PK1 VARBINARY_ENCODED NOT NULL,"
+ + " PK2 VARBINARY_ENCODED NOT NULL, COL1 VARBINARY_ENCODED"
+ + " CONSTRAINT pk PRIMARY KEY(PK1, PK2 DESC)) " + tableOptions);
+ }
+
+ static void upsertEncodedRow(Connection conn, String tableName, byte[] pk, byte[] col1)
+ throws SQLException {
+ upsertEncodedRow(conn, tableName, pk, col1, null, true);
+ }
+
+ static void upsertEncodedRow(Connection conn, String tableName, byte[] pk, byte[] col1,
+ boolean bindParameters) throws SQLException {
+ upsertEncodedRow(conn, tableName, pk, col1, null, bindParameters);
+ }
+
+ static void upsertEncodedRow(Connection conn, String tableName, byte[] pk, byte[] col1,
+ String col2, boolean bindParameters) throws SQLException {
+ if (!bindParameters) {
+ conn.createStatement()
+ .executeUpdate("UPSERT INTO " + tableName + " (PK1, COL1, COL2) VALUES (" + toLiteral(pk)
+ + ", " + toLiteral(col1) + ", " + (col2 == null ? "NULL" : "'" + col2 + "'") + ")");
+ return;
+ }
+ try (PreparedStatement preparedStatement =
+ conn.prepareStatement("UPSERT INTO " + tableName + " (PK1, COL1, COL2) VALUES (?, ?, ?)")) {
+ preparedStatement.setBytes(1, pk);
+ preparedStatement.setBytes(2, col1);
+ preparedStatement.setString(3, col2);
+ preparedStatement.executeUpdate();
+ }
+ }
+
+ static void upsertEncodedMultiPkRow(Connection conn, String tableName, byte[] pk1, byte[] pk2,
+ byte[] col1, boolean bindParameters) throws SQLException {
+ if (!bindParameters) {
+ conn.createStatement().executeUpdate("UPSERT INTO " + tableName + " (PK1, PK2, COL1) VALUES ("
+ + toLiteral(pk1) + ", " + toLiteral(pk2) + ", " + toLiteral(col1) + ")");
+ return;
+ }
+ try (PreparedStatement preparedStatement =
+ conn.prepareStatement("UPSERT INTO " + tableName + " (PK1, PK2, COL1) VALUES (?, ?, ?)")) {
+ preparedStatement.setBytes(1, pk1);
+ preparedStatement.setBytes(2, pk2);
+ preparedStatement.setBytes(3, col1);
+ preparedStatement.executeUpdate();
+ }
+ }
+
+ private static String toLiteral(byte[] value) {
+ return value == null ? "NULL" : PVarbinary.INSTANCE.toStringLiteral(value);
+ }
+
+ static void assertRows(Connection conn, String tableName, String columnList,
+ byte[][]... expectedRows) throws SQLException {
+ try (ResultSet resultSet =
+ conn.createStatement().executeQuery("SELECT /*+ NO_INDEX */ COUNT(*) FROM " + tableName)) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals(expectedRows.length, resultSet.getInt(1));
+ }
+
+ try (ResultSet resultSet = conn.createStatement()
+ .executeQuery("SELECT /*+ NO_INDEX */ " + columnList + " FROM " + tableName)) {
+ for (byte[][] expectedRow : expectedRows) {
+ Assert.assertTrue(resultSet.next());
+ for (int i = 0; i < expectedRow.length; i++) {
+ Assert.assertArrayEquals(expectedRow[i], resultSet.getBytes(i + 1));
+ }
+ }
+ Assert.assertFalse(resultSet.next());
+ }
+ }
+
+ static void assertRowsAndRowKeys(Connection conn, String tableName, byte[][]... expectedRows)
+ throws Exception {
+ assertRows(conn, tableName, SINGLE_PK_COLUMNS, expectedRows);
+
+ List<byte[]> actualRowKeys = new ArrayList<>();
+ try (
+ Table hTable =
+ conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(tableName));
+ ResultScanner scanner = hTable.getScanner(new Scan())) {
+ for (Result result : scanner) {
+ actualRowKeys.add(result.getRow());
+ }
+ }
+
+ byte[][] expectedRowKeys = new byte[expectedRows.length][];
+ for (int i = 0; i < expectedRows.length; i++) {
+ expectedRowKeys[i] = PVarbinaryEncoded.INSTANCE.toBytes(expectedRows[i][0]);
+ }
+ Assert.assertEquals(
+ Arrays.deepToString(expectedRowKeys) + " vs " + Arrays.deepToString(actualRowKeys.toArray()),
+ expectedRowKeys.length, actualRowKeys.size());
+ for (int i = 0; i < expectedRowKeys.length; i++) {
+ Assert.assertArrayEquals(expectedRowKeys[i], actualRowKeys.get(i));
+ }
+ }
+
+ static int countHBaseRows(Connection conn, String tableName) throws Exception {
+ int rowCount = 0;
+ try (
+ Table hTable =
+ conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(tableName));
+ ResultScanner scanner = hTable.getScanner(new Scan())) {
+ while (scanner.next() != null) {
+ rowCount++;
+ }
+ }
+ return rowCount;
+ }
+
+}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java b/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
index 41208464f6..fe438ff0f2 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
@@ -201,7 +201,7 @@ public abstract class BaseTest {
public static TemporaryFolder tmpFolder = new TemporaryFolder();
@ClassRule
public static Timeout CLASS_TIMEOUT =
- Timeout.builder().withTimeout(20, TimeUnit.MINUTES).withLookingForStuckThread(true).build();
+ Timeout.builder().withTimeout(50, TimeUnit.MINUTES).withLookingForStuckThread(true).build();
private static final int dropTableTimeout = 120; // 2 mins should be long enough.
private static final ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true)
.setNameFormat("DROP-TABLE-BASETEST" + "-thread-%s").build();
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/schema/types/PDataTypeTest.java b/phoenix-core/src/test/java/org/apache/phoenix/schema/types/PDataTypeTest.java
index 17a3cfdf02..40009dd8fc 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/schema/types/PDataTypeTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/schema/types/PDataTypeTest.java
@@ -18,6 +18,7 @@
package org.apache.phoenix.schema.types;
import static org.apache.phoenix.query.QueryConstants.MILLIS_IN_DAY;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -1998,4 +1999,35 @@ public class PDataTypeTest {
assertEquals(PVarchar.INSTANCE, PDataType.fromSqlTypeName("varchar"));
}
+ @Test
+ public void testVarbinaryEncodedIsNotInterchangeableWithPlainBinary() {
+ byte[] value = new byte[] { 0, -1, 5 };
+ assertFalse(Arrays.equals(value, PVarbinaryEncoded.INSTANCE.toBytes(value)));
+
+ assertFalse(PVarbinary.INSTANCE.isBytesComparableWith(PVarbinaryEncoded.INSTANCE));
+ assertFalse(PBinary.INSTANCE.isBytesComparableWith(PVarbinaryEncoded.INSTANCE));
+ assertFalse(PVarbinaryEncoded.INSTANCE.isBytesComparableWith(PVarbinary.INSTANCE));
+ assertFalse(PVarbinaryEncoded.INSTANCE.isBytesComparableWith(PBinary.INSTANCE));
+ assertTrue(PVarbinary.INSTANCE.isBytesComparableWith(PBinary.INSTANCE));
+ assertTrue(PBinary.INSTANCE.isBytesComparableWith(PVarbinary.INSTANCE));
+ assertTrue(PVarbinaryEncoded.INSTANCE.isBytesComparableWith(PVarbinaryEncoded.INSTANCE));
+
+ assertArrayEquals(value,
+ (byte[]) PVarbinary.INSTANCE.toObject(value, PVarbinaryEncoded.INSTANCE));
+ assertArrayEquals(value, (byte[]) PBinary.INSTANCE.toObject(value, PVarbinaryEncoded.INSTANCE));
+
+ for (PDataType<?> target : new PDataType<?>[] { PVarbinary.INSTANCE, PBinary.INSTANCE }) {
+ ImmutableBytesWritable ptr =
+ new ImmutableBytesWritable(PVarbinaryEncoded.INSTANCE.toBytes(value));
+ target.coerceBytes(ptr, null, PVarbinaryEncoded.INSTANCE, null, null, SortOrder.ASC, null,
+ null, SortOrder.ASC);
+ assertArrayEquals("coercing VARBINARY_ENCODED to " + target, value, ptr.copyBytes());
+ }
+
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable(PVarbinary.INSTANCE.toBytes(value));
+ PVarbinaryEncoded.INSTANCE.coerceBytes(ptr, null, PVarbinary.INSTANCE, null, null,
+ SortOrder.ASC, null, null, SortOrder.ASC);
+ assertArrayEquals(PVarbinaryEncoded.INSTANCE.toBytes(value), ptr.copyBytes());
+ }
+
}