kylin-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mahong...@apache.org
Subject kylin git commit: KYLIN-1350 - hbase Result.binarySearch is found to be problematic in concurrent environments
Date Wed, 20 Jan 2016 09:54:49 GMT
Repository: kylin
Updated Branches:
  refs/heads/2.x-staging f1cd86e5a -> e5653278c


KYLIN-1350 - hbase Result.binarySearch is found to be problematic in concurrent environments


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

Branch: refs/heads/2.x-staging
Commit: e5653278cb1236aa93e11dd8b68356d04e2e277b
Parents: f1cd86e
Author: honma <honma@ebay.com>
Authored: Wed Jan 20 15:15:54 2016 +0800
Committer: honma <honma@ebay.com>
Committed: Wed Jan 20 17:53:36 2016 +0800

----------------------------------------------------------------------
 .../hbase/cube/v1/CubeSegmentTupleIterator.java |  1 +
 .../cube/v2/SequentialCubeTupleIterator.java    |  2 +
 .../storage/hbase/steps/RowValueDecoder.java    | 14 ++++--
 .../kylin/storage/hbase/util/Results.java       | 49 ++++++++++++++++++++
 4 files changed, 62 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kylin/blob/e5653278/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 f8edee2..1232cb2 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
@@ -307,6 +307,7 @@ public class CubeSegmentTupleIterator implements ITupleIterator {
                 logger.debug("HBase Metrics when scanning " + this.tableName + " count={},
ms={}, bytes={}, remote_bytes={}, regions={}, not_serving_region={}, rpc={}, rpc_retries={},
remote_rpc={}, remote_rpc_retries={}", //
                         new Object[] { scanCount, scanMetrics.sumOfMillisSecBetweenNexts,
scanMetrics.countOfBytesInResults, scanMetrics.countOfBytesInRemoteResults, scanMetrics.countOfRegions,
scanMetrics.countOfNSRE, scanMetrics.countOfRPCcalls, scanMetrics.countOfRPCRetries, scanMetrics.countOfRemoteRPCcalls,
scanMetrics.countOfRemoteRPCRetries });
             }
+            scan = null;
         }
         try {
             if (scanner != null) {

http://git-wip-us.apache.org/repos/asf/kylin/blob/e5653278/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/SequentialCubeTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/SequentialCubeTupleIterator.java
b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/SequentialCubeTupleIterator.java
index 77509e9..f0cebfe 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/SequentialCubeTupleIterator.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/SequentialCubeTupleIterator.java
@@ -144,6 +144,8 @@ public class SequentialCubeTupleIterator implements ITupleIterator {
 
     @Override
     public void close() {
+        // hasNext() loop may exit because of limit, threshold, etc.
+        // close all the remaining segmentIterator
         flushScanCountDelta();
 
         if (curScanner != null)

http://git-wip-us.apache.org/repos/asf/kylin/blob/e5653278/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RowValueDecoder.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RowValueDecoder.java
b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RowValueDecoder.java
index 59a5fed..7d05dc9 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RowValueDecoder.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RowValueDecoder.java
@@ -30,11 +30,16 @@ import org.apache.kylin.metadata.datatype.DoubleMutable;
 import org.apache.kylin.metadata.datatype.LongMutable;
 import org.apache.kylin.metadata.model.FunctionDesc;
 import org.apache.kylin.metadata.model.MeasureDesc;
+import org.apache.kylin.storage.hbase.util.Results;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  */
 public class RowValueDecoder implements Cloneable {
 
+    private static final Logger logger = LoggerFactory.getLogger(RowValueDecoder.class);
+
     private final HBaseColumnDesc hbaseColumn;
     private final byte[] hbaseColumnFamily;
     private final byte[] hbaseColumnQualifier;
@@ -57,13 +62,14 @@ public class RowValueDecoder implements Cloneable {
     public void decodeAndConvertJavaObj(Result hbaseRow) {
         decode(hbaseRow, true);
     }
-    
+
     public void decode(Result hbaseRow) {
         decode(hbaseRow, false);
     }
 
     private void decode(Result hbaseRow, boolean convertToJavaObject) {
-        decode(hbaseRow.getValueAsByteBuffer(hbaseColumnFamily, hbaseColumnQualifier), convertToJavaObject);
+        ByteBuffer buffer = Results.getValueAsByteBuffer(hbaseRow, hbaseColumnFamily, hbaseColumnQualifier);
+        decode(buffer, convertToJavaObject);
     }
 
     public void decodeAndConvertJavaObj(byte[] bytes) {
@@ -73,7 +79,7 @@ public class RowValueDecoder implements Cloneable {
     public void decode(byte[] bytes) {
         decode(ByteBuffer.wrap(bytes), false);
     }
-    
+
     private void decode(ByteBuffer buffer, boolean convertToJavaObject) {
         codec.decode(buffer, values);
         if (convertToJavaObject) {
@@ -113,7 +119,7 @@ public class RowValueDecoder implements Cloneable {
     public MeasureDesc[] getMeasures() {
         return measures;
     }
-    
+
     // result is in order of <code>CubeDesc.getMeasures()</code>
     public void loadCubeMeasureArray(Object result[]) {
         int[] measureIndex = hbaseColumn.getMeasureIndex();

http://git-wip-us.apache.org/repos/asf/kylin/blob/e5653278/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/Results.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/Results.java
b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/Results.java
new file mode 100644
index 0000000..f619007
--- /dev/null
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/Results.java
@@ -0,0 +1,49 @@
+/*
+ * 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.util;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.kylin.common.util.Bytes;
+
+/**
+ * The helper class is introduced because {@link Result#binarySearch(Cell[], byte[], int,
int, byte[], int, int)} 
+ * is found to be problematic in concurrent environments, and unfortunately  {@link Result#getValueAsByteBuffer(byte[],
byte[])} 
+ * calls it.
+ */
+public class Results {
+    public static ByteBuffer getValueAsByteBuffer(Result hbaseRow, byte[] cf, byte[] cq)
{
+        List<Cell> cells = hbaseRow.listCells();
+        if (cells == null || cells.size() == 0) {
+            return null;
+        } else {
+            for (Cell c : cells) {
+                if (Bytes.compareTo(cf, 0, cf.length, c.getFamilyArray(), c.getFamilyOffset(),
c.getFamilyLength()) == 0 && //
+                        Bytes.compareTo(cq, 0, cq.length, c.getQualifierArray(), c.getQualifierOffset(),
c.getQualifierLength()) == 0) {
+                    return ByteBuffer.wrap(c.getValueArray(), c.getValueOffset(), c.getValueLength());
+                }
+            }
+        }
+        return null;
+    }
+
+}


Mime
View raw message