From 49d6929ea04949d31cd82328fee6bcee16284e73 Mon Sep 17 00:00:00 2001 From: Tanuj Khurana Date: Fri, 10 Sep 2021 14:15:38 -0700 Subject: [PATCH 1/2] PHOENIX-6541 Use ROW_TIMESTAMP column value as timestamps for conditional upsert mutations --- .../phoenix/end2end/OnDuplicateKeyIT.java | 91 ++++++++++++++++++- .../hbase/index/IndexRegionObserver.java | 4 +- 2 files changed, 91 insertions(+), 4 deletions(-) diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKeyIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKeyIT.java index 0878c07f480..1d3567ca249 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKeyIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKeyIT.java @@ -26,6 +26,7 @@ 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.Collection; @@ -36,11 +37,9 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; -import org.apache.hadoop.hbase.TableName; +import org.apache.phoenix.util.EnvironmentEdgeManager; 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; @@ -745,6 +744,70 @@ public void testComplexDuplicateKeyExpression() throws Exception { } } + @Test + public void testRowStampCol() throws Exception { + // ROW_TIMESTAMP is not supported for tables with indexes + if (indexDDL.length() > 0) { + return; + } + Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); + String tableName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl(), props)) { + String ddl = "create table " + tableName + + "(\n" + + "ORGANIZATION_ID CHAR(15) NOT NULL,\n" + + "USER_ID CHAR(15) NOT NULL,\n" + + "TIME_STAMP DATE NOT NULL,\n" + + "STATUS VARCHAR,\n" + + "CONSTRAINT PK PRIMARY KEY \n" + + " (\n" + + " ORGANIZATION_ID, \n" + + " USER_ID,\n" + + " TIME_STAMP ROW_TIMESTAMP\n" + // ROW_TIMESTAMP col + " ) \n" + + ")\n"; + + conn.createStatement().execute(ddl); + String orgid = "ORG1"; + String userid = "USER1"; + String original = "ORIGINAL"; + String updated = "UPDATED"; + String duplicate = "DUPLICATE"; + long rowTimestamp = EnvironmentEdgeManager.currentTimeMillis() - 10; + String dml = "UPSERT INTO " + tableName + + "(ORGANIZATION_ID, USER_ID, TIME_STAMP, STATUS) VALUES (?, ?, ?, ?)"; + String ignoreDml = dml + "ON DUPLICATE KEY IGNORE"; + String updateDml = dml + "ON DUPLICATE KEY UPDATE status='" + duplicate + "'"; + String nullDml = dml + "ON DUPLICATE KEY UPDATE status = null"; + String dql = "SELECT count(*) from " + tableName + " WHERE STATUS = ?"; + + // row doesn't exist + upsertRecord(conn, ignoreDml, orgid, userid, rowTimestamp, original); + assertNumRecords(1, conn, dql, original); + + // on duplicate key ignore + upsertRecord(conn, ignoreDml, orgid, userid, rowTimestamp, updated); + assertNumRecords(1, conn, dql, original); + assertNumRecords(0, conn, dql, updated); + + // regular upsert override + upsertRecord(conn, dml, orgid, userid, rowTimestamp, updated); + assertNumRecords(0, conn, dql, original); + assertNumRecords(1, conn, dql, updated); + + // on duplicate key update + upsertRecord(conn, updateDml, orgid, userid, rowTimestamp, ""); + assertNumRecords(0, conn, dql, updated); + assertNumRecords(1, conn, dql, duplicate); + + // set null + upsertRecord(conn, nullDml, orgid, userid, rowTimestamp, ""); + assertNumRecords(0, conn, dql, duplicate); + dql = "SELECT count(*) from " + tableName + " WHERE STATUS is null"; + assertNumRecords(1, conn, dql); + } + } + 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()); @@ -754,6 +817,28 @@ private void assertRow(Connection conn, String tableName, String expectedPK, int assertFalse(rs.next()); } + private void upsertRecord(Connection conn, String dml, String orgid, String userid, long ts, String status) throws SQLException { + try(PreparedStatement stmt = conn.prepareStatement(dml)) { // regular upsert + stmt.setString(1, orgid); + stmt.setString(2, userid); + stmt.setDate(3, new Date(ts)); + stmt.setString(4, status); // status should change now + stmt.executeUpdate(); + conn.commit(); + } + } + + private void assertNumRecords(int count, Connection conn, String dql, String... params) + throws Exception { + PreparedStatement stmt = conn.prepareStatement(dql); + int counter = 1; + for (String param : params) { + stmt.setString(counter++, param); + } + ResultSet rs = stmt.executeQuery(); + assertTrue(rs.next()); + assertEquals(count, rs.getInt(1)); + } } diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java index 96b4f18c493..526c048321a 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java @@ -610,7 +610,9 @@ public static void setTimestamps(MiniBatchOperationInProgress miniBatc } Mutation m = miniBatchOp.getOperation(i); // skip this mutation if we aren't enabling indexing or not an atomic op - if (!builder.isEnabled(m) && !builder.isAtomicOp(m)) { + // or if it is an atomic op and its timestamp is already set + if (!builder.isEnabled(m) && + (!builder.isAtomicOp(m) || getMaxTimestamp(m) != HConstants.LATEST_TIMESTAMP)) { continue; } setTimestampOnMutation(m, ts); From cbb9efd56f0e893d02ed85a4dc1926d17861cd37 Mon Sep 17 00:00:00 2001 From: Tanuj Khurana Date: Fri, 10 Sep 2021 16:50:46 -0700 Subject: [PATCH 2/2] Address review comments --- .../phoenix/end2end/OnDuplicateKeyIT.java | 29 +++++++++++++++++-- .../hbase/index/IndexRegionObserver.java | 4 +-- 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKeyIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKeyIT.java index 1d3567ca249..532a28f015a 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKeyIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKeyIT.java @@ -37,6 +37,14 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.ConnectionFactory; +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.phoenix.query.QueryConstants; +import org.apache.phoenix.util.EncodedColumnsUtil; import org.apache.phoenix.util.EnvironmentEdgeManager; import org.apache.phoenix.util.PropertiesUtil; import org.apache.phoenix.util.QueryUtil; @@ -784,23 +792,27 @@ public void testRowStampCol() throws Exception { // row doesn't exist upsertRecord(conn, ignoreDml, orgid, userid, rowTimestamp, original); assertNumRecords(1, conn, dql, original); + assertHBaseRowTimestamp(tableName, rowTimestamp); // on duplicate key ignore upsertRecord(conn, ignoreDml, orgid, userid, rowTimestamp, updated); assertNumRecords(1, conn, dql, original); assertNumRecords(0, conn, dql, updated); + assertHBaseRowTimestamp(tableName, rowTimestamp); // regular upsert override upsertRecord(conn, dml, orgid, userid, rowTimestamp, updated); assertNumRecords(0, conn, dql, original); assertNumRecords(1, conn, dql, updated); + assertHBaseRowTimestamp(tableName, rowTimestamp); - // on duplicate key update + // on duplicate key update generates extra mutations on the server but those mutations + // don't honor ROW_TIMESTAMP upsertRecord(conn, updateDml, orgid, userid, rowTimestamp, ""); assertNumRecords(0, conn, dql, updated); assertNumRecords(1, conn, dql, duplicate); - // set null + // set null, new mutations generated on the server upsertRecord(conn, nullDml, orgid, userid, rowTimestamp, ""); assertNumRecords(0, conn, dql, duplicate); dql = "SELECT count(*) from " + tableName + " WHERE STATUS is null"; @@ -840,5 +852,18 @@ private void assertNumRecords(int count, Connection conn, String dql, String... assertEquals(count, rs.getInt(1)); } + private void assertHBaseRowTimestamp(String tableName, long expectedTimestamp) throws Exception { + Scan scan = new Scan(); + byte[] emptyKVQualifier = EncodedColumnsUtil.getEmptyKeyValueInfo(true).getFirst(); + try (org.apache.hadoop.hbase.client.Connection hconn = + ConnectionFactory.createConnection(config)) { + Table table = hconn.getTable(TableName.valueOf(tableName)); + ResultScanner resultScanner = table.getScanner(scan); + Result result = resultScanner.next(); + long actualTimestamp = result.getColumnLatestCell( + QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES, emptyKVQualifier).getTimestamp(); + assertEquals(expectedTimestamp, actualTimestamp); + } + } } diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java index 526c048321a..8c40def4403 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java @@ -610,9 +610,9 @@ public static void setTimestamps(MiniBatchOperationInProgress miniBatc } Mutation m = miniBatchOp.getOperation(i); // skip this mutation if we aren't enabling indexing or not an atomic op - // or if it is an atomic op and its timestamp is already set + // or if it is an atomic op and its timestamp is already set(not LATEST) if (!builder.isEnabled(m) && - (!builder.isAtomicOp(m) || getMaxTimestamp(m) != HConstants.LATEST_TIMESTAMP)) { + !(builder.isAtomicOp(m) && getMaxTimestamp(m) == HConstants.LATEST_TIMESTAMP)) { continue; } setTimestampOnMutation(m, ts);