This is an automated email from the ASF dual-hosted git repository.
vjasani 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 9f31e3e1a9 PHOENIX-7411 Atomic Delete: PhoenixStatement API to return
row if single row is atomically deleted (#1985)
9f31e3e1a9 is described below
commit 9f31e3e1a9ae81a13662805bdfb2f7aba872e808
Author: Viraj Jasani <[email protected]>
AuthorDate: Wed Oct 9 09:57:39 2024 -0700
PHOENIX-7411 Atomic Delete: PhoenixStatement API to return row if single
row is atomically deleted (#1985)
---
.../org/apache/phoenix/execute/MutationState.java | 12 ++
.../org/apache/phoenix/jdbc/PhoenixStatement.java | 16 +-
.../phoenix/hbase/index/IndexRegionObserver.java | 19 ++-
.../apache/phoenix/end2end/OnDuplicateKey2IT.java | 180 +++++++++++++++++++--
4 files changed, 209 insertions(+), 18 deletions(-)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/execute/MutationState.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/execute/MutationState.java
index 419b273c72..6d3c9df480 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/execute/MutationState.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/execute/MutationState.java
@@ -495,6 +495,10 @@ public class MutationState implements SQLCloseable {
return this.result;
}
+ public void clearResult() {
+ this.result = null;
+ }
+
private MultiRowMutationState getLastMutationBatch(Map<TableRef,
List<MultiRowMutationState>> mutations, TableRef tableRef) {
List<MultiRowMutationState> mutationBatches = mutations.get(tableRef);
if (mutationBatches == null || mutationBatches.isEmpty()) {
@@ -841,6 +845,14 @@ public class MutationState implements SQLCloseable {
mutation.setAttribute(SOURCE_OPERATION_ATTRIB,
sourceOfDeleteBytes);
}
}
+ if (this.returnResult != null) {
+ if (this.returnResult == ReturnResult.ROW) {
+ for (Mutation mutation : rowMutations) {
+
mutation.setAttribute(PhoenixIndexBuilderHelper.RETURN_RESULT,
+
PhoenixIndexBuilderHelper.RETURN_RESULT_ROW);
+ }
+ }
+ }
// The DeleteCompiler already generates the deletes for
indexes, so no need to do it again
rowMutationsPertainingToIndex = Collections.emptyList();
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index b9f895b777..c361868405 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -647,13 +647,16 @@ public class PhoenixStatement implements
PhoenixMonitoredStatement, SQLCloseable
lastState.getUpdateCount());
Result result = null;
if (connection.getAutoCommit()) {
- state.setReturnResult(returnResult);
+ if (isSingleRowUpdatePlan(isUpsert,
isDelete, plan)) {
+ state.setReturnResult(returnResult);
+ }
connection.commit();
if (isAtomicUpsert) {
lastUpdateCount =
connection.getMutationState()
.getNumUpdatedRowsForAutoCommit();
}
result =
connection.getMutationState().getResult();
+
connection.getMutationState().clearResult();
}
setLastQueryPlan(null);
setLastUpdateCount(lastUpdateCount);
@@ -755,6 +758,17 @@ public class PhoenixStatement implements
PhoenixMonitoredStatement, SQLCloseable
}
}
+ private static boolean isSingleRowUpdatePlan(boolean isUpsert, boolean
isDelete,
+ MutationPlan plan) {
+ boolean isSingleRowUpdate = false;
+ if (isUpsert) {
+ isSingleRowUpdate = true;
+ } else if (isDelete) {
+ isSingleRowUpdate =
plan.getContext().getScanRanges().getPointLookupCount() == 1;
+ }
+ return isSingleRowUpdate;
+ }
+
protected static interface CompilableStatement extends BindableStatement {
public <T extends StatementPlan> T compilePlan (PhoenixStatement stmt,
Sequence.ValueOp seqAction) throws SQLException;
}
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
index cf612f4600..d4c977c5cb 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
@@ -529,7 +529,7 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
BatchMutateContext context) {
for (int i = 0; i < miniBatchOp.size(); i++) {
Mutation m = miniBatchOp.getOperation(i);
- if (this.builder.isAtomicOp(m) || this.builder.returnResult(m) ||
+ if (this.builder.isAtomicOp(m) || context.returnResult ||
this.builder.isEnabled(m)) {
ImmutableBytesPtr row = new ImmutableBytesPtr(m.getRow());
context.rowsToLock.add(row);
@@ -599,6 +599,16 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
miniBatchOp.setOperationStatus(i,
new OperationStatus(SUCCESS, result));
}
+ } else if (context.returnResult) {
+ Map<ColumnReference, Pair<Cell, Boolean>> currColumnCellExprMap
= new HashMap<>();
+ byte[] rowKey = m.getRow();
+ ImmutableBytesPtr rowKeyPtr = new ImmutableBytesPtr(rowKey);
+ Pair<Put, Put> dataRowState =
context.dataRowStates.get(rowKeyPtr);
+ Put currentDataRowState = dataRowState != null ?
dataRowState.getFirst() : null;
+ if (currentDataRowState != null) {
+ updateCurrColumnCellExpr(currentDataRowState,
currColumnCellExprMap);
+ context.currColumnCellExprMap = currColumnCellExprMap;
+ }
}
}
}
@@ -1502,8 +1512,13 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
if (context.returnResult) {
Map<ColumnReference, Pair<Cell, Boolean>> currColumnCellExprMap =
context.currColumnCellExprMap;
+ if (currColumnCellExprMap == null) {
+ return;
+ }
Mutation mutation = miniBatchOp.getOperation(0);
- updateColumnCellExprMap(mutation, currColumnCellExprMap);
+ if (mutation instanceof Put) {
+ updateColumnCellExprMap(mutation, currColumnCellExprMap);
+ }
Mutation[] mutations =
miniBatchOp.getOperationsFromCoprocessors(0);
if (mutations != null) {
for (Mutation m : mutations) {
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKey2IT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKey2IT.java
index e5e2c116e4..064659ac26 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKey2IT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/OnDuplicateKey2IT.java
@@ -30,6 +30,7 @@ import java.net.URL;
import java.nio.charset.Charset;
import java.sql.Connection;
import java.sql.DriverManager;
+import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
@@ -198,37 +199,105 @@ public class OnDuplicateKey2IT extends
ParallelStatsDisabledIT {
+ "END";
validateReturnedRowAfterUpsert(conn, upsertSql, tableName,
2233.99, "col2_001", false
, null, bsonDocument2, 234);
+
+ PhoenixPreparedStatement ps =
+ conn.prepareStatement(
+ "DELETE FROM " + tableName + " WHERE PK1 = ? AND
PK2 " +
+ "= ? AND PK3 =
?").unwrap(PhoenixPreparedStatement.class);
+ ps.setString(1, "pk000");
+ ps.setDouble(2, -123.98);
+ ps.setString(3, "pk003");
+ validateReturnedRowAfterDelete(conn, ps, tableName, 2233.99,
"col2_001", true, true,
+ bsonDocument2, 234);
+ validateReturnedRowAfterDelete(conn, ps, tableName, 2233.99,
"col2_001", true, false,
+ bsonDocument2, 234);
+
+ addRows(tableName, conn);
+
+ ps = conn.prepareStatement(
+ "DELETE FROM " + tableName + " WHERE PK1 = ? AND PK2 = ?")
+ .unwrap(PhoenixPreparedStatement.class);
+ ps.setString(1, "pk001");
+ ps.setDouble(2, 122.34);
+ validateReturnedRowAfterDelete(conn, ps, tableName, 2233.99,
"col2_001", false, false,
+ bsonDocument2, 234);
+
+ ps = conn.prepareStatement(
+ "DELETE FROM " +
tableName).unwrap(PhoenixPreparedStatement.class);
+ validateReturnedRowAfterDelete(conn, ps, tableName, 2233.99,
"col2_001", false, false,
+ bsonDocument2, 234);
+
+ ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM
" + tableName);
+ assertFalse(rs.next());
+
+ addRows(tableName, conn);
+
+ ps = conn.prepareStatement(
+ "DELETE FROM " + tableName + " WHERE PK1 IN (?)
AND PK2 IN (?) AND " +
+ "PK3 IN (?, ?)")
+ .unwrap(PhoenixPreparedStatement.class);
+ ps.setString(1, "pk001");
+ ps.setDouble(2, 122.34);
+ ps.setString(3, "pk004");
+ ps.setString(4, "pk005");
+ validateReturnedRowAfterDelete(conn, ps, tableName, 2233.99,
"col2_001", false, false,
+ bsonDocument2, 234);
}
}
- private static void validateReturnedRowAfterUpsert(Connection conn,
- String upsertSql,
+ private static void addRows(String tableName, Connection conn) throws
SQLException {
+ String upsertSql;
+ upsertSql =
+ "UPSERT INTO " + tableName
+ + " (PK1, PK2, PK3, COUNTER1, COUNTER2)
VALUES('pk001', 122.34, "
+ + "'pk004', 23, 'col2_001')";
+ conn.createStatement().execute(upsertSql);
+ upsertSql =
+ "UPSERT INTO " + tableName
+ + " (PK1, PK2, PK3, COUNTER1, COUNTER2)
VALUES('pk001', 122.34, "
+ + "'pk005', 23, 'col2_001')";
+ conn.createStatement().execute(upsertSql);
+ upsertSql =
+ "UPSERT INTO " + tableName
+ + " (PK1, PK2, PK3, COUNTER1, COUNTER2)
VALUES('pk003', 122.34, "
+ + "'pk005', 23, 'col2_001')";
+ conn.createStatement().execute(upsertSql);
+ }
+
+ private static void validateReturnedRowAfterDelete(Connection conn,
+
PhoenixPreparedStatement ps,
String tableName,
Double col1,
String col2,
- boolean success,
- BsonDocument inputDoc,
+ boolean
isSinglePointLookup,
+ boolean
atomicDeleteSuccessful,
BsonDocument
expectedDoc,
Integer col4)
throws SQLException {
- final Pair<Integer, Tuple> resultPair;
- if (inputDoc != null) {
- PhoenixPreparedStatement ps =
-
conn.prepareStatement(upsertSql).unwrap(PhoenixPreparedStatement.class);
- ps.setObject(1, inputDoc);
- resultPair = ps.executeUpdateReturnRow();
- } else {
- resultPair = conn.createStatement().unwrap(PhoenixStatement.class)
- .executeUpdateReturnRow(upsertSql);
- }
- assertEquals(success ? 1 : 0, resultPair.getFirst().intValue());
+ final Pair<Integer, Tuple> resultPair = ps.executeUpdateReturnRow();
ResultTuple result = (ResultTuple) resultPair.getSecond();
+
PTable table =
conn.unwrap(PhoenixConnection.class).getTable(tableName);
+ if (!isSinglePointLookup) {
+ assertNull(result.getResult());
+ return;
+ }
Cell cell =
result.getResult()
.getColumnLatestCell(table.getColumns().get(3).getFamilyName().getBytes(),
table.getColumns().get(3).getColumnQualifierBytes());
+ if (!atomicDeleteSuccessful) {
+ assertNull(cell);
+ return;
+ }
+
+ validateReturnedRowResult(col1, col2, expectedDoc, col4, table, cell,
result);
+ }
+
+ private static void validateReturnedRowResult(Double col1, String col2,
+ BsonDocument expectedDoc,
Integer col4,
+ PTable table, Cell cell,
ResultTuple result) {
ImmutableBytesPtr ptr = new ImmutableBytesPtr();
table.getRowKeySchema().iterator(cell.getRowArray(), ptr, 1);
assertEquals("pk000",
@@ -281,6 +350,87 @@ public class OnDuplicateKey2IT extends
ParallelStatsDisabledIT {
}
}
+ private static void validateReturnedRowAfterUpsert(Connection conn,
+ String upsertSql,
+ String tableName,
+ Double col1,
+ String col2,
+ boolean success,
+ BsonDocument inputDoc,
+ BsonDocument
expectedDoc,
+ Integer col4)
+ throws SQLException {
+ final Pair<Integer, Tuple> resultPair;
+ if (inputDoc != null) {
+ PhoenixPreparedStatement ps =
+
conn.prepareStatement(upsertSql).unwrap(PhoenixPreparedStatement.class);
+ ps.setObject(1, inputDoc);
+ resultPair = ps.executeUpdateReturnRow();
+ } else {
+ resultPair = conn.createStatement().unwrap(PhoenixStatement.class)
+ .executeUpdateReturnRow(upsertSql);
+ }
+ assertEquals(success ? 1 : 0, resultPair.getFirst().intValue());
+ ResultTuple result = (ResultTuple) resultPair.getSecond();
+ PTable table =
conn.unwrap(PhoenixConnection.class).getTable(tableName);
+
+ Cell cell =
+ result.getResult()
+
.getColumnLatestCell(table.getColumns().get(3).getFamilyName().getBytes(),
+
table.getColumns().get(3).getColumnQualifierBytes());
+ validateReturnedRowResult(col1, col2, expectedDoc, col4, table, cell,
result);
+ }
+
+ @Test
+ public void testReturnRowResultForMultiPointLookup() throws Exception {
+ Assume.assumeTrue("Set correct result to RegionActionResult on hbase
versions " +
+ "2.4.18+, 2.5.9+, and 2.6.0+",
isSetCorrectResultEnabledOnHBase());
+
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE " + tableName
+ + "(PK VARCHAR PRIMARY KEY, COL1 DOUBLE, COL2 VARCHAR,
COUNTER1 "
+ + "DOUBLE, COUNTER2 VARCHAR)";
+ conn.createStatement().execute(ddl);
+ createIndex(conn, tableName);
+
+ addRows2(tableName, conn);
+
+ PhoenixPreparedStatement ps = conn.prepareStatement(
+ "DELETE FROM " + tableName + " WHERE PK IN (?, ?,
?)")
+ .unwrap(PhoenixPreparedStatement.class);
+ ps.setString(1, "pk001");
+ ps.setString(2, "pk002");
+ ps.setString(3, "pk003");
+ validateReturnedRowAfterDelete(conn, ps, tableName, 2233.99,
"col2_001", false, false,
+ null, 234);
+
+ ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM
" + tableName);
+ assertFalse(rs.next());
+ }
+ }
+
+ private static void addRows2(String tableName, Connection conn) throws
SQLException {
+ String upsertSql;
+ upsertSql =
+ "UPSERT INTO " + tableName
+ + " (PK, COL1, COL2, COUNTER1, COUNTER2)
VALUES('pk001', 122.34, "
+ + "'pk004', 23, 'col2_001')";
+ conn.createStatement().execute(upsertSql);
+ upsertSql =
+ "UPSERT INTO " + tableName
+ + " (PK, COL1, COL2, COUNTER1, COUNTER2)
VALUES('pk002', 122.34, "
+ + "'pk005', 23, 'col2_001')";
+ conn.createStatement().execute(upsertSql);
+ upsertSql =
+ "UPSERT INTO " + tableName
+ + " (PK, COL1, COL2, COUNTER1, COUNTER2)
VALUES('pk003', 122.34, "
+ + "'pk005', 23, 'col2_001')";
+ conn.createStatement().execute(upsertSql);
+ }
+
@Test
public void testReturnRowResultWithoutAutoCommit() throws Exception {
Assume.assumeTrue("Set correct result to RegionActionResult on hbase
versions " +
