HBASE-21545 NEW_VERSION_BEHAVIOR breaks Get/Scan with specified columns

Signed-off-by: Duo Zhang <zhangduo@apache.org>
Signed-off-by: stack <stack@apache.org>
Signed-off-by: Sakthi

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/querymatcher/NewVersionBehaviorTracker.java

@@ -277,26 +277,26 @@ public class NewVersionBehaviorTracker implements ColumnTracker, DeleteTracker {
 
   @Override
   public MatchCode checkColumn(Cell cell, byte type) throws IOException {
-    if (done()) {
+    if (columns == null) {
-      // No more columns left, we are done with this query
+        return MatchCode.INCLUDE;
-      return ScanQueryMatcher.MatchCode.SEEK_NEXT_ROW; // done_row
     }
-    if (columns != null) {
+
-      while (columnIndex < columns.length) {
+    while (!done()) {
-        int c = Bytes.compareTo(columns[columnIndex], 0, columns[columnIndex].length,
+      int c = CellUtil.compareQualifiers(cell,
-            cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
+        columns[columnIndex], 0, columns[columnIndex].length);
-        if (c < 0) {
+      if (c < 0) {
-          columnIndex++;
+        return MatchCode.SEEK_NEXT_COL;
-        } else if (c == 0) {
-          // We drop old version in #isDeleted, so here we must return INCLUDE.
-          return MatchCode.INCLUDE;
-        } else {
-          return MatchCode.SEEK_NEXT_COL;
-        }
       }
-      return MatchCode.SEEK_NEXT_ROW;
+
+      if (c == 0) {
+        // We drop old version in #isDeleted, so here we must return INCLUDE.
+        return MatchCode.INCLUDE;
+      }
+
+      columnIndex++;
     }
-    return MatchCode.INCLUDE;
+    // No more columns left, we are done with this query
+    return MatchCode.SEEK_NEXT_ROW;
   }
 
   @Override
@@ -351,10 +351,7 @@ public class NewVersionBehaviorTracker implements ColumnTracker, DeleteTracker {
 
   @Override
   public boolean done() {
-    // lastCq* have been updated to this cell.
+    return columns != null && columnIndex >= columns.length;
-    return !(columns == null || lastCqArray == null) && Bytes
-        .compareTo(lastCqArray, lastCqOffset, lastCqLength, columns[columnIndex], 0,
-            columns[columnIndex].length) > 0;
   }
 
   @Override

hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java

@@ -711,6 +711,18 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
       new Path(root, "mapreduce-am-staging-root-dir").toString());
   }
 
+  /**
+   *  Check whether the tests should assume NEW_VERSION_BEHAVIOR when creating
+   *  new column families. Default to false.
+   */
+  public boolean isNewVersionBehaviorEnabled(){
+    final String propName = "hbase.tests.new.version.behavior";
+    String v = System.getProperty(propName);
+    if (v != null){
+      return Boolean.parseBoolean(v);
+    }
+    return false;
+  }
 
   /**
    *  Get the HBase setting for dfs.client.read.shortcircuit from the conf or a system property.
@@ -1444,9 +1456,13 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
       BloomType type, int blockSize, Configuration c) throws IOException {
     TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(htd);
     for (byte[] family : families) {
-      builder.setColumnFamily(
+      ColumnFamilyDescriptorBuilder cfdb = ColumnFamilyDescriptorBuilder.newBuilder(family)
-          ColumnFamilyDescriptorBuilder.newBuilder(family).setBloomFilterType(type)
+        .setBloomFilterType(type)
-              .setBlocksize(blockSize).build());
+        .setBlocksize(blockSize);
+      if (isNewVersionBehaviorEnabled()) {
+          cfdb.setNewVersionBehavior(true);
+      }
+      builder.setColumnFamily(cfdb.build());
     }
     TableDescriptor td = builder.build();
     getAdmin().createTable(td, splitKeys);
@@ -1465,7 +1481,14 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
    */
   public Table createTable(TableDescriptor htd, byte[][] splitRows)
       throws IOException {
-    getAdmin().createTable(htd, splitRows);
+    TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(htd);
+    if (isNewVersionBehaviorEnabled()) {
+      for (ColumnFamilyDescriptor family : htd.getColumnFamilies()) {
+         builder.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(family)
+           .setNewVersionBehavior(true).build());
+      }
+    }
+    getAdmin().createTable(builder.build(), splitRows);
     // HBaseAdmin only waits for regions to appear in hbase:meta
     // we should wait until they are assigned
     waitUntilAllRegionsAssigned(htd.getTableName());
@@ -1529,6 +1552,9 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
     HTableDescriptor desc = new HTableDescriptor(tableName);
     for (byte[] family : families) {
       HColumnDescriptor hcd = new HColumnDescriptor(family).setMaxVersions(numVersions);
+      if (isNewVersionBehaviorEnabled()) {
+        hcd.setNewVersionBehavior(true);
+      }
       desc.addFamily(hcd);
     }
     getAdmin().createTable(desc, splitKeys);
@@ -1567,6 +1593,9 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
       HColumnDescriptor hcd = new HColumnDescriptor(family)
           .setMaxVersions(numVersions)
           .setBlocksize(blockSize);
+      if (isNewVersionBehaviorEnabled()) {
+        hcd.setNewVersionBehavior(true);
+      }
       desc.addFamily(hcd);
     }
     getAdmin().createTable(desc);
@@ -1583,6 +1612,9 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
         HColumnDescriptor hcd = new HColumnDescriptor(family)
             .setMaxVersions(numVersions)
             .setBlocksize(blockSize);
+        if (isNewVersionBehaviorEnabled()) {
+          hcd.setNewVersionBehavior(true);
+        }
         desc.addFamily(hcd);
       }
       if(cpName != null) {
@@ -1611,6 +1643,9 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
     for (byte[] family : families) {
       HColumnDescriptor hcd = new HColumnDescriptor(family)
           .setMaxVersions(numVersions[i]);
+      if (isNewVersionBehaviorEnabled()) {
+        hcd.setNewVersionBehavior(true);
+      }
       desc.addFamily(hcd);
       i++;
     }
@@ -1633,6 +1668,9 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
       throws IOException {
     HTableDescriptor desc = new HTableDescriptor(tableName);
     HColumnDescriptor hcd = new HColumnDescriptor(family);
+    if (isNewVersionBehaviorEnabled()) {
+      hcd.setNewVersionBehavior(true);
+    }
     desc.addFamily(hcd);
     getAdmin().createTable(desc, splitRows);
     // HBaseAdmin only waits for regions to appear in hbase:meta we should wait until they are
@@ -1754,13 +1792,16 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
       final int minVersions, final int versions, final int ttl, KeepDeletedCells keepDeleted) {
     HTableDescriptor htd = new HTableDescriptor(name);
     for (byte[] cfName : new byte[][]{ fam1, fam2, fam3 }) {
-      htd.addFamily(new HColumnDescriptor(cfName)
+      HColumnDescriptor hcd = new HColumnDescriptor(cfName)
           .setMinVersions(minVersions)
           .setMaxVersions(versions)
           .setKeepDeletedCells(keepDeleted)
           .setBlockCacheEnabled(false)
-          .setTimeToLive(ttl)
+          .setTimeToLive(ttl);
-      );
+      if (isNewVersionBehaviorEnabled()) {
+          hcd.setNewVersionBehavior(true);
+      }
+      htd.addFamily(hcd);
     }
     return htd;
   }
@@ -1786,6 +1827,9 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
     for (byte[] family : families) {
       HColumnDescriptor hcd = new HColumnDescriptor(family)
           .setMaxVersions(maxVersions);
+      if (isNewVersionBehaviorEnabled()) {
+          hcd.setNewVersionBehavior(true);
+      }
       desc.addFamily(hcd);
     }
     return desc;

hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestGetScanColumnsWithNewVersionBehavior.java

@@ -0,0 +1,109 @@
+/**
+ * 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.client;
+
+import static org.junit.Assert.assertArrayEquals;
+
+import java.io.IOException;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.testclassification.ClientTests;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.List;
+import java.util.ArrayList;
+
+/**
+ * Testcase for HBASE-21032, where use the wrong readType from a Scan instance which is actually a
+ * get scan and cause returning only 1 cell per rpc call.
+ */
+@Category({ ClientTests.class, MediumTests.class })
+public class TestGetScanColumnsWithNewVersionBehavior {
+
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+    HBaseClassTestRule.forClass(TestGetScanColumnsWithNewVersionBehavior.class);
+
+  private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+  private static final TableName TABLE = TableName.valueOf("table");
+  private static final byte[] CF = { 'c', 'f' };
+  private static final byte[] ROW = { 'r', 'o', 'w' };
+  private static final byte[] COLA = { 'a' };
+  private static final byte[] COLB = { 'b' };
+  private static final byte[] COLC = { 'c' };
+  private static final long TS = 42;
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    TEST_UTIL.startMiniCluster(1);
+    ColumnFamilyDescriptor cd = ColumnFamilyDescriptorBuilder
+        .newBuilder(CF)
+        .setNewVersionBehavior(true)
+        .build();
+    TEST_UTIL.createTable(TableDescriptorBuilder
+        .newBuilder(TABLE)
+        .setColumnFamily(cd)
+        .build(), null);
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    TEST_UTIL.shutdownMiniCluster();
+  }
+
+  @Test
+  public void test() throws IOException {
+    try (Table t = TEST_UTIL.getConnection().getTable(TABLE)) {
+      Cell [] expected = new Cell[2];
+      expected[0] = new KeyValue(ROW, CF, COLA, TS, COLA);
+      expected[1] = new KeyValue(ROW, CF, COLC, TS, COLC);
+
+      Put p = new Put(ROW);
+      p.addColumn(CF, COLA, TS, COLA);
+      p.addColumn(CF, COLB, TS, COLB);
+      p.addColumn(CF, COLC, TS, COLC);
+      t.put(p);
+
+      // check get request
+      Get get = new Get(ROW);
+      get.addColumn(CF, COLA);
+      get.addColumn(CF, COLC);
+      Result getResult = t.get(get);
+      assertArrayEquals(expected, getResult.rawCells());
+
+      // check scan request
+      Scan scan = new Scan(ROW);
+      scan.addColumn(CF, COLA);
+      scan.addColumn(CF, COLC);
+      ResultScanner scanner = t.getScanner(scan);
+      List scanResult = new ArrayList<Cell>();
+      for (Result result = scanner.next(); (result != null); result = scanner.next()) {
+          scanResult.addAll(result.listCells());
+      }
+      assertArrayEquals(expected, scanResult.toArray(new Cell[scanResult.size()]));
+    }
+  }
+}

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/querymatcher/TestNewVersionBehaviorTracker.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hbase.regionserver.querymatcher;
 
+import java.util.TreeSet;
 import static org.junit.Assert.assertEquals;
 
 import java.io.IOException;
@@ -33,6 +34,7 @@ import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
+
 @Category({ RegionServerTests.class, SmallTests.class })
 public class TestNewVersionBehaviorTracker {
 
@@ -40,12 +42,46 @@ public class TestNewVersionBehaviorTracker {
   public static final HBaseClassTestRule CLASS_RULE =
       HBaseClassTestRule.forClass(TestNewVersionBehaviorTracker.class);
 
+  private final byte[] col0 = Bytes.toBytes("col0");
   private final byte[] col1 = Bytes.toBytes("col1");
   private final byte[] col2 = Bytes.toBytes("col2");
+  private final byte[] col3 = Bytes.toBytes("col3");
+  private final byte[] col4 = Bytes.toBytes("col4");
   private final byte[] row = Bytes.toBytes("row");
   private final byte[] family = Bytes.toBytes("family");
   private final byte[] value = Bytes.toBytes("value");
   private final CellComparator comparator = CellComparatorImpl.COMPARATOR;
+
+  @Test
+  public void testColumns() throws IOException {
+    TreeSet<byte[]> trackedColumns = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
+    trackedColumns.add(col1);
+    trackedColumns.add(col3);
+
+    NewVersionBehaviorTracker tracker =
+        new NewVersionBehaviorTracker(trackedColumns, comparator, 1, 3, 3, 10000);
+
+    KeyValue keyValue = new KeyValue(row, family, col0, 20000, KeyValue.Type.Put, value);
+    assertEquals(DeleteResult.NOT_DELETED, tracker.isDeleted(keyValue));
+    assertEquals(MatchCode.SEEK_NEXT_COL, tracker.checkColumn(keyValue, keyValue.getTypeByte()));
+
+    keyValue = new KeyValue(row, family, col1, 20000, KeyValue.Type.Put, value);
+    assertEquals(DeleteResult.NOT_DELETED, tracker.isDeleted(keyValue));
+    assertEquals(MatchCode.INCLUDE, tracker.checkColumn(keyValue, keyValue.getTypeByte()));
+
+    keyValue = new KeyValue(row, family, col2, 20000, KeyValue.Type.Put, value);
+    assertEquals(DeleteResult.NOT_DELETED, tracker.isDeleted(keyValue));
+    assertEquals(MatchCode.SEEK_NEXT_COL, tracker.checkColumn(keyValue, keyValue.getTypeByte()));
+
+    keyValue = new KeyValue(row, family, col3, 20000, KeyValue.Type.Put, value);
+    assertEquals(DeleteResult.NOT_DELETED, tracker.isDeleted(keyValue));
+    assertEquals(MatchCode.INCLUDE, tracker.checkColumn(keyValue, keyValue.getTypeByte()));
+
+    keyValue = new KeyValue(row, family, col4, 20000, KeyValue.Type.Put, value);
+    assertEquals(DeleteResult.NOT_DELETED, tracker.isDeleted(keyValue));
+    assertEquals(MatchCode.SEEK_NEXT_ROW, tracker.checkColumn(keyValue, keyValue.getTypeByte()));
+  }
+
   @Test
   public void testMaxVersionMask() {
     NewVersionBehaviorTracker tracker =