You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2016/05/12 07:12:50 UTC
[09/28] kylin git commit: refactor
refactor
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/2fc706ca
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/2fc706ca
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/2fc706ca
Branch: refs/heads/1.5.x-HBase1.x
Commit: 2fc706ca132dfad43b2f68c41757ddfbadc595b0
Parents: 656c645
Author: Hongbin Ma <ma...@apache.org>
Authored: Mon May 9 18:38:13 2016 +0800
Committer: Hongbin Ma <ma...@apache.org>
Committed: Mon May 9 18:40:06 2016 +0800
----------------------------------------------------------------------
build/conf/kylin.properties | 3 +-
.../apache/kylin/common/KylinConfigBase.java | 14 +++-
.../kylin/common/debug/BackdoorToggles.java | 24 ++++++
.../kylin/common/restclient/RestClient.java | 16 ++--
.../common/util/LoggableCachedThreadPool.java | 57 ++++++++++++++
.../kylin/common/util/NumberIterators.java | 34 ++++++++
.../apache/kylin/gridtable/ScannerWorker.java | 3 +-
.../apache/kylin/storage/cache/EhcacheTest.java | 49 ++++++++----
.../apache/kylin/rest/response/SQLResponse.java | 26 +++----
.../hbase/cube/v2/CubeHBaseEndpointRPC.java | 82 +++++++-------------
10 files changed, 213 insertions(+), 95 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/2fc706ca/build/conf/kylin.properties
----------------------------------------------------------------------
diff --git a/build/conf/kylin.properties b/build/conf/kylin.properties
index 80ae004..9a2fd81 100644
--- a/build/conf/kylin.properties
+++ b/build/conf/kylin.properties
@@ -168,4 +168,5 @@ kylin.web.hive.limit=20
# E.g, for cube whose capacity be marked as "SMALL", split region per 5GB by default
kylin.hbase.region.cut.small=5
kylin.hbase.region.cut.medium=10
-kylin.hbase.region.cut.large=50
\ No newline at end of file
+kylin.hbase.region.cut.large=50
+
http://git-wip-us.apache.org/repos/asf/kylin/blob/2fc706ca/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
index 850ea17..3db97b3 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -361,8 +361,13 @@ abstract public class KylinConfigBase implements Serializable {
return getOptionalStringArray("kylin.rest.servers", new String[0]);
}
- public String[] getAdminDls() {
- return getOptionalStringArray("kylin.job.admin.dls", null);
+ public int getWorkersPerServer() {
+ //for sequence sql use
+ return Integer.parseInt(getOptional("kylin.rest.workers.per.server", "1"));
+ }
+
+ public String getAdminDls() {
+ return getOptional("kylin.job.admin.dls", null);
}
public long getJobStepTimeout() {
@@ -498,6 +503,10 @@ abstract public class KylinConfigBase implements Serializable {
return Integer.parseInt(this.getOptional("kylin.hbase.client.keyvalue.maxsize", "10485760"));
}
+ public String getDefaultIGTStorage() {
+ return getOptional("kylin.query.storage.default.gtstorage", "org.apache.kylin.storage.hbase.cube.v2.CubeHBaseEndpointRPC");
+ }
+
public int getHBaseScanCacheRows() {
return Integer.parseInt(this.getOptional("kylin.hbase.scan.cache_rows", "1024"));
}
@@ -632,4 +641,5 @@ abstract public class KylinConfigBase implements Serializable {
public int getCubeStatsHLLPrecision() {
return Integer.parseInt(getOptional("kylin.job.cubing.inmem.sampling.hll.precision", "14"));
}
+
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2fc706ca/core-common/src/main/java/org/apache/kylin/common/debug/BackdoorToggles.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/debug/BackdoorToggles.java b/core-common/src/main/java/org/apache/kylin/common/debug/BackdoorToggles.java
index 0feff2d..f88781a 100644
--- a/core-common/src/main/java/org/apache/kylin/common/debug/BackdoorToggles.java
+++ b/core-common/src/main/java/org/apache/kylin/common/debug/BackdoorToggles.java
@@ -20,6 +20,9 @@ package org.apache.kylin.common.debug;
import java.util.Map;
+import org.apache.commons.lang.StringUtils;
+import org.apache.kylin.common.util.Pair;
+
/**
*/
public class BackdoorToggles {
@@ -62,6 +65,16 @@ public class BackdoorToggles {
return Integer.valueOf(v);
}
+ public static Pair<Short, Short> getShardAssignment() {
+ String v = getString(DEBUG_TOGGLE_SHARD_ASSIGNMENT);
+ if (v == null) {
+ return null;
+ } else {
+ String[] parts = StringUtils.split(v, "#");
+ return Pair.newPair(Short.valueOf(parts[0]), Short.valueOf(parts[1]));
+ }
+ }
+
private static String getString(String key) {
Map<String, String> toggles = _backdoorToggles.get();
if (toggles == null) {
@@ -153,4 +166,15 @@ public class BackdoorToggles {
}
*/
public final static String DEBUG_TOGGLE_QUERY_TIMEOUT = "DEBUG_TOGGLE_QUERY_TIMEOUT";
+
+ /**
+ * set DEBUG_TOGGLE_SHARD_ASSIGNMENT="totalAssignedWorkers#assignedWorkerID" to specify subset of shards to deal with
+ *
+ example:(put it into request body)
+ "backdoorToggles": {
+ "DEBUG_TOGGLE_SHARD_ASSIGNMENT": "4#0"
+ }
+ */
+ public final static String DEBUG_TOGGLE_SHARD_ASSIGNMENT = "DEBUG_TOGGLE_SHARD_ASSIGNMENT";
+
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2fc706ca/core-common/src/main/java/org/apache/kylin/common/restclient/RestClient.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/restclient/RestClient.java b/core-common/src/main/java/org/apache/kylin/common/restclient/RestClient.java
index a4d0761..069a142 100644
--- a/core-common/src/main/java/org/apache/kylin/common/restclient/RestClient.java
+++ b/core-common/src/main/java/org/apache/kylin/common/restclient/RestClient.java
@@ -39,14 +39,14 @@ import org.apache.kylin.common.util.JsonUtil;
*/
public class RestClient {
- String host;
- int port;
- String baseUrl;
- String userName;
- String password;
- HttpClient client;
-
- private static Pattern fullRestPattern = Pattern.compile("(?:([^:]+)[:]([^@]+)[@])?([^:]+)(?:[:](\\d+))?");
+ protected String host;
+ protected int port;
+ protected String baseUrl;
+ protected String userName;
+ protected String password;
+ protected HttpClient client;
+
+ protected static Pattern fullRestPattern = Pattern.compile("(?:([^:]+)[:]([^@]+)[@])?([^:]+)(?:[:](\\d+))?");
public static boolean matchFullRestPattern(String uri) {
Matcher m = fullRestPattern.matcher(uri);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2fc706ca/core-common/src/main/java/org/apache/kylin/common/util/LoggableCachedThreadPool.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/LoggableCachedThreadPool.java b/core-common/src/main/java/org/apache/kylin/common/util/LoggableCachedThreadPool.java
new file mode 100644
index 0000000..455951b
--- /dev/null
+++ b/core-common/src/main/java/org/apache/kylin/common/util/LoggableCachedThreadPool.java
@@ -0,0 +1,57 @@
+/*
+ * 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.common.util;
+
+import org.slf4j.*;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class LoggableCachedThreadPool extends ThreadPoolExecutor {
+
+ public static final org.slf4j.Logger logger = LoggerFactory.getLogger(LoggableCachedThreadPool.class);
+
+ public LoggableCachedThreadPool() {
+ super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
+ }
+
+ @Override
+ protected void afterExecute(Runnable r, Throwable t) {
+ super.afterExecute(r, t);
+ if (t == null && r instanceof Future<?>) {
+ try {
+ ((Future<?>) r).get();
+ } catch (ExecutionException ee) {
+ logger.error("Execution exception when running task in " + Thread.currentThread().getName());
+ t = ee.getCause();
+ } catch (InterruptedException ie) {
+ logger.error("Thread interrupted: ");
+ Thread.currentThread().interrupt(); // ignore/reset
+ } catch (Throwable throwable) {
+ t = throwable;
+ }
+ }
+ if (t != null) {
+ logger.error("Caught exception in thread " + Thread.currentThread().getName() + ": ", t);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2fc706ca/core-common/src/main/java/org/apache/kylin/common/util/NumberIterators.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/NumberIterators.java b/core-common/src/main/java/org/apache/kylin/common/util/NumberIterators.java
new file mode 100644
index 0000000..5bfbac7
--- /dev/null
+++ b/core-common/src/main/java/org/apache/kylin/common/util/NumberIterators.java
@@ -0,0 +1,34 @@
+/*
+ * 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.common.util;
+
+import java.util.Iterator;
+
+public class NumberIterators {
+
+ public static <T extends Number> long sum(Iterator<T> iterator) {
+ long ret = 0;
+ while (iterator.hasNext()) {
+ Number element = iterator.next();
+ ret += element.longValue();
+ }
+ return ret;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2fc706ca/core-cube/src/main/java/org/apache/kylin/gridtable/ScannerWorker.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/ScannerWorker.java b/core-cube/src/main/java/org/apache/kylin/gridtable/ScannerWorker.java
index 75fa94b..1ac3b02 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/ScannerWorker.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/ScannerWorker.java
@@ -24,6 +24,7 @@ import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.Iterator;
+import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.debug.BackdoorToggles;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
@@ -49,7 +50,7 @@ public class ScannerWorker {
if ("scan".equalsIgnoreCase(BackdoorToggles.getHbaseCubeQueryProtocol())) {
rpc = (IGTStorage) Class.forName("org.apache.kylin.storage.hbase.cube.v2.CubeHBaseScanRPC").getConstructor(CubeSegment.class, Cuboid.class, GTInfo.class).newInstance(cubeSeg, cuboid, info); // for local debug
} else {
- rpc = (IGTStorage) Class.forName("org.apache.kylin.storage.hbase.cube.v2.CubeHBaseEndpointRPC").getConstructor(CubeSegment.class, Cuboid.class, GTInfo.class).newInstance(cubeSeg, cuboid, info); // default behavior
+ rpc = (IGTStorage) Class.forName(KylinConfig.getInstanceFromEnv().getDefaultIGTStorage()).getConstructor(CubeSegment.class, Cuboid.class, GTInfo.class).newInstance(cubeSeg, cuboid, info); // default behavior
}
internal = rpc.getGTScanner(scanRequest);
} catch (IOException | InstantiationException | InvocationTargetException | IllegalAccessException | ClassNotFoundException | NoSuchMethodException e) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2fc706ca/core-storage/src/test/java/org/apache/kylin/storage/cache/EhcacheTest.java
----------------------------------------------------------------------
diff --git a/core-storage/src/test/java/org/apache/kylin/storage/cache/EhcacheTest.java b/core-storage/src/test/java/org/apache/kylin/storage/cache/EhcacheTest.java
index 5a88f2f..b069979 100644
--- a/core-storage/src/test/java/org/apache/kylin/storage/cache/EhcacheTest.java
+++ b/core-storage/src/test/java/org/apache/kylin/storage/cache/EhcacheTest.java
@@ -18,7 +18,7 @@
package org.apache.kylin.storage.cache;
-import java.util.List;
+import java.util.Random;
import net.sf.ehcache.Cache;
import net.sf.ehcache.CacheManager;
@@ -31,11 +31,9 @@ import net.sf.ehcache.store.MemoryStoreEvictionPolicy;
import org.junit.Ignore;
import org.junit.Test;
-import com.google.common.collect.Lists;
-
@Ignore("trial for dev")
public class EhcacheTest {
-
+
@Test
public void basicTest() throws InterruptedException {
System.out.println("runtime used memory: " + (Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()) / 1024 / 1024 + "M");
@@ -46,35 +44,52 @@ public class EhcacheTest {
//Create a Cache specifying its configuration.
Cache testCache = //Create a Cache specifying its configuration.
- new Cache(new CacheConfiguration("test", 0).//
- memoryStoreEvictionPolicy(MemoryStoreEvictionPolicy.LRU).//
- eternal(false).//
- timeToIdleSeconds(86400).//
- diskExpiryThreadIntervalSeconds(0).//
- //maxBytesLocalHeap(1000, MemoryUnit.MEGABYTES).//
- persistence(new PersistenceConfiguration().strategy(PersistenceConfiguration.Strategy.NONE)));
+ new Cache(new CacheConfiguration("test", 0).//
+ memoryStoreEvictionPolicy(MemoryStoreEvictionPolicy.LRU).//
+ eternal(false).//
+ timeToIdleSeconds(86400).//
+ diskExpiryThreadIntervalSeconds(0).//
+ //maxBytesLocalHeap(1000, MemoryUnit.MEGABYTES).//
+ persistence(new PersistenceConfiguration().strategy(PersistenceConfiguration.Strategy.NONE)));
cacheManager.addCache(testCache);
System.out.println("runtime used memory: " + (Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()) / 1024 / 1024 + "M");
- byte[] blob = new byte[(1024 * 40 * 1024)];//400M
-
- List<String> manyObjects = Lists.newArrayList();
- for (int i = 0; i < 10000; i++) {
- manyObjects.add(new String("" + i));
+ byte[] blob = new byte[(1024 * 80 * 1024)];//400M
+ Random random = new Random();
+ for (int i = 0; i < blob.length; i++) {
+ blob[i] = (byte) random.nextInt();
}
- testCache.put(new Element("0", manyObjects));
+
+
+// List<String> manyObjects = Lists.newArrayList();
+// for (int i = 0; i < 10000; i++) {
+// manyObjects.add(new String("" + i));
+// }
+// testCache.put(new Element("0", manyObjects));
testCache.put(new Element("1", blob));
System.out.println(testCache.get("1") == null);
System.out.println(testCache.getSize());
System.out.println(testCache.getStatistics().getLocalHeapSizeInBytes());
System.out.println("runtime used memory: " + (Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()) / 1024 / 1024 + "M");
+
+ blob = new byte[(1024 * 80 * 1024)];//400M
+ for (int i = 0; i < blob.length; i++) {
+ blob[i] = (byte) random.nextInt();
+ }
+
testCache.put(new Element("2", blob));
System.out.println(testCache.get("1") == null);
+ System.out.println(testCache.get("2") == null);
System.out.println(testCache.getSize());
System.out.println(testCache.getStatistics().getLocalHeapSizeInBytes());
System.out.println("runtime used memory: " + (Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()) / 1024 / 1024 + "M");
+
+ blob = new byte[(1024 * 80 * 1024)];//400M
+ for (int i = 0; i < blob.length; i++) {
+ blob[i] = (byte) random.nextInt();
+ }
testCache.put(new Element("3", blob));
System.out.println(testCache.get("1") == null);
System.out.println(testCache.get("2") == null);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2fc706ca/server/src/main/java/org/apache/kylin/rest/response/SQLResponse.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/response/SQLResponse.java b/server/src/main/java/org/apache/kylin/rest/response/SQLResponse.java
index 8fb8788..75b6ce8 100644
--- a/server/src/main/java/org/apache/kylin/rest/response/SQLResponse.java
+++ b/server/src/main/java/org/apache/kylin/rest/response/SQLResponse.java
@@ -24,40 +24,40 @@ import java.util.List;
import org.apache.kylin.rest.model.SelectedColumnMeta;
public class SQLResponse implements Serializable {
- private static final long serialVersionUID = 1L;
+ protected static final long serialVersionUID = 1L;
// private static final Logger logger =
// LoggerFactory.getLogger(SQLResponse.class);
// the data type for each column
- private List<SelectedColumnMeta> columnMetas;
+ protected List<SelectedColumnMeta> columnMetas;
// the results rows, each row contains several columns
- private List<List<String>> results;
+ protected List<List<String>> results;
/**
* for historical reasons it is named "cube", however it might also refer to any realizations like hybrid, II or etc.
*/
- private String cube;
+ protected String cube;
// if not select query, only return affected row count
- private int affectedRowCount;
+ protected int affectedRowCount;
// flag indicating whether an exception occurred
- private boolean isException;
+ protected boolean isException;
// if isException, the detailed exception message
- private String exceptionMessage;
+ protected String exceptionMessage;
- private long duration;
+ protected long duration;
- private boolean isPartial = false;
+ protected boolean isPartial = false;
- private long totalScanCount;
+ protected long totalScanCount;
- private boolean hitExceptionCache = false;
-
- private boolean storageCacheUsed = false;
+ protected boolean hitExceptionCache = false;
+
+ protected boolean storageCacheUsed = false;
public SQLResponse() {
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2fc706ca/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
index cdacec7..12b1fa8 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
@@ -26,11 +26,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.DataFormatException;
@@ -51,6 +47,7 @@ import org.apache.kylin.common.util.BytesSerializer;
import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.CompressionUtils;
import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.common.util.LoggableCachedThreadPool;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
@@ -78,39 +75,7 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
public static final Logger logger = LoggerFactory.getLogger(CubeHBaseEndpointRPC.class);
- private static ExecutorService executorService = newLoggableCachedThreadPool();
-
- public static ExecutorService newLoggableCachedThreadPool() {
- return new LoggableCachedThreadPool();
- }
-
- public static class LoggableCachedThreadPool extends ThreadPoolExecutor {
-
- public LoggableCachedThreadPool() {
- super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
- }
-
- @Override
- protected void afterExecute(Runnable r, Throwable t) {
- super.afterExecute(r, t);
- if (t == null && r instanceof Future<?>) {
- try {
- ((Future<?>) r).get();
- } catch (ExecutionException ee) {
- logger.error("Execution exception when running task in " + Thread.currentThread().getName());
- t = ee.getCause();
- } catch (InterruptedException ie) {
- logger.error("Thread interrupted: ");
- Thread.currentThread().interrupt(); // ignore/reset
- } catch (Throwable throwable) {
- t = throwable;
- }
- }
- if (t != null) {
- logger.error("Caught exception in thread " + Thread.currentThread().getName() + ": ", t);
- }
- }
- }
+ private static ExecutorService executorService = new LoggableCachedThreadPool();
static class ExpectedSizeIterator implements Iterator<byte[]> {
@@ -131,7 +96,7 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
if (BackdoorToggles.getQueryTimeout() != -1) {
this.timeout = BackdoorToggles.getQueryTimeout();
}
-
+
this.timeout *= 1.1;//allow for some delay
logger.info("Timeout for ExpectedSizeIterator is: " + this.timeout);
@@ -262,7 +227,14 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
@SuppressWarnings("unchecked")
private List<Pair<byte[], byte[]>> getEPKeyRanges(short baseShard, short shardNum, int totalShards) {
- if (baseShard + shardNum <= totalShards) {
+ if (shardNum == 0) {
+ return Lists.newArrayList();
+ }
+
+ if (shardNum == totalShards) {
+ //all shards
+ return Lists.newArrayList(Pair.newPair(getByteArrayForShort((short) 0), getByteArrayForShort((short) (shardNum - 1))));
+ } else if (baseShard + shardNum <= totalShards) {
//endpoint end key is inclusive, so no need to append 0 or anything
return Lists.newArrayList(Pair.newPair(getByteArrayForShort(baseShard), getByteArrayForShort((short) (baseShard + shardNum - 1))));
} else {
@@ -272,6 +244,10 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
}
}
+ protected Pair<Short, Short> getShardNumAndBaseShard() {
+ return Pair.newPair(cubeSeg.getCuboidShardNum(cuboid.getId()), cubeSeg.getCuboidBaseShard(cuboid.getId()));
+ }
+
@Override
public IGTScanner getGTScanner(final GTScanRequest scanRequest) throws IOException {
@@ -279,8 +255,9 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
logger.debug("New scanner for current segment {} will use {} as endpoint's behavior", cubeSeg, toggle);
- short cuboidBaseShard = cubeSeg.getCuboidBaseShard(this.cuboid.getId());
- short shardNum = cubeSeg.getCuboidShardNum(this.cuboid.getId());
+ Pair<Short, Short> shardNumAndBaseShard = getShardNumAndBaseShard();
+ short shardNum = shardNumAndBaseShard.getFirst();
+ short cuboidBaseShard = shardNumAndBaseShard.getSecond();
int totalShards = cubeSeg.getTotalShards();
ByteString scanRequestByteString = null;
@@ -339,26 +316,25 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
logScan(rs, cubeSeg.getStorageLocationIdentifier());
}
- logger.debug("Submitting rpc to {} shards starting from shard {}, scan range count {}", new Object[] { shardNum, cuboidBaseShard, rawScans.size() });
+ logger.debug("Submitting rpc to {} shards starting from shard {}, scan range count {}", shardNum, cuboidBaseShard, rawScans.size());
final AtomicInteger totalScannedCount = new AtomicInteger(0);
final ExpectedSizeIterator epResultItr = new ExpectedSizeIterator(shardNum);
+ final CubeVisitProtos.CubeVisitRequest.Builder builder = CubeVisitProtos.CubeVisitRequest.newBuilder();
+ builder.setGtScanRequest(scanRequestByteString).setHbaseRawScan(rawScanByteString);
+ for (IntList intList : hbaseColumnsToGTIntList) {
+ builder.addHbaseColumnsToGT(intList);
+ }
+ builder.setRowkeyPreambleSize(cubeSeg.getRowKeyPreambleSize());
+ builder.setBehavior(toggle);
+ builder.setStartTime(System.currentTimeMillis());
+ builder.setTimeout(epResultItr.getTimeout());
for (final Pair<byte[], byte[]> epRange : getEPKeyRanges(cuboidBaseShard, shardNum, totalShards)) {
- final ByteString finalScanRequestByteString = scanRequestByteString;
- final ByteString finalRawScanByteString = rawScanByteString;
executorService.submit(new Runnable() {
@Override
public void run() {
- CubeVisitProtos.CubeVisitRequest.Builder builder = CubeVisitProtos.CubeVisitRequest.newBuilder();
- builder.setGtScanRequest(finalScanRequestByteString).setHbaseRawScan(finalRawScanByteString);
- for (IntList intList : hbaseColumnsToGTIntList) {
- builder.addHbaseColumnsToGT(intList);
- }
- builder.setRowkeyPreambleSize(cubeSeg.getRowKeyPreambleSize());
- builder.setBehavior(toggle);
- builder.setStartTime(System.currentTimeMillis());
- builder.setTimeout(epResultItr.getTimeout());
+
String logHeader = "<sub-thread for GTScanRequest " + Integer.toHexString(System.identityHashCode(scanRequest)) + "> ";
Map<byte[], CubeVisitProtos.CubeVisitResponse> results;