You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by ma...@apache.org on 2015/08/25 12:41:03 UTC
[3/3] incubator-kylin git commit: pass test case and add toggle on
fuzzyfilterv2
pass test case and add toggle on fuzzyfilterv2
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/f1dfe849
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/f1dfe849
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/f1dfe849
Branch: refs/heads/KYLIN-740
Commit: f1dfe84957943cab4fa239c6c7cf26d6e8af7add
Parents: 747075a
Author: honma <ho...@ebay.com>
Authored: Tue Aug 25 18:42:26 2015 +0800
Committer: honma <ho...@ebay.com>
Committed: Tue Aug 25 18:42:26 2015 +0800
----------------------------------------------------------------------
.../hbase/cube/v1/CubeSegmentTupleIterator.java | 12 +-
.../hbase/cube/v1/filter/FuzzyRowFilterV2.java | 98 +++---
.../v1/filter/TestFuzzyRowFilterEndToEnd.java | 352 -------------------
.../v1/filter/TestFuzzyRowFilterV2EndToEnd.java | 352 +++++++++++++++++++
4 files changed, 402 insertions(+), 412 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/f1dfe849/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeSegmentTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeSegmentTupleIterator.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeSegmentTupleIterator.java
index 346b92f..4813898 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeSegmentTupleIterator.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeSegmentTupleIterator.java
@@ -44,6 +44,7 @@ import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.tuple.ITupleIterator;
import org.apache.kylin.storage.StorageContext;
import org.apache.kylin.storage.hbase.cube.v1.coprocessor.observer.ObserverEnabler;
+import org.apache.kylin.storage.hbase.cube.v1.filter.FuzzyRowFilterV2;
import org.apache.kylin.storage.hbase.steps.RowValueDecoder;
import org.apache.kylin.storage.translate.HBaseKeyRange;
import org.apache.kylin.storage.tuple.Tuple;
@@ -212,13 +213,20 @@ public class CubeSegmentTupleIterator implements ITupleIterator {
private void applyFuzzyFilter(Scan scan, HBaseKeyRange keyRange) {
List<org.apache.kylin.common.util.Pair<byte[], byte[]>> fuzzyKeys = keyRange.getFuzzyKeys();
if (fuzzyKeys != null && fuzzyKeys.size() > 0) {
- FuzzyRowFilter rowFilter = new FuzzyRowFilter(convertToHBasePair(fuzzyKeys));
+
+ boolean useFuzzyRowFilterV2 = true;
+ Filter fuzzyFilter = null;
+ if (useFuzzyRowFilterV2) {
+ fuzzyFilter = new FuzzyRowFilterV2(convertToHBasePair(fuzzyKeys));
+ } else {
+ fuzzyFilter = new FuzzyRowFilter(convertToHBasePair(fuzzyKeys));
+ }
Filter filter = scan.getFilter();
if (filter != null) {
throw new RuntimeException("Scan filter not empty : " + filter);
} else {
- scan.setFilter(rowFilter);
+ scan.setFilter(fuzzyFilter);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/f1dfe849/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/filter/FuzzyRowFilterV2.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/filter/FuzzyRowFilterV2.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/filter/FuzzyRowFilterV2.java
index 0877058..eeb2eab 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/filter/FuzzyRowFilterV2.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/filter/FuzzyRowFilterV2.java
@@ -18,13 +18,15 @@
package org.apache.kylin.storage.hbase.cube.v1.filter;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.PriorityQueue;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.protobuf.InvalidProtocolBufferException;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.filter.Filter;
@@ -34,13 +36,8 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.kylin.storage.hbase.cube.v1.filter.generated.FilterProtosExt;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.List;
-import java.util.PriorityQueue;
-
-
+import com.google.common.annotations.VisibleForTesting;
+import com.google.protobuf.InvalidProtocolBufferException;
/**
* This is optimized version of a standard FuzzyRowFilter Filters data based on fuzzy row key.
@@ -83,9 +80,7 @@ public class FuzzyRowFilterV2 extends FilterBase {
for (int i = 0; i < fuzzyKeysData.size(); i++) {
p = fuzzyKeysData.get(i);
if (p.getFirst().length != p.getSecond().length) {
- Pair<String, String> readable =
- new Pair<String, String>(Bytes.toStringBinary(p.getFirst()), Bytes.toStringBinary(p
- .getSecond()));
+ Pair<String, String> readable = new Pair<String, String>(Bytes.toStringBinary(p.getFirst()), Bytes.toStringBinary(p.getSecond()));
throw new IllegalArgumentException("Fuzzy pair lengths do not match: " + readable);
}
// update mask ( 0 -> -1 (0xff), 1 -> 0)
@@ -105,7 +100,8 @@ public class FuzzyRowFilterV2 extends FilterBase {
byte[] mask = p.getSecond();
for (int i = 0; i < mask.length; i++) {
// set non-fixed part of a search key to 0.
- if (mask[i] == 0) key[i] = 0;
+ if (mask[i] == 0)
+ key[i] = 0;
}
}
@@ -120,7 +116,8 @@ public class FuzzyRowFilterV2 extends FilterBase {
// do nothing
return mask;
}
- if (isPreprocessedMask(mask)) return mask;
+ if (isPreprocessedMask(mask))
+ return mask;
for (int i = 0; i < mask.length; i++) {
if (mask[i] == 0) {
mask[i] = -1; // 0 -> -1
@@ -147,9 +144,7 @@ public class FuzzyRowFilterV2 extends FilterBase {
for (int i = startIndex; i < size + startIndex; i++) {
final int index = i % size;
Pair<byte[], byte[]> fuzzyData = fuzzyKeysData.get(index);
- SatisfiesCode satisfiesCode =
- satisfies(isReversed(), c.getRowArray(), c.getRowOffset(), c.getRowLength(),
- fuzzyData.getFirst(), fuzzyData.getSecond());
+ SatisfiesCode satisfiesCode = satisfies(isReversed(), c.getRowArray(), c.getRowOffset(), c.getRowLength(), fuzzyData.getFirst(), fuzzyData.getSecond());
if (satisfiesCode == SatisfiesCode.YES) {
lastFoundIndex = index;
return ReturnCode.INCLUDE;
@@ -185,26 +180,22 @@ public class FuzzyRowFilterV2 extends FilterBase {
private boolean initialized = false;
RowTracker() {
- nextRows =
- new PriorityQueue<Pair<byte[], Pair<byte[], byte[]>>>(fuzzyKeysData.size(),
- new Comparator<Pair<byte[], Pair<byte[], byte[]>>>() {
- @Override
- public int compare(Pair<byte[], Pair<byte[], byte[]>> o1,
- Pair<byte[], Pair<byte[], byte[]>> o2) {
- int compare = Bytes.compareTo(o1.getFirst(), o2.getFirst());
- if (!isReversed()) {
- return compare;
- } else {
- return -compare;
- }
- }
- });
+ nextRows = new PriorityQueue<Pair<byte[], Pair<byte[], byte[]>>>(fuzzyKeysData.size(), new Comparator<Pair<byte[], Pair<byte[], byte[]>>>() {
+ @Override
+ public int compare(Pair<byte[], Pair<byte[], byte[]>> o1, Pair<byte[], Pair<byte[], byte[]>> o2) {
+ int compare = Bytes.compareTo(o1.getFirst(), o2.getFirst());
+ if (!isReversed()) {
+ return compare;
+ } else {
+ return -compare;
+ }
+ }
+ });
}
byte[] nextRow() {
if (nextRows.isEmpty()) {
- throw new IllegalStateException(
- "NextRows should not be empty, make sure to call nextRow() after updateTracker() return true");
+ throw new IllegalStateException("NextRows should not be empty, make sure to call nextRow() after updateTracker() return true");
} else {
return nextRows.peek().getFirst();
}
@@ -227,16 +218,12 @@ public class FuzzyRowFilterV2 extends FilterBase {
}
boolean lessThan(Cell currentCell, byte[] nextRowKey) {
- int compareResult =
- Bytes.compareTo(nextRowKey, 0, nextRowKey.length, currentCell.getRowArray(),
- currentCell.getRowOffset(), currentCell.getRowLength());
+ int compareResult = Bytes.compareTo(currentCell.getRowArray(), currentCell.getRowOffset(), currentCell.getRowLength(), nextRowKey, 0, nextRowKey.length);
return (!isReversed() && compareResult < 0) || (isReversed() && compareResult > 0);
}
void updateWith(Cell currentCell, Pair<byte[], byte[]> fuzzyData) {
- byte[] nextRowKeyCandidate =
- getNextForFuzzyRule(isReversed(), currentCell.getRowArray(), currentCell.getRowOffset(),
- currentCell.getRowLength(), fuzzyData.getFirst(), fuzzyData.getSecond());
+ byte[] nextRowKeyCandidate = getNextForFuzzyRule(isReversed(), currentCell.getRowArray(), currentCell.getRowOffset(), currentCell.getRowLength(), fuzzyData.getFirst(), fuzzyData.getSecond());
if (nextRowKeyCandidate != null) {
nextRows.add(new Pair<byte[], Pair<byte[], byte[]>>(nextRowKeyCandidate, fuzzyData));
}
@@ -311,13 +298,11 @@ public class FuzzyRowFilterV2 extends FilterBase {
}
@VisibleForTesting
- static SatisfiesCode satisfies(boolean reverse, byte[] row, byte[] fuzzyKeyBytes,
- byte[] fuzzyKeyMeta) {
+ static SatisfiesCode satisfies(boolean reverse, byte[] row, byte[] fuzzyKeyBytes, byte[] fuzzyKeyMeta) {
return satisfies(reverse, row, 0, row.length, fuzzyKeyBytes, fuzzyKeyMeta);
}
- static SatisfiesCode satisfies(boolean reverse, byte[] row, int offset, int length,
- byte[] fuzzyKeyBytes, byte[] fuzzyKeyMeta) {
+ static SatisfiesCode satisfies(boolean reverse, byte[] row, int offset, int length, byte[] fuzzyKeyBytes, byte[] fuzzyKeyMeta) {
if (UnsafeAccess.isAvailable() == false) {
return satisfiesNoUnsafe(reverse, row, offset, length, fuzzyKeyBytes, fuzzyKeyMeta);
@@ -380,8 +365,7 @@ public class FuzzyRowFilterV2 extends FilterBase {
return SatisfiesCode.YES;
}
- static SatisfiesCode satisfiesNoUnsafe(boolean reverse, byte[] row, int offset, int length,
- byte[] fuzzyKeyBytes, byte[] fuzzyKeyMeta) {
+ static SatisfiesCode satisfiesNoUnsafe(boolean reverse, byte[] row, int offset, int length, byte[] fuzzyKeyBytes, byte[] fuzzyKeyMeta) {
if (row == null) {
// do nothing, let scan to proceed
return SatisfiesCode.YES;
@@ -432,8 +416,7 @@ public class FuzzyRowFilterV2 extends FilterBase {
}
@VisibleForTesting
- static byte[] getNextForFuzzyRule(boolean reverse, byte[] row, byte[] fuzzyKeyBytes,
- byte[] fuzzyKeyMeta) {
+ static byte[] getNextForFuzzyRule(boolean reverse, byte[] row, byte[] fuzzyKeyBytes, byte[] fuzzyKeyMeta) {
return getNextForFuzzyRule(reverse, row, 0, row.length, fuzzyKeyBytes, fuzzyKeyMeta);
}
@@ -509,8 +492,7 @@ public class FuzzyRowFilterV2 extends FilterBase {
* otherwise
*/
@VisibleForTesting
- static byte[] getNextForFuzzyRule(boolean reverse, byte[] row, int offset, int length,
- byte[] fuzzyKeyBytes, byte[] fuzzyKeyMeta) {
+ static byte[] getNextForFuzzyRule(boolean reverse, byte[] row, int offset, int length, byte[] fuzzyKeyBytes, byte[] fuzzyKeyMeta) {
// To find out the next "smallest" byte array that satisfies fuzzy rule and "greater" than
// the given one we do the following:
// 1. setting values on all "fixed" positions to the values from fuzzyKeyBytes
@@ -519,8 +501,7 @@ public class FuzzyRowFilterV2 extends FilterBase {
// It is easier to perform this by using fuzzyKeyBytes copy and setting "non-fixed" position
// values than otherwise.
- byte[] result =
- Arrays.copyOf(fuzzyKeyBytes, length > fuzzyKeyBytes.length ? length : fuzzyKeyBytes.length);
+ byte[] result = Arrays.copyOf(fuzzyKeyBytes, length > fuzzyKeyBytes.length ? length : fuzzyKeyBytes.length);
if (reverse && length > fuzzyKeyBytes.length) {
// we need trailing 0xff's instead of trailing 0x00's
for (int i = fuzzyKeyBytes.length; i < result.length; i++) {
@@ -578,20 +559,21 @@ public class FuzzyRowFilterV2 extends FilterBase {
* corresponding fields in other. Used for testing.
*/
boolean areSerializedFieldsEqual(Filter o) {
- if (o == this) return true;
- if (!(o instanceof FuzzyRowFilterV2)) return false;
+ if (o == this)
+ return true;
+ if (!(o instanceof FuzzyRowFilterV2))
+ return false;
FuzzyRowFilterV2 other = (FuzzyRowFilterV2) o;
- if (this.fuzzyKeysData.size() != other.fuzzyKeysData.size()) return false;
+ if (this.fuzzyKeysData.size() != other.fuzzyKeysData.size())
+ return false;
for (int i = 0; i < fuzzyKeysData.size(); ++i) {
Pair<byte[], byte[]> thisData = this.fuzzyKeysData.get(i);
Pair<byte[], byte[]> otherData = other.fuzzyKeysData.get(i);
- if (!(Bytes.equals(thisData.getFirst(), otherData.getFirst()) && Bytes.equals(
- thisData.getSecond(), otherData.getSecond()))) {
+ if (!(Bytes.equals(thisData.getFirst(), otherData.getFirst()) && Bytes.equals(thisData.getSecond(), otherData.getSecond()))) {
return false;
}
}
return true;
}
}
-
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/f1dfe849/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/filter/TestFuzzyRowFilterEndToEnd.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/filter/TestFuzzyRowFilterEndToEnd.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/filter/TestFuzzyRowFilterEndToEnd.java
deleted file mode 100644
index 99797a9..0000000
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/filter/TestFuzzyRowFilterEndToEnd.java
+++ /dev/null
@@ -1,352 +0,0 @@
-/*
- * 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.kylin.storage.hbase.cube.v1.filter;
-
-import static org.junit.Assert.assertEquals;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.HBaseTestingUtility;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.Durability;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.filter.Filter;
-import org.apache.hadoop.hbase.filter.FilterList;
-import org.apache.hadoop.hbase.filter.FilterList.Operator;
-import org.apache.hadoop.hbase.regionserver.ConstantSizeRegionSplitPolicy;
-import org.apache.hadoop.hbase.regionserver.HRegion;
-import org.apache.hadoop.hbase.regionserver.RegionScanner;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.Pair;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-import com.google.common.collect.Lists;
-
-public class TestFuzzyRowFilterEndToEnd {
- private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
- private final static byte fuzzyValue = (byte) 63;
- private static final Log LOG = LogFactory.getLog(TestFuzzyRowFilterEndToEnd.class);
-
- private static int firstPartCardinality = 50;
- private static int secondPartCardinality = 50;
- private static int thirdPartCardinality = 50;
- private static int colQualifiersTotal = 5;
- private static int totalFuzzyKeys = thirdPartCardinality / 2;
-
- private static String table = "TestFuzzyRowFilterEndToEnd";
-
- /**
- * @throws java.lang.Exception
- */
- @BeforeClass
- public static void setUpBeforeClass() throws Exception {
- Configuration conf = TEST_UTIL.getConfiguration();
- conf.setInt("hbase.client.scanner.caching", 1000);
- conf.set(HConstants.HBASE_REGION_SPLIT_POLICY_KEY,
- ConstantSizeRegionSplitPolicy.class.getName());
- // set no splits
- conf.setLong(HConstants.HREGION_MAX_FILESIZE, ((long) 1024) * 1024 * 1024 * 10);
-
- TEST_UTIL.startMiniCluster();
- }
-
- /**
- * @throws java.lang.Exception
- */
- @AfterClass
- public static void tearDownAfterClass() throws Exception {
- TEST_UTIL.shutdownMiniCluster();
- }
-
- /**
- * @throws java.lang.Exception
- */
- @Before
- public void setUp() throws Exception {
- // Nothing to do.
- }
-
- /**
- * @throws java.lang.Exception
- */
- @After
- public void tearDown() throws Exception {
- // Nothing to do.
- }
-
- @Test
- public void testEndToEnd() throws Exception {
- String cf = "f";
-
- HTable ht =
- TEST_UTIL.createTable(TableName.valueOf(table), Bytes.toBytes(cf), Integer.MAX_VALUE);
-
- // 10 byte row key - (2 bytes 4 bytes 4 bytes)
- // 4 byte qualifier
- // 4 byte value
-
- for (int i0 = 0; i0 < firstPartCardinality; i0++) {
-
- for (int i1 = 0; i1 < secondPartCardinality; i1++) {
-
- for (int i2 = 0; i2 < thirdPartCardinality; i2++) {
- byte[] rk = new byte[10];
-
- ByteBuffer buf = ByteBuffer.wrap(rk);
- buf.clear();
- buf.putShort((short) i0);
- buf.putInt(i1);
- buf.putInt(i2);
- for (int c = 0; c < colQualifiersTotal; c++) {
- byte[] cq = new byte[4];
- Bytes.putBytes(cq, 0, Bytes.toBytes(c), 0, 4);
-
- Put p = new Put(rk);
- p.setDurability(Durability.SKIP_WAL);
- p.add(cf.getBytes(), cq, Bytes.toBytes(c));
- ht.put(p);
- }
- }
- }
- }
-
- TEST_UTIL.flush();
-
- // test passes
- runTest1(ht);
- runTest2(ht);
-
- }
-
- private void runTest1(HTable hTable) throws IOException {
- // [0, 2, ?, ?, ?, ?, 0, 0, 0, 1]
-
- byte[] mask = new byte[] { 0, 0, 1, 1, 1, 1, 0, 0, 0, 0 };
-
- List<Pair<byte[], byte[]>> list = new ArrayList<Pair<byte[], byte[]>>();
- for (int i = 0; i < totalFuzzyKeys; i++) {
- byte[] fuzzyKey = new byte[10];
- ByteBuffer buf = ByteBuffer.wrap(fuzzyKey);
- buf.clear();
- buf.putShort((short) 2);
- for (int j = 0; j < 4; j++) {
- buf.put(fuzzyValue);
- }
- buf.putInt(i);
-
- Pair<byte[], byte[]> pair = new Pair<byte[], byte[]>(fuzzyKey, mask);
- list.add(pair);
- }
-
- int expectedSize = secondPartCardinality * totalFuzzyKeys * colQualifiersTotal;
- FuzzyRowFilterV2 fuzzyRowFilter0 = new FuzzyRowFilterV2(list);
- // Filters are not stateless - we can't reuse them
- FuzzyRowFilterV2 fuzzyRowFilter1 = new FuzzyRowFilterV2(list);
-
- // regular test
- runScanner(hTable, expectedSize, fuzzyRowFilter0);
- // optimized from block cache
- runScanner(hTable, expectedSize, fuzzyRowFilter1);
-
- }
-
- private void runTest2(HTable hTable) throws IOException {
- // [0, 0, ?, ?, ?, ?, 0, 0, 0, 0] , [0, 1, ?, ?, ?, ?, 0, 0, 0, 1]...
-
- byte[] mask = new byte[] { 0, 0, 1, 1, 1, 1, 0, 0, 0, 0 };
-
- List<Pair<byte[], byte[]>> list = new ArrayList<Pair<byte[], byte[]>>();
-
- for (int i = 0; i < totalFuzzyKeys; i++) {
- byte[] fuzzyKey = new byte[10];
- ByteBuffer buf = ByteBuffer.wrap(fuzzyKey);
- buf.clear();
- buf.putShort((short) (i * 2));
- for (int j = 0; j < 4; j++) {
- buf.put(fuzzyValue);
- }
- buf.putInt(i * 2);
-
- Pair<byte[], byte[]> pair = new Pair<byte[], byte[]>(fuzzyKey, mask);
- list.add(pair);
- }
-
- int expectedSize = totalFuzzyKeys * secondPartCardinality * colQualifiersTotal;
-
- FuzzyRowFilterV2 fuzzyRowFilter0 = new FuzzyRowFilterV2(list);
- // Filters are not stateless - we can't reuse them
- FuzzyRowFilterV2 fuzzyRowFilter1 = new FuzzyRowFilterV2(list);
-
- // regular test
- runScanner(hTable, expectedSize, fuzzyRowFilter0);
- // optimized from block cache
- runScanner(hTable, expectedSize, fuzzyRowFilter1);
-
- }
-
- private void runScanner(HTable hTable, int expectedSize, Filter filter) throws IOException {
-
- String cf = "f";
- Scan scan = new Scan();
- scan.addFamily(cf.getBytes());
- scan.setFilter(filter);
- List<HRegion> regions = TEST_UTIL.getHBaseCluster().getRegions(table.getBytes());
- HRegion first = regions.get(0);
- first.getScanner(scan);
- RegionScanner scanner = first.getScanner(scan);
- List<Cell> results = new ArrayList<Cell>();
- // Result result;
- long timeBeforeScan = System.currentTimeMillis();
- int found = 0;
- while (scanner.next(results)) {
- found += results.size();
- results.clear();
- }
- found += results.size();
- long scanTime = System.currentTimeMillis() - timeBeforeScan;
- scanner.close();
-
- LOG.info("\nscan time = " + scanTime + "ms");
- LOG.info("found " + found + " results\n");
-
- assertEquals(expectedSize, found);
- }
-
- @SuppressWarnings("deprecation")
- @Test
- public void testFilterList() throws Exception {
- String cf = "f";
- String table = "TestFuzzyRowFiltersInFilterList";
- HTable ht =
- TEST_UTIL.createTable(TableName.valueOf(table), Bytes.toBytes(cf), Integer.MAX_VALUE);
-
- // 10 byte row key - (2 bytes 4 bytes 4 bytes)
- // 4 byte qualifier
- // 4 byte value
-
- for (int i1 = 0; i1 < 5; i1++) {
- for (int i2 = 0; i2 < 5; i2++) {
- byte[] rk = new byte[10];
-
- ByteBuffer buf = ByteBuffer.wrap(rk);
- buf.clear();
- buf.putShort((short) 2);
- buf.putInt(i1);
- buf.putInt(i2);
-
- // Each row contains 5 columns
- for (int c = 0; c < 5; c++) {
- byte[] cq = new byte[4];
- Bytes.putBytes(cq, 0, Bytes.toBytes(c), 0, 4);
-
- Put p = new Put(rk);
- p.setDurability(Durability.SKIP_WAL);
- p.add(cf.getBytes(), cq, Bytes.toBytes(c));
- ht.put(p);
- LOG.info("Inserting: rk: " + Bytes.toStringBinary(rk) + " cq: "
- + Bytes.toStringBinary(cq));
- }
- }
- }
-
- TEST_UTIL.flush();
-
- // test passes if we get back 5 KV's (1 row)
- runTest(ht, 5);
-
- }
-
- @SuppressWarnings("unchecked")
- private void runTest(HTable hTable, int expectedSize) throws IOException {
- // [0, 2, ?, ?, ?, ?, 0, 0, 0, 1]
- byte[] fuzzyKey1 = new byte[10];
- ByteBuffer buf = ByteBuffer.wrap(fuzzyKey1);
- buf.clear();
- buf.putShort((short) 2);
- for (int i = 0; i < 4; i++)
- buf.put(fuzzyValue);
- buf.putInt((short) 1);
- byte[] mask1 = new byte[] { 0, 0, 1, 1, 1, 1, 0, 0, 0, 0 };
-
- byte[] fuzzyKey2 = new byte[10];
- buf = ByteBuffer.wrap(fuzzyKey2);
- buf.clear();
- buf.putShort((short) 2);
- buf.putInt((short) 2);
- for (int i = 0; i < 4; i++)
- buf.put(fuzzyValue);
-
- byte[] mask2 = new byte[] { 0, 0, 0, 0, 0, 0, 1, 1, 1, 1 };
-
- Pair<byte[], byte[]> pair1 = new Pair<byte[], byte[]>(fuzzyKey1, mask1);
- Pair<byte[], byte[]> pair2 = new Pair<byte[], byte[]>(fuzzyKey2, mask2);
-
- FuzzyRowFilterV2 fuzzyRowFilter1 = new FuzzyRowFilterV2(Lists.newArrayList(pair1));
- FuzzyRowFilterV2 fuzzyRowFilter2 = new FuzzyRowFilterV2(Lists.newArrayList(pair2));
- // regular test - we expect 1 row back (5 KVs)
- runScanner(hTable, expectedSize, fuzzyRowFilter1, fuzzyRowFilter2);
- }
-
- private void runScanner(HTable hTable, int expectedSize, Filter filter1, Filter filter2)
- throws IOException {
- String cf = "f";
- Scan scan = new Scan();
- scan.addFamily(cf.getBytes());
- FilterList filterList = new FilterList(Operator.MUST_PASS_ALL, filter1, filter2);
- scan.setFilter(filterList);
-
- ResultScanner scanner = hTable.getScanner(scan);
- List<Cell> results = new ArrayList<Cell>();
- Result result;
- long timeBeforeScan = System.currentTimeMillis();
- while ((result = scanner.next()) != null) {
- for (Cell kv : result.listCells()) {
- LOG.info("Got rk: " + Bytes.toStringBinary(CellUtil.cloneRow(kv)) + " cq: "
- + Bytes.toStringBinary(CellUtil.cloneQualifier(kv)));
- results.add(kv);
- }
- }
- long scanTime = System.currentTimeMillis() - timeBeforeScan;
- scanner.close();
-
- LOG.info("scan time = " + scanTime + "ms");
- LOG.info("found " + results.size() + " results");
-
- assertEquals(expectedSize, results.size());
- }
-}
-
-
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/f1dfe849/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/filter/TestFuzzyRowFilterV2EndToEnd.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/filter/TestFuzzyRowFilterV2EndToEnd.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/filter/TestFuzzyRowFilterV2EndToEnd.java
new file mode 100644
index 0000000..7d45fd6
--- /dev/null
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/filter/TestFuzzyRowFilterV2EndToEnd.java
@@ -0,0 +1,352 @@
+/*
+ * 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.kylin.storage.hbase.cube.v1.filter;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.FilterList;
+import org.apache.hadoop.hbase.filter.FilterList.Operator;
+import org.apache.hadoop.hbase.regionserver.ConstantSizeRegionSplitPolicy;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+public class TestFuzzyRowFilterV2EndToEnd {
+ private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+ private final static byte fuzzyValue = (byte) 63;
+ private static final Log LOG = LogFactory.getLog(TestFuzzyRowFilterV2EndToEnd.class);
+
+ private static int firstPartCardinality = 4;
+ private static int secondPartCardinality = 4;
+ private static int thirdPartCardinality = 4;
+ private static int colQualifiersTotal = 1;
+ private static int totalFuzzyKeys = thirdPartCardinality / 2;
+
+ private static String table = "TestFuzzyRowFilterEndToEnd";
+
+ /**
+ * @throws java.lang.Exception
+ */
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ Configuration conf = TEST_UTIL.getConfiguration();
+ conf.setInt("hbase.client.scanner.caching", 1000);
+ conf.set(HConstants.HBASE_REGION_SPLIT_POLICY_KEY,
+ ConstantSizeRegionSplitPolicy.class.getName());
+ // set no splits
+ conf.setLong(HConstants.HREGION_MAX_FILESIZE, ((long) 1024) * 1024 * 1024 * 10);
+
+ TEST_UTIL.startMiniCluster();
+ }
+
+ /**
+ * @throws java.lang.Exception
+ */
+ @AfterClass
+ public static void tearDownAfterClass() throws Exception {
+ TEST_UTIL.shutdownMiniCluster();
+ }
+
+ /**
+ * @throws java.lang.Exception
+ */
+ @Before
+ public void setUp() throws Exception {
+ // Nothing to do.
+ }
+
+ /**
+ * @throws java.lang.Exception
+ */
+ @After
+ public void tearDown() throws Exception {
+ // Nothing to do.
+ }
+
+ @Test
+ public void testEndToEnd() throws Exception {
+ String cf = "f";
+
+ HTable ht =
+ TEST_UTIL.createTable(TableName.valueOf(table), Bytes.toBytes(cf), Integer.MAX_VALUE);
+
+ // 10 byte row key - (2 bytes 4 bytes 4 bytes)
+ // 4 byte qualifier
+ // 4 byte value
+
+ for (int i0 = 0; i0 < firstPartCardinality; i0++) {
+
+ for (int i1 = 0; i1 < secondPartCardinality; i1++) {
+
+ for (int i2 = 0; i2 < thirdPartCardinality; i2++) {
+ byte[] rk = new byte[10];
+
+ ByteBuffer buf = ByteBuffer.wrap(rk);
+ buf.clear();
+ buf.putShort((short) i0);
+ buf.putInt(i1);
+ buf.putInt(i2);
+ for (int c = 0; c < colQualifiersTotal; c++) {
+ byte[] cq = new byte[4];
+ Bytes.putBytes(cq, 0, Bytes.toBytes(c), 0, 4);
+
+ Put p = new Put(rk);
+ p.setDurability(Durability.SKIP_WAL);
+ p.add(cf.getBytes(), cq, Bytes.toBytes(c));
+ ht.put(p);
+ }
+ }
+ }
+ }
+
+ TEST_UTIL.flush();
+
+ // test passes
+ runTest1(ht);
+ runTest2(ht);
+
+ }
+
+ private void runTest1(HTable hTable) throws IOException {
+ // [0, 2, ?, ?, ?, ?, 0, 0, 0, 1]
+
+ byte[] mask = new byte[] { 0, 0, 1, 1, 1, 1, 0, 0, 0, 0 };
+
+ List<Pair<byte[], byte[]>> list = new ArrayList<Pair<byte[], byte[]>>();
+ for (int i = 0; i < totalFuzzyKeys; i++) {
+ byte[] fuzzyKey = new byte[10];
+ ByteBuffer buf = ByteBuffer.wrap(fuzzyKey);
+ buf.clear();
+ buf.putShort((short) 2);
+ for (int j = 0; j < 4; j++) {
+ buf.put(fuzzyValue);
+ }
+ buf.putInt(i);
+
+ Pair<byte[], byte[]> pair = new Pair<byte[], byte[]>(fuzzyKey, mask);
+ list.add(pair);
+ }
+
+ int expectedSize = secondPartCardinality * totalFuzzyKeys * colQualifiersTotal;
+ FuzzyRowFilterV2 fuzzyRowFilter0 = new FuzzyRowFilterV2(list);
+ // Filters are not stateless - we can't reuse them
+ FuzzyRowFilterV2 fuzzyRowFilter1 = new FuzzyRowFilterV2(list);
+
+ // regular test
+ runScanner(hTable, expectedSize, fuzzyRowFilter0);
+ // optimized from block cache
+ runScanner(hTable, expectedSize, fuzzyRowFilter1);
+
+ }
+
+ private void runTest2(HTable hTable) throws IOException {
+ // [0, 0, ?, ?, ?, ?, 0, 0, 0, 0] , [0, 1, ?, ?, ?, ?, 0, 0, 0, 1]...
+
+ byte[] mask = new byte[] { 0, 0, 1, 1, 1, 1, 0, 0, 0, 0 };
+
+ List<Pair<byte[], byte[]>> list = new ArrayList<Pair<byte[], byte[]>>();
+
+ for (int i = 0; i < totalFuzzyKeys; i++) {
+ byte[] fuzzyKey = new byte[10];
+ ByteBuffer buf = ByteBuffer.wrap(fuzzyKey);
+ buf.clear();
+ buf.putShort((short) (i * 2));
+ for (int j = 0; j < 4; j++) {
+ buf.put(fuzzyValue);
+ }
+ buf.putInt(i * 2);
+
+ Pair<byte[], byte[]> pair = new Pair<byte[], byte[]>(fuzzyKey, mask);
+ list.add(pair);
+ }
+
+ int expectedSize = totalFuzzyKeys * secondPartCardinality * colQualifiersTotal;
+
+ FuzzyRowFilterV2 fuzzyRowFilter0 = new FuzzyRowFilterV2(list);
+ // Filters are not stateless - we can't reuse them
+ FuzzyRowFilterV2 fuzzyRowFilter1 = new FuzzyRowFilterV2(list);
+
+ // regular test
+ runScanner(hTable, expectedSize, fuzzyRowFilter0);
+ // optimized from block cache
+ runScanner(hTable, expectedSize, fuzzyRowFilter1);
+
+ }
+
+ private void runScanner(HTable hTable, int expectedSize, Filter filter) throws IOException {
+
+ String cf = "f";
+ Scan scan = new Scan();
+ scan.addFamily(cf.getBytes());
+ scan.setFilter(filter);
+ List<HRegion> regions = TEST_UTIL.getHBaseCluster().getRegions(table.getBytes());
+ HRegion first = regions.get(0);
+ first.getScanner(scan);
+ RegionScanner scanner = first.getScanner(scan);
+ List<Cell> results = new ArrayList<Cell>();
+ // Result result;
+ long timeBeforeScan = System.currentTimeMillis();
+ int found = 0;
+ while (scanner.next(results)) {
+ found += results.size();
+ results.clear();
+ }
+ found += results.size();
+ long scanTime = System.currentTimeMillis() - timeBeforeScan;
+ scanner.close();
+
+ LOG.info("\nscan time = " + scanTime + "ms");
+ LOG.info("found " + found + " results\n");
+
+ assertEquals(expectedSize, found);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testFilterList() throws Exception {
+ String cf = "f";
+ String table = "TestFuzzyRowFiltersInFilterList";
+ HTable ht =
+ TEST_UTIL.createTable(TableName.valueOf(table), Bytes.toBytes(cf), Integer.MAX_VALUE);
+
+ // 10 byte row key - (2 bytes 4 bytes 4 bytes)
+ // 4 byte qualifier
+ // 4 byte value
+
+ for (int i1 = 0; i1 < 5; i1++) {
+ for (int i2 = 0; i2 < 5; i2++) {
+ byte[] rk = new byte[10];
+
+ ByteBuffer buf = ByteBuffer.wrap(rk);
+ buf.clear();
+ buf.putShort((short) 2);
+ buf.putInt(i1);
+ buf.putInt(i2);
+
+ // Each row contains 5 columns
+ for (int c = 0; c < 5; c++) {
+ byte[] cq = new byte[4];
+ Bytes.putBytes(cq, 0, Bytes.toBytes(c), 0, 4);
+
+ Put p = new Put(rk);
+ p.setDurability(Durability.SKIP_WAL);
+ p.add(cf.getBytes(), cq, Bytes.toBytes(c));
+ ht.put(p);
+ LOG.info("Inserting: rk: " + Bytes.toStringBinary(rk) + " cq: "
+ + Bytes.toStringBinary(cq));
+ }
+ }
+ }
+
+ TEST_UTIL.flush();
+
+ // test passes if we get back 5 KV's (1 row)
+ runTest(ht, 5);
+
+ }
+
+ @SuppressWarnings("unchecked")
+ private void runTest(HTable hTable, int expectedSize) throws IOException {
+ // [0, 2, ?, ?, ?, ?, 0, 0, 0, 1]
+ byte[] fuzzyKey1 = new byte[10];
+ ByteBuffer buf = ByteBuffer.wrap(fuzzyKey1);
+ buf.clear();
+ buf.putShort((short) 2);
+ for (int i = 0; i < 4; i++)
+ buf.put(fuzzyValue);
+ buf.putInt((short) 1);
+ byte[] mask1 = new byte[] { 0, 0, 1, 1, 1, 1, 0, 0, 0, 0 };
+
+ byte[] fuzzyKey2 = new byte[10];
+ buf = ByteBuffer.wrap(fuzzyKey2);
+ buf.clear();
+ buf.putShort((short) 2);
+ buf.putInt((short) 2);
+ for (int i = 0; i < 4; i++)
+ buf.put(fuzzyValue);
+
+ byte[] mask2 = new byte[] { 0, 0, 0, 0, 0, 0, 1, 1, 1, 1 };
+
+ Pair<byte[], byte[]> pair1 = new Pair<byte[], byte[]>(fuzzyKey1, mask1);
+ Pair<byte[], byte[]> pair2 = new Pair<byte[], byte[]>(fuzzyKey2, mask2);
+
+ FuzzyRowFilterV2 fuzzyRowFilter1 = new FuzzyRowFilterV2(Lists.newArrayList(pair1));
+ FuzzyRowFilterV2 fuzzyRowFilter2 = new FuzzyRowFilterV2(Lists.newArrayList(pair2));
+ // regular test - we expect 1 row back (5 KVs)
+ runScanner(hTable, expectedSize, fuzzyRowFilter1, fuzzyRowFilter2);
+ }
+
+ private void runScanner(HTable hTable, int expectedSize, Filter filter1, Filter filter2)
+ throws IOException {
+ String cf = "f";
+ Scan scan = new Scan();
+ scan.addFamily(cf.getBytes());
+ FilterList filterList = new FilterList(Operator.MUST_PASS_ALL, filter1, filter2);
+ scan.setFilter(filterList);
+
+ ResultScanner scanner = hTable.getScanner(scan);
+ List<Cell> results = new ArrayList<Cell>();
+ Result result;
+ long timeBeforeScan = System.currentTimeMillis();
+ while ((result = scanner.next()) != null) {
+ for (Cell kv : result.listCells()) {
+ LOG.info("Got rk: " + Bytes.toStringBinary(CellUtil.cloneRow(kv)) + " cq: "
+ + Bytes.toStringBinary(CellUtil.cloneQualifier(kv)));
+ results.add(kv);
+ }
+ }
+ long scanTime = System.currentTimeMillis() - timeBeforeScan;
+ scanner.close();
+
+ LOG.info("scan time = " + scanTime + "ms");
+ LOG.info("found " + results.size() + " results");
+
+ assertEquals(expectedSize, results.size());
+ }
+}
+
+