DRILL-1426: Add IO wait stats to HBaseRecordReader
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/2a45d959 Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/2a45d959 Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/2a45d959 Branch: refs/heads/master Commit: 2a45d9597671378d50108ccdcb5abe9cf8c47c08 Parents: 2eb04e7 Author: Steven Phillips <sphill...@maprtech.com> Authored: Tue Sep 16 17:24:47 2014 -0700 Committer: Steven Phillips <sphill...@maprtech.com> Committed: Sun Sep 28 00:10:45 2014 -0700 ---------------------------------------------------------------------- .../apache/drill/exec/store/hbase/HBaseRecordReader.java | 11 ++++++++++- .../org/apache/drill/exec/store/hbase/HBaseSubScan.java | 3 ++- .../org/apache/drill/exec/physical/impl/ScanBatch.java | 1 + 3 files changed, 13 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/2a45d959/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseRecordReader.java ---------------------------------------------------------------------- diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseRecordReader.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseRecordReader.java index a5b6c94..33bf376 100644 --- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseRecordReader.java +++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseRecordReader.java @@ -182,7 +182,16 @@ public class HBaseRecordReader extends AbstractRecordReader implements DrillHBas result = leftOver; leftOver = null; } else { - result = resultScanner.next(); + if (operatorContext != null) { + operatorContext.getStats().startWait(); + } + try { + result = resultScanner.next(); + } finally { + if (operatorContext != null) { + operatorContext.getStats().stopWait(); + } + } } } catch (IOException e) { throw new DrillRuntimeException(e); http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/2a45d959/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSubScan.java ---------------------------------------------------------------------- diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSubScan.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSubScan.java index fbac159..23d8c5a 100644 --- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSubScan.java +++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseSubScan.java @@ -28,6 +28,7 @@ import org.apache.drill.exec.physical.base.AbstractBase; import org.apache.drill.exec.physical.base.PhysicalOperator; import org.apache.drill.exec.physical.base.PhysicalVisitor; import org.apache.drill.exec.physical.base.SubScan; +import org.apache.drill.exec.proto.UserBitShared.CoreOperatorType; import org.apache.drill.exec.store.StoragePluginRegistry; import org.apache.hadoop.hbase.filter.Filter; import org.apache.hadoop.hbase.util.Bytes; @@ -211,7 +212,7 @@ public class HBaseSubScan extends AbstractBase implements SubScan { @Override public int getOperatorType() { - return 1001; + return CoreOperatorType.HBASE_SUB_SCAN_VALUE; } } http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/2a45d959/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java ---------------------------------------------------------------------- diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java index ad8bf96..fc23441 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java @@ -176,6 +176,7 @@ public class ScanBatch implements RecordBatch { currentReader = readers.next(); partitionValues = partitionColumns.hasNext() ? partitionColumns.next() : null; currentReader.setup(mutator); + currentReader.setOperatorContext(oContext); try { currentReader.allocate(fieldVectorMap); } catch (OutOfMemoryException e) {
