kylin-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From billy...@apache.org
Subject [1/4] kylin git commit: minor code refactors [Forced Update!]
Date Fri, 23 Dec 2016 07:57:54 GMT
Repository: kylin
Updated Branches:
  refs/heads/yang22-cdh5.7 74658e925 -> 42002f605 (forced update)


minor code refactors


Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/7c77a17f
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/7c77a17f
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/7c77a17f

Branch: refs/heads/yang22-cdh5.7
Commit: 7c77a17f56404d8c263254b8567c1ec105a85ff6
Parents: 85313ee
Author: Hongbin Ma <mahongbin@apache.org>
Authored: Thu Dec 22 09:59:43 2016 +0800
Committer: Hongbin Ma <mahongbin@apache.org>
Committed: Fri Dec 23 11:32:13 2016 +0800

----------------------------------------------------------------------
 .../apache/kylin/common/KylinConfigBase.java    |  6 ++-
 .../org/apache/kylin/common/util/BasicTest.java |  1 +
 .../gtrecord/DummyPartitionStreamer.java        | 40 ++++++++++++++++++++
 .../storage/gtrecord/IPartitionStreamer.java    | 26 +++++++++++++
 .../gtrecord/StorageResponseGTScatter.java      |  9 +++--
 .../apache/kylin/rest/service/QueryService.java |  6 ++-
 .../hbase/cube/v2/CubeHBaseEndpointRPC.java     |  9 +++--
 7 files changed, 87 insertions(+), 10 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kylin/blob/7c77a17f/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 a1c9050..ceda95c 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
@@ -782,7 +782,7 @@ abstract public class KylinConfigBase implements Serializable {
         return Boolean.valueOf(getOptional("kylin.query.skip-empty-segments", "true"));
     }
 
-    @Deprecated//Limit is good even it's large. This config is meaning less since we already
have scan threshold 
+    @Deprecated //Limit is good even it's large. This config is meaning less since we already
have scan threshold 
     public int getStoragePushDownLimitMax() {
         return Integer.parseInt(getOptional("kylin.query.max-limit-pushdown", "10000"));
     }
@@ -791,6 +791,10 @@ abstract public class KylinConfigBase implements Serializable {
         return Integer.parseInt(getOptional("kylin.query.scan-threshold", "10000000"));
     }
 
+    public int getLargeQueryThreshold() {
+        return Integer.parseInt(getOptional("kylin.query.large-query-threshold", String.valueOf((int)
(getScanThreshold() * 0.1))));
+    }
+
     public int getDerivedInThreshold() {
         return Integer.parseInt(getOptional("kylin.query.derived-filter-translation-threshold",
"20"));
     }

http://git-wip-us.apache.org/repos/asf/kylin/blob/7c77a17f/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java b/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
index 5eaa011..9105245 100644
--- a/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
+++ b/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
@@ -34,6 +34,7 @@ import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import com.google.common.collect.Iterators;
 import org.apache.commons.lang3.time.FastDateFormat;
 import org.junit.Ignore;
 import org.junit.Test;

http://git-wip-us.apache.org/repos/asf/kylin/blob/7c77a17f/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/DummyPartitionStreamer.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/DummyPartitionStreamer.java
b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/DummyPartitionStreamer.java
new file mode 100644
index 0000000..4caaed0
--- /dev/null
+++ b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/DummyPartitionStreamer.java
@@ -0,0 +1,40 @@
+/*
+ * 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.gtrecord;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+public class DummyPartitionStreamer implements IPartitionStreamer {
+    private Iterator<byte[]> iterator;
+
+    public DummyPartitionStreamer(Iterator<byte[]> iterator) {
+        this.iterator = iterator;
+    }
+
+    @Override
+    public void close() throws IOException {
+        //do nothing
+    }
+
+    @Override
+    public Iterator<byte[]> asByteArrayIterator() {
+        return this.iterator;
+    }
+}

http://git-wip-us.apache.org/repos/asf/kylin/blob/7c77a17f/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/IPartitionStreamer.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/IPartitionStreamer.java
b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/IPartitionStreamer.java
new file mode 100644
index 0000000..42f1151
--- /dev/null
+++ b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/IPartitionStreamer.java
@@ -0,0 +1,26 @@
+/*
+ * 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.gtrecord;
+
+import java.io.Closeable;
+import java.util.Iterator;
+
+public interface IPartitionStreamer extends Closeable {
+    public Iterator<byte[]> asByteArrayIterator();
+}

http://git-wip-us.apache.org/repos/asf/kylin/blob/7c77a17f/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/StorageResponseGTScatter.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/StorageResponseGTScatter.java
b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/StorageResponseGTScatter.java
index fe1afd3..6283340 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/StorageResponseGTScatter.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/gtrecord/StorageResponseGTScatter.java
@@ -42,14 +42,16 @@ public class StorageResponseGTScatter implements IGTScanner {
     private static final Logger logger = LoggerFactory.getLogger(StorageResponseGTScatter.class);
 
     private GTInfo info;
+    private IPartitionStreamer partitionStreamer;
     private Iterator<byte[]> blocks;
     private ImmutableBitSet columns;
     private long totalScannedCount;
     private int storagePushDownLimit = -1;
 
-    public StorageResponseGTScatter(GTInfo info, Iterator<byte[]> blocks, ImmutableBitSet
columns, long totalScannedCount, int storagePushDownLimit) {
+    public StorageResponseGTScatter(GTInfo info, IPartitionStreamer partitionStreamer, ImmutableBitSet
columns, long totalScannedCount, int storagePushDownLimit) {
         this.info = info;
-        this.blocks = blocks;
+        this.partitionStreamer = partitionStreamer;
+        this.blocks = partitionStreamer.asByteArrayIterator();
         this.columns = columns;
         this.totalScannedCount = totalScannedCount;
         this.storagePushDownLimit = storagePushDownLimit;
@@ -67,7 +69,8 @@ public class StorageResponseGTScatter implements IGTScanner {
 
     @Override
     public void close() throws IOException {
-        //do nothing
+        //If upper consumer failed while consuming the GTRecords, the consumer should call
IGTScanner's close method to ensure releasing resource
+        partitionStreamer.close();
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/kylin/blob/7c77a17f/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java b/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java
index 8810c85..82e4a87 100644
--- a/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java
+++ b/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java
@@ -360,8 +360,10 @@ public class QueryService extends BasicService {
                     sqlResponse.setDuration(System.currentTimeMillis() - startTime);
                     logger.info("Stats of SQL response: isException: {}, duration: {}, total
scan count {}", //
                             String.valueOf(sqlResponse.getIsException()), String.valueOf(sqlResponse.getDuration()),
String.valueOf(sqlResponse.getTotalScanCount()));
-                    if (queryCacheEnabled && !sqlResponse.getIsException() //
-                            && (sqlResponse.getDuration() > durationThreshold
|| sqlResponse.getTotalScanCount() > scancountThreshold)) {
+                    if (queryCacheEnabled && //
+                            !sqlResponse.getIsException() && //
+                            (sqlResponse.getDuration() > durationThreshold || sqlResponse.getTotalScanCount()
> scancountThreshold) && //
+                            (sqlResponse.getResults().size() < kylinConfig.getLargeQueryThreshold()))
{ //don't cache too large response
                         cacheManager.getCache(SUCCESS_QUERY_CACHE).put(new Element(sqlRequest,
sqlResponse));
                     }
                 } else {

http://git-wip-us.apache.org/repos/asf/kylin/blob/7c77a17f/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 d99f80e..ebacb26 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
@@ -40,12 +40,13 @@ 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.metadata.model.ISegment;
 import org.apache.kylin.cube.cuboid.Cuboid;
 import org.apache.kylin.gridtable.GTInfo;
 import org.apache.kylin.gridtable.GTScanRequest;
 import org.apache.kylin.gridtable.GTScanSelfTerminatedException;
 import org.apache.kylin.gridtable.IGTScanner;
+import org.apache.kylin.metadata.model.ISegment;
+import org.apache.kylin.storage.gtrecord.DummyPartitionStreamer;
 import org.apache.kylin.storage.gtrecord.StorageResponseGTScatter;
 import org.apache.kylin.storage.hbase.HBaseConnection;
 import org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos;
@@ -127,14 +128,14 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
         //TODO: raw scan can be constructed at region side to reduce traffic
         List<RawScan> rawScans = preparedHBaseScans(scanRequest.getGTScanRanges(),
selectedColBlocks);
         rawScanByteString = serializeRawScans(rawScans);
-        
+
         int coprocessorTimeout = getCoprocessorTimeoutMillis();
         scanRequest.setTimeout(coprocessorTimeout);
         scanRequest.clearScanRanges();//since raw scans are sent to coprocessor, we don't
need to duplicate sending it
         scanRequestByteString = serializeGTScanReq(scanRequest);
 
         final ExpectedSizeIterator epResultItr = new ExpectedSizeIterator(shardNum, coprocessorTimeout);
-        
+
         logger.info("Serialized scanRequestBytes {} bytes, rawScanBytesString {} bytes",
scanRequestByteString.size(), rawScanByteString.size());
 
         logger.info("The scan {} for segment {} is as below with {} separate raw scans, shard
part of start/end key is set to 0", Integer.toHexString(System.identityHashCode(scanRequest)),
cubeSeg, rawScans.size());
@@ -230,7 +231,7 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
             });
         }
 
-        return new StorageResponseGTScatter(fullGTInfo, epResultItr, scanRequest.getColumns(),
totalScannedCount.get(), scanRequest.getStoragePushDownLimit());
+        return new StorageResponseGTScatter(fullGTInfo, new DummyPartitionStreamer(epResultItr),
scanRequest.getColumns(), totalScannedCount.get(), scanRequest.getStoragePushDownLimit());
     }
 
     private ByteString serializeGTScanReq(GTScanRequest scanRequest) {


Mime
View raw message