Phoenix 6387 Conditional Updates

phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKeyIT.java

@@ -36,7 +36,11 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
+import org.apache.hadoop.hbase.TableName;
 import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.TestUtil;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -64,6 +68,12 @@ public static synchronized Collection<Object> data() {
  testCases.add(new String[] {
  "create local index %s_IDX on %s(counter1, counter2)",
  });
+ testCases.add(new String[] {
+ "create index %s_IDX on %s(counter1) include (counter2)",
+ });
+ testCases.add(new String[] {
+ "create index %s_IDX on %s(counter1, counter2)",
+ });
  return testCases;
  }
 
@@ -498,18 +508,29 @@ public void run() {
  exec.shutdownNow();
 
  int finalResult = nThreads * nCommits * nIncrementsPerCommit;
- //assertEquals(finalResult,resultHolder[0]);
- ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM " + tableName + " WHERE counter1 >= 0");
+ boolean isIndexCreated = this.indexDDL != null && this.indexDDL.length() > 0;
+
+ ResultSet rs;
+ String selectSql = "SELECT * FROM " + tableName + " WHERE counter1 >= 0";
+ if (isIndexCreated) {
+ rs = conn.createStatement().executeQuery("EXPLAIN " + selectSql);
+ String actualExplainPlan = QueryUtil.getExplainPlan(rs);
+ IndexToolIT.assertExplainPlan(this.indexDDL.contains("local"), actualExplainPlan,
+ tableName, tableName + "_IDX");
+ }
+ rs = conn.createStatement().executeQuery(selectSql);
  assertTrue(rs.next());
  assertEquals("a",rs.getString(1));
  assertEquals(finalResult,rs.getInt(2));
  assertFalse(rs.next());
 
- rs = conn.createStatement().executeQuery("SELECT /*+ NO_INDEX */ * FROM " + tableName + " WHERE counter1 >= 0");
- assertTrue(rs.next());
- assertEquals("a",rs.getString(1));
- assertEquals(finalResult,rs.getInt(2));
- assertFalse(rs.next());
+ if (isIndexCreated) {
+ rs = conn.createStatement().executeQuery("SELECT /*+ NO_INDEX */ * FROM " + tableName + " WHERE counter1 >= 0");
+ assertTrue(rs.next());
+ assertEquals("a", rs.getString(1));
+ assertEquals(finalResult, rs.getInt(2));
+ assertFalse(rs.next());
+ }
 
  conn.close();
  }
@@ -648,6 +669,82 @@ public void testOnDupAndUpsertInSameCommitBatch() throws Exception {
  }
  }
 
+ @Test
+ public void testMultiplePartialUpdatesInSameBatch() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ String tableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String ddl = "create table " + tableName + "(pk varchar primary key, counter1 bigint, counter2 bigint)";
+ conn.createStatement().execute(ddl);
+ createIndex(conn, tableName);
+ String dml;
+ ResultSet rs;
+ // first commit
+ dml = String.format("UPSERT INTO %s VALUES('a',0,0)", tableName);
+ conn.createStatement().execute(dml);
+ conn.commit();
+ // batch multiple conditional updates (partial) in a single batch
+ dml = String.format(
+ "UPSERT INTO %s VALUES('a',2,3) ON DUPLICATE KEY UPDATE counter1 = counter1 + 1", tableName);
+ conn.createStatement().execute(dml);
+ dml = String.format(
+ "UPSERT INTO %s VALUES('a',2,3) ON DUPLICATE KEY UPDATE counter2 = counter2 + 2", tableName);
+ conn.createStatement().execute(dml);
+ dml = String.format(
+ "UPSERT INTO %s VALUES('a',2,3) ON DUPLICATE KEY UPDATE counter1 = counter1 + 100", tableName);
+ conn.createStatement().execute(dml);
+ dml = String.format(
+ "UPSERT INTO %s VALUES('a',2,3) ON DUPLICATE KEY UPDATE counter2 = counter2 + 200", tableName);
+ conn.createStatement().execute(dml);
+ conn.commit();
+ String dql = String.format("SELECT counter1, counter2 FROM %s WHERE counter1 > 0", tableName);
+ rs = conn.createStatement().executeQuery(dql);
+ assertTrue(rs.next());
+ assertEquals(101, rs.getInt(1));
+ assertEquals(202, rs.getInt(2));
+ }
+ }
+
+ @Test
+ public void testComplexDuplicateKeyExpression() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ String tableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String ddl = "create table " + tableName +
+ "(pk varchar primary key, counter1 bigint, counter2 bigint, approval varchar)";
+ conn.createStatement().execute(ddl);
+ createIndex(conn, tableName);
+ String dml;
+ dml = String.format("UPSERT INTO %s VALUES('abc', 0, 100, 'NONE')", tableName);
+ conn.createStatement().execute(dml);
+ conn.commit();
+ dml = String.format("UPSERT INTO %s(pk, counter1, counter2) VALUES ('abc', 0, 10) " +
+ "ON DUPLICATE KEY UPDATE " +
+ "counter1 = counter1 + counter2," +
+ "approval = CASE WHEN counter1 < 100 THEN 'NONE' " +
+ "WHEN counter1 < 1000 THEN 'MANAGER_APPROVAL' " +
+ "ELSE 'VP_APPROVAL' END", tableName);
+ conn.createStatement().execute(dml);
+ conn.commit();
+ String dql = "SELECT * from " + tableName;
+ ResultSet rs = conn.createStatement().executeQuery(dql);
+ assertTrue(rs.next());
+ assertEquals("abc", rs.getString("pk"));
+ assertEquals(100, rs.getInt("counter1"));
+ assertEquals(100, rs.getInt("counter2"));
+ assertEquals("NONE", rs.getString("approval"));
+
+ conn.createStatement().execute(dml);
+ conn.commit();
+ rs = conn.createStatement().executeQuery(dql);
+ assertTrue(rs.next());
+ assertEquals("abc", rs.getString("pk"));
+ assertEquals(200, rs.getInt("counter1"));
+ assertEquals(100, rs.getInt("counter2"));
+ assertEquals("MANAGER_APPROVAL", rs.getString("approval"));
+ }
+ }
+
  private void assertRow(Connection conn, String tableName, String expectedPK, int expectedCol1, String expectedCol2) throws SQLException {
  ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM " + tableName);
  assertTrue(rs.next());

phoenix-core/src/it/java/org/apache/phoenix/end2end/WALAnnotationIT.java

@@ -456,6 +456,40 @@ public void testTenantViewUpsertWithIndex() throws Exception {
  tenantViewHelper(true);
  }
 
+ @Test
+ public void testOnDuplicateUpsertWithIndex() throws Exception {
+ Assume.assumeFalse(this.isImmutable); // on duplicate is not supported for immutable tables
+ Assume.assumeTrue(HbaseCompatCapabilities.hasPreWALAppend());
+ SchemaBuilder builder = new SchemaBuilder(getUrl());
+ try (Connection conn = getConnection()) {
+ SchemaBuilder.TableOptions tableOptions = getTableOptions();
+ builder.withTableOptions(tableOptions).withTableIndexDefaults().build();
+ PTable table = PhoenixRuntime.getTableNoCache(conn, builder.getEntityTableName());
+ assertEquals("Change Detection Enabled is false!", true, table.isChangeDetectionEnabled());
+ Long ddlTimestamp = table.getLastDDLTimestamp();
+ String upsertSql = "UPSERT INTO " + builder.getEntityTableName() + " VALUES" +
+ " ('a', 'b', 'c', 'd')";
+ conn.createStatement().execute(upsertSql);
+ conn.commit();
+ List<String> columns = builder.getTableOptions().getTableColumns();
+ assertTrue(columns.size() >= 2);
+ String col1 = columns.get(0);
+ String col2 = columns.get(1);
+ // col1 = col1 || col1, col2 = null
+ String onDupClause = String.format("%s = %s || %s, %s = null", col1, col1, col1, col2);
+ // this will result in one Put and one Delete (because of null) mutation
+ upsertSql = "UPSERT INTO " + builder.getEntityTableName() + " VALUES" +
+ " ('a', 'b', 'c', 'd') ON DUPLICATE KEY UPDATE " + onDupClause;
+ conn.createStatement().execute(upsertSql);
+ conn.commit();
+ assertAnnotation(2, builder.getPhysicalTableName(false), null,
+ builder.getTableOptions().getSchemaName(),
+ builder.getDataOptions().getTableName(), PTableType.TABLE, ddlTimestamp);
+ assertAnnotation(0, builder.getPhysicalTableIndexName(false),
+ null, null, null, null, ddlTimestamp);
+ }
+ }
+
  private List<Map<String, byte[]>> getEntriesForTable(TableName tableName) throws IOException {
  AnnotatedWALObserver c = getTestCoprocessor(tableName);
  List<Map<String, byte[]>> entries = c.getWalAnnotationsByTable(tableName);

phoenix-core/src/it/java/org/apache/phoenix/end2end/index/GlobalIndexCheckerIT.java

@@ -1022,6 +1022,34 @@ public void testViewIndexRowUpdate() throws Exception {
  }
  }
 
+ @Test
+ public void testOnDuplicateKeyWithIndex() throws Exception {
+ if (async || encoded) { // run only once with single cell encoding enabled
+ return;
+ }
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ String dataTableName = generateUniqueName();
+ String indexTableName = generateUniqueName();
+ populateTable(dataTableName); // with two rows ('a', 'ab', 'abc', 'abcd') and ('b', 'bc', 'bcd', 'bcde')
+ conn.createStatement().execute("CREATE INDEX " + indexTableName + " on " +
+ dataTableName + " (val1) include (val2, val3)" + this.indexDDLOptions);
+ conn.commit();
+ String upsertSql = "UPSERT INTO " + dataTableName + " VALUES ('a') ON DUPLICATE KEY UPDATE " +
+ "val1 = val1 || val1, val2 = val2 || val2";
+ conn.createStatement().execute(upsertSql);
+ conn.commit();
+ String selectSql = "SELECT * from " + dataTableName + " WHERE val1 = 'abab'";
+ assertExplainPlan(conn, selectSql, dataTableName, indexTableName);
+ ResultSet rs = conn.createStatement().executeQuery(selectSql);
+ assertTrue(rs.next());
+ assertEquals("a", rs.getString(1));
+ assertEquals("abab", rs.getString(2));
+ assertEquals("abcabc", rs.getString(3));
+ assertEquals("abcd", rs.getString(4));
+ assertFalse(rs.next());
+ }
+ }
+
  static private void commitWithException(Connection conn) {
  try {
  conn.commit();

phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java

@@ -113,7 +113,7 @@ static void assertMutationMetrics(String tableName, int numRows, boolean isUpser
  String t = entry.getKey();
  assertEquals("Table names didn't match!", tableName, t);
  Map<MetricType, Long> p = entry.getValue();
- assertEquals("There should have been fifteen metrics", 15, p.size());
+ assertEquals("There should have been sixteen metrics", 16, p.size());
  boolean mutationBatchSizePresent = false;
  boolean mutationCommitTimePresent = false;
  boolean mutationBytesPresent = false;

phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java

@@ -487,7 +487,7 @@ public void testMetricsForUpsert() throws Exception {
  String t = entry.getKey();
  assertEquals("Table names didn't match!", tableName, t);
  Map<MetricType, Long> p = entry.getValue();
- assertEquals("There should have been five metrics", 15, p.size());
+ assertEquals("There should have been sixteen metrics", 16, p.size());
  boolean mutationBatchSizePresent = false;
  boolean mutationCommitTimePresent = false;
  boolean mutationBytesPresent = false;

phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java

@@ -56,6 +56,8 @@
 import static org.apache.phoenix.exception.SQLExceptionCode.DATA_EXCEEDS_MAX_CAPACITY;
 import static org.apache.phoenix.exception.SQLExceptionCode.GET_TABLE_REGIONS_FAIL;
 import static org.apache.phoenix.exception.SQLExceptionCode.OPERATION_TIMED_OUT;
+import static org.apache.phoenix.monitoring.MetricType.ATOMIC_UPSERT_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.ATOMIC_UPSERT_SQL_COUNTER;
 import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_FAILURE_SQL_COUNTER;
 import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_SUCCESS_SQL_COUNTER;
 import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_COUNTER;
@@ -1148,6 +1150,50 @@ private static void assertMetricValue(Metric m, MetricType checkType, long compa
  }
  }
 
+ @Test public void testTableLevelMetricsForAtomicUpserts() throws Throwable {
+ String tableName = generateUniqueName();
+ Connection conn = null;
+ Throwable exception = null;
+ int numAtomicUpserts = 4;
+ try {
+ conn = getConnFromTestDriver();
+ String ddl = "create table " + tableName + "(pk varchar primary key, counter1 bigint)";
+ conn.createStatement().execute(ddl);
+ String dml;
+ ResultSet rs;
+ dml = String.format("UPSERT INTO %s VALUES('a', 0)", tableName);
+ conn.createStatement().execute(dml);
+ dml = String.format("UPSERT INTO %s VALUES('a', 0) ON DUPLICATE KEY UPDATE counter1 = counter1 + 1", tableName);
+ for (int i = 0; i < numAtomicUpserts; ++i) {
+ conn.createStatement().execute(dml);
+ }
+ conn.commit();
+ String dql = String.format("SELECT counter1 FROM %s WHERE counter1 > 0", tableName);
+ rs = conn.createStatement().executeQuery(dql);
+ assertTrue(rs.next());
+ assertEquals(4, rs.getInt(1));
+ }catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ // Get write metrics before closing the connection since that clears those metrics
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ // 1 regular upsert + numAtomicUpserts
+ // 2 mutations (regular and atomic on the same row in the same batch will be split)
+ assertMutationTableMetrics(true, tableName, 1 + numAtomicUpserts, 0, 0, true, 2, 0, 0, 2, 0,
+ writeMutMetrics, conn);
+ assertEquals(numAtomicUpserts, getMetricFromTableMetrics(tableName, ATOMIC_UPSERT_SQL_COUNTER));
+ assertTrue(getMetricFromTableMetrics(tableName, ATOMIC_UPSERT_COMMIT_TIME) > 0);
+ }
+ }
+
  private Connection getConnFromTestDriver() throws SQLException {
  Connection conn = DriverManager.getConnection(url);
  assertTrue(conn.unwrap(PhoenixConnection.class)

phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java

@@ -864,12 +864,6 @@ public MutationPlan compile(UpsertStatement upsert) throws SQLException {
  .setTableName(table.getTableName().getString())
  .build().buildException();
  }
- if (SchemaUtil.hasGlobalIndex(table)) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_USE_ON_DUP_KEY_WITH_GLOBAL_IDX)
- .setSchemaName(table.getSchemaName().getString())
- .setTableName(table.getTableName().getString())
- .build().buildException();
- }
  if (onDupKeyPairs.isEmpty()) { // ON DUPLICATE KEY IGNORE
  onDupKeyBytesToBe = PhoenixIndexBuilder.serializeOnDupKeyIgnore();
  } else { // ON DUPLICATE KEY UPDATE;

phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java

@@ -256,26 +256,30 @@ public SimpleValueGetter (final Put put) {
  }
  @Override
  public ImmutableBytesWritable getLatestValue(ColumnReference ref, long ts) {
-List<Cell> cellList = put.get(ref.getFamily(), ref.getQualifier());
- if (cellList == null || cellList.isEmpty()) {
+ Cell cell = getLatestCell(ref, ts);
+ if (cell == null) {
  return null;
  }
- Cell cell = cellList.get(0);
  valuePtr.set(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  return valuePtr;
  }
- @Override
- public KeyValue getLatestKeyValue(ColumnReference ref, long ts) {
+ public Cell getLatestCell(ColumnReference ref, long ts) {
  List<Cell> cellList = put.get(ref.getFamily(), ref.getQualifier());
  if (cellList == null || cellList.isEmpty()) {
  return null;
  }
- Cell cell = cellList.get(0);
- return new KeyValue(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
+ return cellList.get(0);
+ }
+ @Override
+ public KeyValue getLatestKeyValue(ColumnReference ref, long ts) {
+ Cell cell = getLatestCell(ref, ts);
+ KeyValue kv = cell == null ? null :
+ new KeyValue(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
  cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
  cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
  cell.getTimestamp(), KeyValue.Type.codeToType(cell.getTypeByte()),
  cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+ return kv;
  }
  @Override
  public byte[] getRowKey() {