You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@phoenix.apache.org by GitBox <gi...@apache.org> on 2020/11/20 23:39:53 UTC
[GitHub] [phoenix] gjacoby126 commented on a change in pull request #978: [PHOENIX-6213] Extend Cell Tags to Delete object to store source of operation.
gjacoby126 commented on a change in pull request #978:
URL: https://github.com/apache/phoenix/pull/978#discussion_r528022823
##########
File path: phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
##########
@@ -991,6 +999,46 @@ public void preBatchMutateWithExceptions(ObserverContext<RegionCoprocessorEnviro
}
}
+ /*
+ Set Cell Tags to delete markers with source of operation attribute.
+ */
+ private void setDeleteAttributes(MiniBatchOperationInProgress<Mutation> miniBatchOp)
+ throws IOException {
+ for (int i = 0; i < miniBatchOp.size(); i++) {
+ Mutation m = miniBatchOp.getOperation(i);
+ if (!(m instanceof Delete)) {
+ // Ignore if it is not Delete type.
+ continue;
+ }
+ byte[] sourceOpAttr = m.getAttribute(QueryServices.SOURCE_OPERATION_ATTRIB);
+ if (sourceOpAttr == null) {
+ continue;
+ }
+
+ // TODO: Which tag implementation to use ? ArrayBackedTag or ByteBufferTag ?
+ Tag sourceOpTag = new ArrayBackedTag(PhoenixTagType.SOURCE_OPERATION_TAG_TYPE, sourceOpAttr);
+ List<Cell> updatedCells = new ArrayList<>();
+ for (CellScanner cellScanner = m.cellScanner(); cellScanner.advance();) {
+ Cell cell = cellScanner.current();
+ // TODO: Cell#getTagsArray, Cell#getTagsOffset, Cell#getTagsLength are deprecated.
Review comment:
See RawCell, which unlike the Cell tag methods, isn't deprecated, and is IA LimitedPrivate.COPROC. You'll probably have to downcast.
##########
File path: phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
##########
@@ -991,6 +999,46 @@ public void preBatchMutateWithExceptions(ObserverContext<RegionCoprocessorEnviro
}
}
+ /*
+ Set Cell Tags to delete markers with source of operation attribute.
+ */
+ private void setDeleteAttributes(MiniBatchOperationInProgress<Mutation> miniBatchOp)
+ throws IOException {
+ for (int i = 0; i < miniBatchOp.size(); i++) {
+ Mutation m = miniBatchOp.getOperation(i);
+ if (!(m instanceof Delete)) {
+ // Ignore if it is not Delete type.
+ continue;
+ }
+ byte[] sourceOpAttr = m.getAttribute(QueryServices.SOURCE_OPERATION_ATTRIB);
+ if (sourceOpAttr == null) {
+ continue;
+ }
+
+ // TODO: Which tag implementation to use ? ArrayBackedTag or ByteBufferTag ?
Review comment:
I believe ArrayBackedTag is for Cells built on the heap, and ByteBufferTag is for Cells built off-heap. You're building Cells on-heap, so I assume you'd want ArrayBackedTag. That said, we need to be very careful we're not accidentally leaking memory here by swapping the provided Cells with new ones -- I'm not clear on when the underlying HBase is using on-heap vs off-heap Cells. Needs more research
##########
File path: phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
##########
@@ -991,6 +999,46 @@ public void preBatchMutateWithExceptions(ObserverContext<RegionCoprocessorEnviro
}
}
+ /*
+ Set Cell Tags to delete markers with source of operation attribute.
+ */
+ private void setDeleteAttributes(MiniBatchOperationInProgress<Mutation> miniBatchOp)
+ throws IOException {
+ for (int i = 0; i < miniBatchOp.size(); i++) {
+ Mutation m = miniBatchOp.getOperation(i);
+ if (!(m instanceof Delete)) {
+ // Ignore if it is not Delete type.
+ continue;
+ }
+ byte[] sourceOpAttr = m.getAttribute(QueryServices.SOURCE_OPERATION_ATTRIB);
+ if (sourceOpAttr == null) {
+ continue;
+ }
+
+ // TODO: Which tag implementation to use ? ArrayBackedTag or ByteBufferTag ?
+ Tag sourceOpTag = new ArrayBackedTag(PhoenixTagType.SOURCE_OPERATION_TAG_TYPE, sourceOpAttr);
+ List<Cell> updatedCells = new ArrayList<>();
+ for (CellScanner cellScanner = m.cellScanner(); cellScanner.advance();) {
+ Cell cell = cellScanner.current();
+ // TODO: Cell#getTagsArray, Cell#getTagsOffset, Cell#getTagsLength are deprecated.
+ //TODO: Need to replace them with new methods.
+ List<Tag> tags = TagUtil.asList(cell.getTagsArray(), cell.getTagsOffset(),
+ cell.getTagsLength());
+ tags.add(sourceOpTag);
+ // TODO: PrivateCellUtil's IA is Private. Need to change it to LP with
+ // TODO: IA.COPROC.
Review comment:
See RawCellBuilder, which is IA LimitedPrivate.COPROC.
##########
File path: phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
##########
@@ -943,6 +960,164 @@ public void testDeleteFilterWithMultipleIndexes() throws Exception {
}
}
}
-}
+ /*
+ Tests whether we have cell tags in delete marker for
+ ClientSelectDeleteMutationPlan.
+ */
+ @Test
+ public void testDeleteClientDeleteMutationPlan() throws Exception {
+ String tableName = generateUniqueName();
+ String indexName = generateUniqueName();
+ String tagValue = "customer-delete";
+ String delete = "DELETE FROM " + tableName + " WHERE v1 = 'foo'";
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ // Add tag "customer-delete" to delete marker.
+ props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
+
+ createAndUpsertTable(tableName, indexName, props);
+ // Make sure that the plan creates is of ClientSelectDeleteMutationPlan
+ verifyDeletePlan(delete, "ClientSelectDeleteMutationPlan", props);
+ executeDelete(delete, props, 1);
+ String startRowKeyForBaseTable = "1";
+ String startRowKeyForIndexTable = "foo";
+ // Make sure that Delete Marker has cell tag for base table
+ // and has no cell tag for index table.
+ checkTagPresentInDeleteMarker(tableName, startRowKeyForBaseTable, true, tagValue);
+ checkTagPresentInDeleteMarker(indexName, startRowKeyForIndexTable, false, null);
+ }
+
+ /*
+ Tests whether we have cell tags in delete marker for
+ ServerSelectDeleteMutationPlan.
+ */
+ @Test
+ public void testDeleteServerDeleteMutationPlan() throws Exception {
+ String tableName = generateUniqueName();
+ String indexName = generateUniqueName();
+ String tagValue = "customer-delete";
+ String delete = "DELETE FROM " + tableName;
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
+
+ createAndUpsertTable(tableName, indexName, props);
+ // Make sure that the plan creates is of ServerSelectDeleteMutationPlan
+ verifyDeletePlan(delete, "ServerSelectDeleteMutationPlan", props);
+ executeDelete(delete, props, 2);
+
+ String startRowKeyForBaseTable = "1";
+ String startRowKeyForIndexTable = "foo";
+ // Make sure that Delete Marker has cell tag for base table
+ // and has no cell tag for index table.
+ checkTagPresentInDeleteMarker(tableName, startRowKeyForBaseTable, true, tagValue);
+ checkTagPresentInDeleteMarker(indexName, startRowKeyForIndexTable, false, null);
+ }
+
+ /*
+ Tests whether we have cell tags in delete marker for
+ MultiRowDeleteMutationPlan.
+ */
+ @Test
+ public void testDeleteMultiRowDeleteMutationPlan() throws Exception {
+ String tableName = generateUniqueName();
+ String tagValue = "customer-delete";
+ String delete = "DELETE FROM " + tableName + " WHERE k = 1";
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
+ // Don't create index table. We will use MultiRowDeleteMutationPlan
+ // if there is no index present for a table.
+ createAndUpsertTable(tableName, null, props);
+ // Make sure that the plan creates is of MultiRowDeleteMutationPlan
+ verifyDeletePlan(delete, "MultiRowDeleteMutationPlan", props);
+ executeDelete(delete, props, 1);
+ String startRowKeyForBaseTable = "1";
+ // Make sure that Delete Marker has cell tag for base table.
+ // We haven't created index table for this test case.
+ checkTagPresentInDeleteMarker(tableName, startRowKeyForBaseTable, true, tagValue);
+ }
+
+ /*
+ Verify whether plan that we create for delete statement is of planName
+ */
+ private void verifyDeletePlan(String delete, String planName, Properties props)
+ throws SQLException {
+ try(Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ PhoenixStatement stmt = conn.createStatement().unwrap(PhoenixStatement.class);
+ SQLParser parser = new SQLParser(delete);
+ DeleteStatement deleteStmt = (DeleteStatement) parser.parseStatement();
+ DeleteCompiler compiler = new DeleteCompiler(stmt, null);
+ MutationPlan plan = compiler.compile(deleteStmt);
+ assertTrue(plan.getClass().getName().contains(planName));
+ }
+ }
+ private void createAndUpsertTable(String tableName, String indexName, Properties props)
+ throws SQLException {
+ String ddl = "CREATE TABLE " + tableName +
+ " (k INTEGER NOT NULL PRIMARY KEY, v1 VARCHAR, v2 VARCHAR)";
+ try(Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ try (Statement statement = conn.createStatement()) {
+ statement.execute(ddl);
+ if (indexName != null) {
+ String indexDdl1 = "CREATE INDEX " + indexName + " ON " + tableName + "(v1,v2)";
+ statement.execute(indexDdl1);
+ }
+ }
+ conn.createStatement().execute(
+ "upsert into " + tableName + " values (1, 'foo', 'foo1')");
+ conn.createStatement().execute(
+ "upsert into " + tableName + " values (2, 'bar', 'bar1')");
+ conn.commit();
+ }
+ }
+
+ private void executeDelete(String delete, Properties props, int deleteRowCount)
+ throws SQLException {
+ try(Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ try (Statement statement = conn.createStatement()) {
+ int rs = statement.executeUpdate(delete);
+ assertEquals( deleteRowCount, rs);
+ }
+ }
+ }
+ /*
+ Verify whether we have tags present for base table and not present for
+ index tables.
+ */
+ private void checkTagPresentInDeleteMarker(String tableName, String startRowKey,
+ boolean tagPresent, String tagValue) throws IOException {
+ List<Cell> values = new ArrayList<>();
+ TableName table = TableName.valueOf(tableName);
+ // Scan table with specified startRowKey
+ for (HRegion region : getUtility().getHBaseCluster().getRegions(table)) {
+ values.clear();
+ Scan scan = new Scan();
+ // Make sure to set rawScan to true so that we will get Delete Markers.
+ scan.setRaw(true);
+ scan.withStartRow(Bytes.toBytes(startRowKey));
+ RegionScanner scanner = region.getScanner(scan);
+ scanner.next(values);
+ if (!values.isEmpty()) {
+ break;
+ }
+ }
+ assertTrue("Values shouldn't be empty", !values.isEmpty());
+ Cell first = values.get(0);
+ assertTrue("First cell should be delete marker ",
+ PrivateCellUtil.isDelete(first.getType().getCode()));
+ List<Tag> tags = PrivateCellUtil.getTags(first);
+ if (tagPresent) {
+ assertEquals(1, tags.size());
+ Optional<Tag> optional =
+ PrivateCellUtil.getTag(first, PhoenixTagType.SOURCE_OPERATION_TAG_TYPE);
Review comment:
PrivateCellUtil is IA.Private -- need to use one of the IA.LimitedPrivate or better classes such as RawCell
##########
File path: phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
##########
@@ -943,6 +960,164 @@ public void testDeleteFilterWithMultipleIndexes() throws Exception {
}
}
}
-}
+ /*
+ Tests whether we have cell tags in delete marker for
+ ClientSelectDeleteMutationPlan.
+ */
+ @Test
+ public void testDeleteClientDeleteMutationPlan() throws Exception {
+ String tableName = generateUniqueName();
+ String indexName = generateUniqueName();
+ String tagValue = "customer-delete";
+ String delete = "DELETE FROM " + tableName + " WHERE v1 = 'foo'";
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ // Add tag "customer-delete" to delete marker.
+ props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
+
+ createAndUpsertTable(tableName, indexName, props);
+ // Make sure that the plan creates is of ClientSelectDeleteMutationPlan
+ verifyDeletePlan(delete, "ClientSelectDeleteMutationPlan", props);
+ executeDelete(delete, props, 1);
+ String startRowKeyForBaseTable = "1";
+ String startRowKeyForIndexTable = "foo";
+ // Make sure that Delete Marker has cell tag for base table
+ // and has no cell tag for index table.
+ checkTagPresentInDeleteMarker(tableName, startRowKeyForBaseTable, true, tagValue);
+ checkTagPresentInDeleteMarker(indexName, startRowKeyForIndexTable, false, null);
+ }
+
+ /*
+ Tests whether we have cell tags in delete marker for
+ ServerSelectDeleteMutationPlan.
+ */
+ @Test
+ public void testDeleteServerDeleteMutationPlan() throws Exception {
+ String tableName = generateUniqueName();
+ String indexName = generateUniqueName();
+ String tagValue = "customer-delete";
+ String delete = "DELETE FROM " + tableName;
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
+
+ createAndUpsertTable(tableName, indexName, props);
+ // Make sure that the plan creates is of ServerSelectDeleteMutationPlan
+ verifyDeletePlan(delete, "ServerSelectDeleteMutationPlan", props);
Review comment:
rather than hardcode the string, can use the class name
##########
File path: phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
##########
@@ -943,6 +960,164 @@ public void testDeleteFilterWithMultipleIndexes() throws Exception {
}
}
}
-}
+ /*
+ Tests whether we have cell tags in delete marker for
+ ClientSelectDeleteMutationPlan.
+ */
+ @Test
+ public void testDeleteClientDeleteMutationPlan() throws Exception {
+ String tableName = generateUniqueName();
+ String indexName = generateUniqueName();
+ String tagValue = "customer-delete";
+ String delete = "DELETE FROM " + tableName + " WHERE v1 = 'foo'";
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ // Add tag "customer-delete" to delete marker.
+ props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
+
+ createAndUpsertTable(tableName, indexName, props);
+ // Make sure that the plan creates is of ClientSelectDeleteMutationPlan
+ verifyDeletePlan(delete, "ClientSelectDeleteMutationPlan", props);
+ executeDelete(delete, props, 1);
+ String startRowKeyForBaseTable = "1";
+ String startRowKeyForIndexTable = "foo";
+ // Make sure that Delete Marker has cell tag for base table
+ // and has no cell tag for index table.
+ checkTagPresentInDeleteMarker(tableName, startRowKeyForBaseTable, true, tagValue);
+ checkTagPresentInDeleteMarker(indexName, startRowKeyForIndexTable, false, null);
+ }
+
+ /*
+ Tests whether we have cell tags in delete marker for
+ ServerSelectDeleteMutationPlan.
+ */
+ @Test
+ public void testDeleteServerDeleteMutationPlan() throws Exception {
+ String tableName = generateUniqueName();
+ String indexName = generateUniqueName();
+ String tagValue = "customer-delete";
+ String delete = "DELETE FROM " + tableName;
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
+
+ createAndUpsertTable(tableName, indexName, props);
+ // Make sure that the plan creates is of ServerSelectDeleteMutationPlan
+ verifyDeletePlan(delete, "ServerSelectDeleteMutationPlan", props);
+ executeDelete(delete, props, 2);
+
+ String startRowKeyForBaseTable = "1";
+ String startRowKeyForIndexTable = "foo";
+ // Make sure that Delete Marker has cell tag for base table
+ // and has no cell tag for index table.
+ checkTagPresentInDeleteMarker(tableName, startRowKeyForBaseTable, true, tagValue);
+ checkTagPresentInDeleteMarker(indexName, startRowKeyForIndexTable, false, null);
+ }
+
+ /*
+ Tests whether we have cell tags in delete marker for
+ MultiRowDeleteMutationPlan.
+ */
+ @Test
+ public void testDeleteMultiRowDeleteMutationPlan() throws Exception {
+ String tableName = generateUniqueName();
+ String tagValue = "customer-delete";
+ String delete = "DELETE FROM " + tableName + " WHERE k = 1";
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
+ // Don't create index table. We will use MultiRowDeleteMutationPlan
+ // if there is no index present for a table.
+ createAndUpsertTable(tableName, null, props);
+ // Make sure that the plan creates is of MultiRowDeleteMutationPlan
+ verifyDeletePlan(delete, "MultiRowDeleteMutationPlan", props);
+ executeDelete(delete, props, 1);
+ String startRowKeyForBaseTable = "1";
+ // Make sure that Delete Marker has cell tag for base table.
+ // We haven't created index table for this test case.
+ checkTagPresentInDeleteMarker(tableName, startRowKeyForBaseTable, true, tagValue);
+ }
+
+ /*
+ Verify whether plan that we create for delete statement is of planName
+ */
+ private void verifyDeletePlan(String delete, String planName, Properties props)
+ throws SQLException {
+ try(Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ PhoenixStatement stmt = conn.createStatement().unwrap(PhoenixStatement.class);
+ SQLParser parser = new SQLParser(delete);
+ DeleteStatement deleteStmt = (DeleteStatement) parser.parseStatement();
+ DeleteCompiler compiler = new DeleteCompiler(stmt, null);
+ MutationPlan plan = compiler.compile(deleteStmt);
+ assertTrue(plan.getClass().getName().contains(planName));
+ }
+ }
+ private void createAndUpsertTable(String tableName, String indexName, Properties props)
+ throws SQLException {
+ String ddl = "CREATE TABLE " + tableName +
+ " (k INTEGER NOT NULL PRIMARY KEY, v1 VARCHAR, v2 VARCHAR)";
+ try(Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ try (Statement statement = conn.createStatement()) {
+ statement.execute(ddl);
+ if (indexName != null) {
+ String indexDdl1 = "CREATE INDEX " + indexName + " ON " + tableName + "(v1,v2)";
+ statement.execute(indexDdl1);
+ }
+ }
+ conn.createStatement().execute(
+ "upsert into " + tableName + " values (1, 'foo', 'foo1')");
+ conn.createStatement().execute(
+ "upsert into " + tableName + " values (2, 'bar', 'bar1')");
+ conn.commit();
+ }
+ }
+
+ private void executeDelete(String delete, Properties props, int deleteRowCount)
+ throws SQLException {
+ try(Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ try (Statement statement = conn.createStatement()) {
+ int rs = statement.executeUpdate(delete);
+ assertEquals( deleteRowCount, rs);
+ }
+ }
+ }
+ /*
+ Verify whether we have tags present for base table and not present for
+ index tables.
+ */
+ private void checkTagPresentInDeleteMarker(String tableName, String startRowKey,
+ boolean tagPresent, String tagValue) throws IOException {
+ List<Cell> values = new ArrayList<>();
+ TableName table = TableName.valueOf(tableName);
+ // Scan table with specified startRowKey
+ for (HRegion region : getUtility().getHBaseCluster().getRegions(table)) {
+ values.clear();
+ Scan scan = new Scan();
+ // Make sure to set rawScan to true so that we will get Delete Markers.
+ scan.setRaw(true);
+ scan.withStartRow(Bytes.toBytes(startRowKey));
+ RegionScanner scanner = region.getScanner(scan);
+ scanner.next(values);
+ if (!values.isEmpty()) {
+ break;
+ }
+ }
+ assertTrue("Values shouldn't be empty", !values.isEmpty());
+ Cell first = values.get(0);
+ assertTrue("First cell should be delete marker ",
+ PrivateCellUtil.isDelete(first.getType().getCode()));
Review comment:
Can use CellUtil.isDelete, which is IA.Public
##########
File path: phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
##########
@@ -943,6 +960,164 @@ public void testDeleteFilterWithMultipleIndexes() throws Exception {
}
}
}
-}
+ /*
+ Tests whether we have cell tags in delete marker for
+ ClientSelectDeleteMutationPlan.
+ */
+ @Test
+ public void testDeleteClientDeleteMutationPlan() throws Exception {
+ String tableName = generateUniqueName();
+ String indexName = generateUniqueName();
+ String tagValue = "customer-delete";
+ String delete = "DELETE FROM " + tableName + " WHERE v1 = 'foo'";
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ // Add tag "customer-delete" to delete marker.
+ props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
+
+ createAndUpsertTable(tableName, indexName, props);
+ // Make sure that the plan creates is of ClientSelectDeleteMutationPlan
+ verifyDeletePlan(delete, "ClientSelectDeleteMutationPlan", props);
Review comment:
Rather than hardcode the string, can use the classname
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org