Skip to content

PHOENIX-6541 Use ROW_TIMESTAMP column value as timestamps for conditional upsert mutations#1300

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gjacoby126 merged 2 commits into from
Sep 14, 2021
Merged

PHOENIX-6541 Use ROW_TIMESTAMP column value as timestamps for conditional upsert mutations #1300

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
49d6929
PHOENIX-6541 Use ROW_TIMESTAMP column value as timestamps for conditi…
tkhurana Sep 10, 2021
cbb9efd
Address review comments
tkhurana Sep 10, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 112 additions & 2 deletions phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKeyIT.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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;
Expand All@@ -37,10 +38,16 @@
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;
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;
Expand DownExpand Up@@ -745,6 +752,74 @@ 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;

@ankitjain64ankitjain64Sep 10, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Curious is there a particular reason to subtract 10 ms from the rowTimestamp. ?

@tkhuranatkhuranaSep 13, 2021•
edited
Loading

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ankitjain64 We just wanted a value that is less than the current time. 10 is chosen arbitrarily.

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);
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 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, 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";
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());
Expand All@@ -754,6 +829,41 @@ 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));

@ankitjain64ankitjain64Sep 10, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

from the jira description looks like we want to make sure correct timestamps are used for the mutation but we are not validating that in the test case and just the row count. Is that sufficient @tkhurana ?

@tkhuranatkhuranaSep 13, 2021

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ankitjain64 I was able to repro the problem with just checking row count but I will add the asserts for timestamps also

}

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);
}
}
}

4 changes: 3 additions & 1 deletion phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -610,7 +610,9 @@ public static void setTimestamps(MiniBatchOperationInProgress<Mutation> 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(not LATEST)
if (!builder.isEnabled(m) &&
!(builder.isAtomicOp(m) && getMaxTimestamp(m) == HConstants.LATEST_TIMESTAMP)) {
continue;
}
setTimestampOnMutation(m, ts);
Expand Down