Repository: hbase
Updated Branches:
  refs/heads/master 7755d4bee -> 3ab895979


HBASE-21660 Apply the cell to right memstore for increment/append operation


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/3ab89597
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/3ab89597
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/3ab89597

Branch: refs/heads/master
Commit: 3ab895979b643a2980bcdb7fee2078f14b614210
Parents: 7755d4b
Author: Guanghao Zhang <zg...@apache.org>
Authored: Sun Dec 30 18:52:03 2018 +0800
Committer: Guanghao Zhang <zg...@apache.org>
Committed: Tue Jan 1 17:32:44 2019 +0800

----------------------------------------------------------------------
 .../hadoop/hbase/regionserver/HRegion.java      |  14 +-
 .../TestPostIncrementAndAppendBeforeWAL.java    | 235 +++++++++++++++++++
 2 files changed, 245 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/3ab89597/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
index ec222c7..5ab61fa 100644
--- 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
@@ -7980,12 +7980,18 @@ public class HRegion implements HeapSize, 
PropagatingConfigurationObserver, Regi
     for (Map.Entry<byte [], List<Cell>> entry: 
mutation.getFamilyCellMap().entrySet()) {
       final byte[] columnFamilyName = entry.getKey();
       List<Cell> deltas = entry.getValue();
-      HStore store = this.stores.get(columnFamilyName);
       // Reckon for the Store what to apply to WAL and MemStore.
-      List<Cell> toApply =
-        reckonDeltasByStore(store, op, mutation, effectiveDurability, now, 
deltas, results);
+      List<Cell> toApply = reckonDeltasByStore(stores.get(columnFamilyName), 
op, mutation,
+        effectiveDurability, now, deltas, results);
       if (!toApply.isEmpty()) {
-        forMemStore.put(store, toApply);
+        for (Cell cell : toApply) {
+          HStore store = getStore(cell);
+          if (store == null) {
+            checkFamily(CellUtil.cloneFamily(cell));
+          } else {
+            forMemStore.computeIfAbsent(store, key -> new 
ArrayList<>()).add(cell);
+          }
+        }
         if (writeToWAL) {
           if (walEdit == null) {
             walEdit = new WALEdit();

http://git-wip-us.apache.org/repos/asf/hbase/blob/3ab89597/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestPostIncrementAndAppendBeforeWAL.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestPostIncrementAndAppendBeforeWAL.java
 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestPostIncrementAndAppendBeforeWAL.java
new file mode 100644
index 0000000..031960b
--- /dev/null
+++ 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestPostIncrementAndAppendBeforeWAL.java
@@ -0,0 +1,235 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.coprocessor;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellBuilderType;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.ExtendedCellBuilderFactory;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.client.TestFromClientSide;
+import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException;
+import org.apache.hadoop.hbase.testclassification.CoprocessorTests;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test coprocessor methods
+ * {@link RegionObserver#postIncrementBeforeWAL(ObserverContext, Mutation, 
List)} and
+ * {@link RegionObserver#postAppendBeforeWAL(ObserverContext, Mutation, 
List)}. These methods may
+ * change the cells which will be applied to memstore and WAL. So add unit 
test for the case which
+ * change the cell's column family.
+ */
+@Category({CoprocessorTests.class, MediumTests.class})
+public class TestPostIncrementAndAppendBeforeWAL {
+
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+      HBaseClassTestRule.forClass(TestPostIncrementAndAppendBeforeWAL.class);
+
+  @Rule
+  public TestName name = new TestName();
+
+  private static final Logger LOG = 
LoggerFactory.getLogger(TestFromClientSide.class);
+
+  private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();
+
+  private static Connection connection;
+
+  private static final byte [] ROW = Bytes.toBytes("row");
+  private static final String CF1 = "cf1";
+  private static final byte[] CF1_BYTES = Bytes.toBytes(CF1);
+  private static final String CF2 = "cf2";
+  private static final byte[] CF2_BYTES = Bytes.toBytes(CF2);
+  private static final String CF_NOT_EXIST = "cf_not_exist";
+  private static final byte[] CF_NOT_EXIST_BYTES = Bytes.toBytes(CF_NOT_EXIST);
+  private static final byte[] CQ1 = Bytes.toBytes("cq1");
+  private static final byte[] CQ2 = Bytes.toBytes("cq2");
+  private static final byte[] VALUE = Bytes.toBytes("value");
+
+  @BeforeClass
+  public static void setupBeforeClass() throws Exception {
+    UTIL.startMiniCluster();
+    connection = UTIL.getConnection();
+  }
+
+  @AfterClass
+  public static void tearDownAfterClass() throws Exception {
+    connection.close();
+    UTIL.shutdownMiniCluster();
+  }
+
+  private void createTableWithCoprocessor(TableName tableName, String 
coprocessor)
+      throws IOException {
+    TableDescriptor tableDesc = TableDescriptorBuilder.newBuilder(tableName)
+        
.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(CF1_BYTES).build())
+        
.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(CF2_BYTES).build())
+        .setCoprocessor(coprocessor).build();
+    connection.getAdmin().createTable(tableDesc);
+  }
+
+  @Test
+  public void testChangeCellWithDifferntColumnFamily() throws Exception {
+    TableName tableName = TableName.valueOf(name.getMethodName());
+    createTableWithCoprocessor(tableName,
+      ChangeCellWithDifferntColumnFamilyObserver.class.getName());
+
+    try (Table table = connection.getTable(tableName)) {
+      Increment increment = new Increment(ROW).addColumn(CF1_BYTES, CQ1, 1);
+      table.increment(increment);
+      Get get = new Get(ROW).addColumn(CF2_BYTES, CQ1);
+      Result result = table.get(get);
+      assertEquals(1, result.size());
+      assertEquals(1, Bytes.toLong(result.getValue(CF2_BYTES, CQ1)));
+
+      Append append = new Append(ROW).addColumn(CF1_BYTES, CQ2, VALUE);
+      table.append(append);
+      get = new Get(ROW).addColumn(CF2_BYTES, CQ2);
+      result = table.get(get);
+      assertEquals(1, result.size());
+      assertTrue(Bytes.equals(VALUE, result.getValue(CF2_BYTES, CQ2)));
+    }
+  }
+
+  @Test
+  public void testChangeCellWithNotExistColumnFamily() throws Exception {
+    TableName tableName = TableName.valueOf(name.getMethodName());
+    createTableWithCoprocessor(tableName,
+      ChangeCellWithNotExistColumnFamilyObserver.class.getName());
+
+    try (Table table = connection.getTable(tableName)) {
+      try {
+        Increment increment = new Increment(ROW).addColumn(CF1_BYTES, CQ1, 1);
+        table.increment(increment);
+        fail("should throw NoSuchColumnFamilyException");
+      } catch (Exception e) {
+        assertTrue(e instanceof NoSuchColumnFamilyException);
+      }
+      try {
+        Append append = new Append(ROW).addColumn(CF1_BYTES, CQ2, VALUE);
+        table.append(append);
+        fail("should throw NoSuchColumnFamilyException");
+      } catch (Exception e) {
+        assertTrue(e instanceof NoSuchColumnFamilyException);
+      }
+    }
+  }
+
+  public static class ChangeCellWithDifferntColumnFamilyObserver
+      implements RegionCoprocessor, RegionObserver {
+    @Override
+    public Optional<RegionObserver> getRegionObserver() {
+      return Optional.of(this);
+    }
+
+    @Override
+    public List<Pair<Cell, Cell>> postIncrementBeforeWAL(
+        ObserverContext<RegionCoprocessorEnvironment> ctx, Mutation mutation,
+        List<Pair<Cell, Cell>> cellPairs) throws IOException {
+      return cellPairs.stream()
+          .map(
+            pair -> new Pair<>(pair.getFirst(), 
newCellWithDifferentColumnFamily(pair.getSecond())))
+          .collect(Collectors.toList());
+    }
+
+    private Cell newCellWithDifferentColumnFamily(Cell cell) {
+      return ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY)
+          .setRow(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())
+          .setFamily(CF2_BYTES, 0, 
CF2_BYTES.length).setQualifier(CellUtil.cloneQualifier(cell))
+          .setTimestamp(cell.getTimestamp()).setType(cell.getType().getCode())
+          .setValue(CellUtil.cloneValue(cell)).build();
+    }
+
+    @Override
+    public List<Pair<Cell, Cell>> postAppendBeforeWAL(
+        ObserverContext<RegionCoprocessorEnvironment> ctx, Mutation mutation,
+        List<Pair<Cell, Cell>> cellPairs) throws IOException {
+      return cellPairs.stream()
+          .map(
+            pair -> new Pair<>(pair.getFirst(), 
newCellWithDifferentColumnFamily(pair.getSecond())))
+          .collect(Collectors.toList());
+    }
+  }
+
+  public static class ChangeCellWithNotExistColumnFamilyObserver
+      implements RegionCoprocessor, RegionObserver {
+    @Override
+    public Optional<RegionObserver> getRegionObserver() {
+      return Optional.of(this);
+    }
+
+    @Override
+    public List<Pair<Cell, Cell>> postIncrementBeforeWAL(
+        ObserverContext<RegionCoprocessorEnvironment> ctx, Mutation mutation,
+        List<Pair<Cell, Cell>> cellPairs) throws IOException {
+      return cellPairs.stream()
+          .map(
+            pair -> new Pair<>(pair.getFirst(), 
newCellWithNotExistColumnFamily(pair.getSecond())))
+          .collect(Collectors.toList());
+    }
+
+    private Cell newCellWithNotExistColumnFamily(Cell cell) {
+      return ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY)
+          .setRow(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())
+          .setFamily(CF_NOT_EXIST_BYTES, 0, CF_NOT_EXIST_BYTES.length)
+          
.setQualifier(CellUtil.cloneQualifier(cell)).setTimestamp(cell.getTimestamp())
+          
.setType(cell.getType().getCode()).setValue(CellUtil.cloneValue(cell)).build();
+    }
+
+    @Override
+    public List<Pair<Cell, Cell>> postAppendBeforeWAL(
+        ObserverContext<RegionCoprocessorEnvironment> ctx, Mutation mutation,
+        List<Pair<Cell, Cell>> cellPairs) throws IOException {
+      return cellPairs.stream()
+          .map(
+            pair -> new Pair<>(pair.getFirst(), 
newCellWithNotExistColumnFamily(pair.getSecond())))
+          .collect(Collectors.toList());
+    }
+  }
+}
\ No newline at end of file

Reply via email to