HBASE-17613 avoid copy of family when initializing the FSWALEntry

Signed-off-by: tedyu <yuzhihong@gmail.com>

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSWALEntry.java

@@ -18,14 +18,17 @@
 package org.apache.hadoop.hbase.regionserver.wal;
 
 
-import com.google.common.collect.Sets;
+import com.google.common.annotations.VisibleForTesting;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
 import java.util.Set;
+import java.util.TreeSet;
+import static java.util.stream.Collectors.toCollection;
 
 import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellComparator;
 import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
@@ -64,24 +67,26 @@ class FSWALEntry extends Entry {
     this.txid = txid;
     if (inMemstore) {
       // construct familyNames here to reduce the work of log sinker.
-      ArrayList<Cell> cells = this.getEdit().getCells();
+      this.familyNames = collectFamilies(edit.getCells());
-      if (CollectionUtils.isEmpty(cells)) {
-        this.familyNames = Collections.<byte[]> emptySet();
-      } else {
-        Set<byte[]> familySet = Sets.newTreeSet(Bytes.BYTES_COMPARATOR);
-        for (Cell cell : cells) {
-          if (!CellUtil.matchingFamily(cell, WALEdit.METAFAMILY)) {
-            // TODO: Avoid this clone?
-            familySet.add(CellUtil.cloneFamily(cell));
-          }
-        }
-        this.familyNames = Collections.unmodifiableSet(familySet);
-      }
     } else {
       this.familyNames = Collections.<byte[]> emptySet();
     }
   }
 
+  @VisibleForTesting
+  static Set<byte[]> collectFamilies(List<Cell> cells) {
+    if (CollectionUtils.isEmpty(cells)) {
+      return Collections.<byte[]> emptySet();
+    } else {
+      return cells.stream()
+           .filter(v -> !CellUtil.matchingFamily(v, WALEdit.METAFAMILY))
+           .collect(toCollection(() -> new TreeSet<>(CellComparator::compareFamilies)))
+           .stream()
+           .map(CellUtil::cloneFamily)
+           .collect(toCollection(() -> new TreeSet<>(Bytes.BYTES_COMPARATOR)));
+    }
+  }
+
   public String toString() {
     return "sequence=" + this.txid + ", " + super.toString();
   };

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestFSWALEntry.java

@@ -0,0 +1,62 @@
+/**
+ *
+ * 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.regionserver.wal;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+
+@Category({ RegionServerTests.class, SmallTests.class })
+public class TestFSWALEntry {
+  @Test
+  public void testCollectFamilies() {
+    byte[] family0 = Bytes.toBytes("family0");
+    byte[] family1 = Bytes.toBytes("family1");
+    byte[] family2 = Bytes.toBytes("family2");
+
+    List<Cell> cells = new ArrayList<>();
+    assertEquals(0, FSWALEntry.collectFamilies(cells).size());
+
+    cells.add(CellUtil.createCell(family0, family0, family0));
+    assertEquals(1, FSWALEntry.collectFamilies(cells).size());
+
+    cells.add(CellUtil.createCell(family1, family1, family1));
+    assertEquals(2, FSWALEntry.collectFamilies(cells).size());
+
+    cells.add(CellUtil.createCell(family0, family0, family0));
+    cells.add(CellUtil.createCell(family1, family1, family1));
+    assertEquals(2, FSWALEntry.collectFamilies(cells).size());
+
+    cells.add(CellUtil.createCell(family2, family2, family2));
+    assertEquals(3, FSWALEntry.collectFamilies(cells).size());
+
+    cells.add(CellUtil.createCell(WALEdit.METAFAMILY, WALEdit.METAFAMILY, WALEdit.METAFAMILY));
+    assertEquals(3, FSWALEntry.collectFamilies(cells).size());
+  }
+}
+