Repository: phoenix Updated Branches: refs/heads/master f0881a137 -> 1c042c25e
PHOENIX-4983: Allow using a connection with a SCN set to write data to tables EXCEPT transactional tables or mutable tables with indexes or tables with ROW_TIMESTAMP column. Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/1c042c25 Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/1c042c25 Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/1c042c25 Branch: refs/heads/master Commit: 1c042c25ed833945abe07958b82d5e9fb701ac89 Parents: f0881a1 Author: s.kadam <s.ka...@salesforce.com> Authored: Mon Dec 10 14:40:17 2018 -0800 Committer: Thomas D'Silva <tdsi...@apache.org> Committed: Tue Dec 11 17:04:53 2018 -0800 ---------------------------------------------------------------------- .../apache/phoenix/end2end/UpsertWithSCNIT.java | 139 +++++++++++++++++++ .../apache/phoenix/compile/UpsertCompiler.java | 23 ++- .../phoenix/exception/SQLExceptionCode.java | 13 +- .../apache/phoenix/jdbc/PhoenixConnection.java | 2 +- 4 files changed, 172 insertions(+), 5 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/phoenix/blob/1c042c25/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertWithSCNIT.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertWithSCNIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertWithSCNIT.java new file mode 100644 index 0000000..6f231ff --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertWithSCNIT.java @@ -0,0 +1,139 @@ +package org.apache.phoenix.end2end; + +import org.apache.phoenix.exception.SQLExceptionCode; +import org.apache.phoenix.exception.SQLExceptionInfo; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; + +import java.sql.Connection; +import java.sql.Date; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.Properties; + +import static org.hamcrest.CoreMatchers.containsString; +import static org.hamcrest.core.Is.is; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + + +public class UpsertWithSCNIT extends ParallelStatsDisabledIT { + + @Rule + public final ExpectedException exception = ExpectedException.none(); + Properties props = null; + PreparedStatement prep = null; + String tableName =null; + + private void helpTestUpserWithSCNIT(boolean rowColumn, boolean txTable, + boolean mutable, boolean local, boolean global) + throws SQLException { + + tableName = generateUniqueName(); + String indx; + String createTable = "CREATE TABLE "+tableName+" (" + + (rowColumn ? "CREATED_DATE DATE NOT NULL, ":"") + + "METRIC_ID CHAR(15) NOT NULL,METRIC_VALUE VARCHAR(50) CONSTRAINT PK PRIMARY KEY(" + + (rowColumn? "CREATED_DATE ROW_TIMESTAMP, ":"") + "METRIC_ID)) " + + (mutable? "IMMUTABLE_ROWS=false":"" ) + + (txTable ? "TRANSACTION_PROVIDER='TEPHRA',TRANSACTIONAL=true":""); + props = new Properties(); + Connection conn = DriverManager.getConnection(getUrl(), props); + conn.createStatement().execute(createTable); + + if(local || global ){ + indx = "CREATE "+ (local? "LOCAL " : "") + "INDEX "+tableName+"_idx ON " + + ""+tableName+" (METRIC_VALUE)"; + conn.createStatement().execute(indx); + } + + props.setProperty("CurrentSCN", Long.toString(System.currentTimeMillis())); + conn = DriverManager.getConnection(getUrl(), props); + conn.setAutoCommit(true); + String upsert = "UPSERT INTO "+tableName+" (METRIC_ID, METRIC_VALUE) VALUES (?,?)"; + prep = conn.prepareStatement(upsert); + prep.setString(1,"abc"); + prep.setString(2,"This is the first comment!"); + } + + @Test // See https://issues.apache.org/jira/browse/PHOENIX-4983 + public void testUpsertOnSCNSetTxnTable() throws SQLException { + + helpTestUpserWithSCNIT(false, true, false, false, false); + exception.expect(SQLException.class); + exception.expectMessage(containsString(String.valueOf( + SQLExceptionCode + .CANNOT_SPECIFY_SCN_FOR_TXN_TABLE + .getErrorCode()))); + prep.executeUpdate(); + } + + @Test + public void testUpsertOnSCNSetMutTableWithoutIdx() throws Exception { + + helpTestUpserWithSCNIT(false, false, true, false, false); + prep.executeUpdate(); + props = new Properties(); + Connection conn = DriverManager.getConnection(getUrl(),props); + ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM "+tableName); + assertTrue(rs.next()); + assertEquals("abc", rs.getString(1)); + assertEquals("This is the first comment!", rs.getString(2)); + assertFalse(rs.next()); + } + + @Test + public void testUpsertOnSCNSetTable() throws Exception { + + helpTestUpserWithSCNIT(false, false, false, false, false); + prep.executeUpdate(); + props = new Properties(); + Connection conn = DriverManager.getConnection(getUrl(),props); + ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM "+tableName); + assertTrue(rs.next()); + assertEquals("abc", rs.getString(1)); + assertEquals("This is the first comment!", rs.getString(2)); + assertFalse(rs.next()); + } + + @Test + public void testUpsertOnSCNSetMutTableWithLocalIdx() throws Exception { + + helpTestUpserWithSCNIT(false, false, true, true, false); + exception.expect(SQLException.class); + exception.expectMessage(containsString(String.valueOf( + SQLExceptionCode + .CANNOT_UPSERT_WITH_SCN_FOR_MUTABLE_TABLE_WITH_INDEXES + .getErrorCode()))); + prep.executeUpdate(); + } + @Test + public void testUpsertOnSCNSetMutTableWithGlobalIdx() throws Exception { + + helpTestUpserWithSCNIT(false, false, true, false, true); + exception.expect(SQLException.class); + exception.expectMessage(containsString(String.valueOf( + SQLExceptionCode + .CANNOT_UPSERT_WITH_SCN_FOR_MUTABLE_TABLE_WITH_INDEXES + .getErrorCode()))); + prep.executeUpdate(); + + } + @Test + public void testUpsertOnSCNSetWithRowTSColumn() throws Exception { + + helpTestUpserWithSCNIT(true, false, false, false, false); + exception.expect(SQLException.class); + exception.expectMessage(containsString(String.valueOf( + SQLExceptionCode + .CANNOT_UPSERT_WITH_SCN_FOR_ROW_TIMSTAMP_COLUMN + .getErrorCode()))); + prep.executeUpdate(); + } +} + http://git-wip-us.apache.org/repos/asf/phoenix/blob/1c042c25/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java index ed07373..7ec96a3 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java @@ -366,6 +366,8 @@ public class UpsertCompiler { // Cannot update: // - read-only VIEW // - transactional table with a connection having an SCN + // - mutable table with indexes and SCN set + // - tables with ROW_TIMESTAMP columns if (table.getType() == PTableType.VIEW && table.getViewType().isReadOnly()) { throw new ReadOnlyTableException(schemaName,tableName); } else if (connection.isBuildingIndex() && table.getType() != PTableType.INDEX) { @@ -374,8 +376,25 @@ public class UpsertCompiler { .setTableName(tableName) .build().buildException(); } else if (table.isTransactional() && connection.getSCN() != null) { - throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_SPECIFY_SCN_FOR_TXN_TABLE).setSchemaName(schemaName) - .setTableName(tableName).build().buildException(); + throw new SQLExceptionInfo.Builder(SQLExceptionCode + .CANNOT_SPECIFY_SCN_FOR_TXN_TABLE) + .setSchemaName(schemaName) + .setTableName(tableName).build().buildException(); + } else if (!table.isImmutableRows() && connection.getSCN() != null + && !table.getIndexes().isEmpty() && !connection.isRunningUpgrade() + && !connection.isBuildingIndex()) { + throw new SQLExceptionInfo + .Builder(SQLExceptionCode + .CANNOT_UPSERT_WITH_SCN_FOR_MUTABLE_TABLE_WITH_INDEXES) + .setSchemaName(schemaName) + .setTableName(tableName).build().buildException(); + } else if(connection.getSCN() != null && !connection.isRunningUpgrade() + && !connection.isBuildingIndex() && table.getRowTimestampColPos() >= 0) { + throw new SQLExceptionInfo + .Builder(SQLExceptionCode + .CANNOT_UPSERT_WITH_SCN_FOR_ROW_TIMSTAMP_COLUMN) + .setSchemaName(schemaName) + .setTableName(tableName).build().buildException(); } boolean isSalted = table.getBucketNum() != null; isTenantSpecific = table.isMultiTenant() && connection.getTenantId() != null; http://git-wip-us.apache.org/repos/asf/phoenix/blob/1c042c25/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java index d29c90f..2c52d44 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java @@ -289,7 +289,8 @@ public enum SQLExceptionCode { TRANSACTION_FAILED(1077, "44A08", "Transaction Failure "), CANNOT_CREATE_TXN_TABLE_IF_TXNS_DISABLED(1078, "44A09", "Cannot create a transactional table if transactions are disabled."), CANNOT_ALTER_TO_BE_TXN_IF_TXNS_DISABLED(1079, "44A10", "Cannot alter table to be transactional table if transactions are disabled."), - CANNOT_CREATE_TXN_TABLE_WITH_ROW_TIMESTAMP(1080, "44A11", "Cannot create a transactional table if transactions are disabled."), + CANNOT_CREATE_TXN_TABLE_WITH_ROW_TIMESTAMP(1080, "44A11", "Cannot create a transactional" + + " table with ROW_TIMESTAMP column."), CANNOT_ALTER_TO_BE_TXN_WITH_ROW_TIMESTAMP(1081, "44A12", "Cannot alter table to be transactional table if transactions are disabled."), TX_MUST_BE_ENABLED_TO_SET_TX_CONTEXT(1082, "44A13", "Cannot set transaction context if transactions are disabled."), TX_MUST_BE_ENABLED_TO_SET_AUTO_FLUSH(1083, "44A14", "Cannot set auto flush if transactions are disabled."), @@ -461,7 +462,15 @@ public enum SQLExceptionCode { MAX_MUTATION_SIZE_EXCEEDED(729, "LIM01", "MutationState size is bigger than maximum allowed number of rows"), MAX_MUTATION_SIZE_BYTES_EXCEEDED(730, "LIM02", "MutationState size is bigger than maximum allowed number of bytes"), INSUFFICIENT_MEMORY(999, "50M01", "Unable to allocate enough memory."), - HASH_JOIN_CACHE_NOT_FOUND(900, "HJ01", "Hash Join cache not found"); + HASH_JOIN_CACHE_NOT_FOUND(900, "HJ01", "Hash Join cache not found"), + + CANNOT_UPSERT_WITH_SCN_FOR_ROW_TIMSTAMP_COLUMN(901,"43M12", + "Cannot use a connection with SCN set to upsert data for " + + "table with ROW_TIMESTAMP column."), + CANNOT_UPSERT_WITH_SCN_FOR_MUTABLE_TABLE_WITH_INDEXES(903,"43M14", + "Cannot use a connection with SCN set to " + + "upsert data for a mutable table with indexes."); + private final int errorCode; private final String sqlState; http://git-wip-us.apache.org/repos/asf/phoenix/blob/1c042c25/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java index 6da579f..596e27c 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java @@ -858,7 +858,7 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea @Override public boolean isReadOnly() throws SQLException { - return readOnly || (scn != null && !buildingIndex && !isRunningUpgrade); + return readOnly; } @Override
