You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ja...@apache.org on 2015/11/13 03:37:50 UTC
[20/45] drill git commit: DRILL-3987: (REFACTOR) Common and Vector
modules building.
DRILL-3987: (REFACTOR) Common and Vector modules building.
- Extract Accountor interface from Implementation
- Separate FMPP modules to separate out Vector Needs versus external needs
- Separate out Vector classes from those that are VectorAccessible.
- Cleanup Memory Exception hiearchy
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/4524fdbe
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/4524fdbe
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/4524fdbe
Branch: refs/heads/master
Commit: 4524fdbe938324af9eebeea2f8863ecb62c66a25
Parents: 9969d8b
Author: Jacques Nadeau <ja...@apache.org>
Authored: Sat Nov 7 18:16:05 2015 -0800
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Thu Nov 12 17:00:46 2015 -0800
----------------------------------------------------------------------
.../src/main/codegen/includes/vv_imports.ftl | 2 +-
.../src/main/codegen/templates/TypeHelper.java | 95 +++
.../apache/drill/exec/client/DrillClient.java | 2 +-
.../exec/expr/fn/impl/ByteFunctionHelpers.java | 236 ------
.../drill/exec/expr/fn/impl/DateUtility.java | 682 -----------------
.../org/apache/drill/exec/memory/Accountor.java | 462 ------------
.../apache/drill/exec/memory/AccountorImpl.java | 427 +++++++++++
.../drill/exec/memory/OutOfMemoryException.java | 51 --
.../drill/exec/memory/TopLevelAllocator.java | 70 +-
.../apache/drill/exec/ops/BufferManager.java | 81 --
.../drill/exec/ops/BufferManagerImpl.java | 46 ++
.../apache/drill/exec/ops/FragmentContext.java | 45 +-
.../drill/exec/ops/OperatorContextImpl.java | 52 +-
.../org/apache/drill/exec/ops/QueryContext.java | 15 +-
.../drill/exec/physical/impl/BaseRootExec.java | 2 +-
.../physical/impl/MergingReceiverCreator.java | 2 +-
.../drill/exec/physical/impl/ScanBatch.java | 7 +-
.../drill/exec/physical/impl/ScreenCreator.java | 7 +-
.../exec/physical/impl/SingleSenderCreator.java | 6 +-
.../exec/physical/impl/TopN/TopNBatch.java | 2 +-
.../exec/physical/impl/WriterRecordBatch.java | 2 +-
.../impl/aggregate/StreamingAggBatch.java | 2 +-
.../BroadcastSenderRootExec.java | 6 +-
.../physical/impl/filter/FilterRecordBatch.java | 2 +-
.../physical/impl/filter/FilterTemplate2.java | 4 +-
.../impl/flatten/FlattenRecordBatch.java | 2 +-
.../exec/physical/impl/join/HashJoinBatch.java | 2 +-
.../exec/physical/impl/join/MergeJoinBatch.java | 2 +-
.../physical/impl/join/NestedLoopJoinBatch.java | 3 +-
.../physical/impl/limit/LimitRecordBatch.java | 2 +-
.../impl/mergereceiver/MergingRecordBatch.java | 3 +-
.../OrderedPartitionRecordBatch.java | 2 +-
.../PartitionSenderRootExec.java | 6 +-
.../impl/producer/ProducerConsumerBatch.java | 8 +-
.../impl/project/ProjectRecordBatch.java | 2 +-
.../exec/physical/impl/sort/SortBatch.java | 2 +-
.../impl/svremover/RemovingRecordBatch.java | 2 +-
.../impl/union/UnionAllRecordBatch.java | 6 +-
.../UnorderedReceiverBatch.java | 2 +-
.../impl/window/WindowFrameRecordBatch.java | 3 +-
.../physical/impl/xsort/ExternalSortBatch.java | 11 +-
.../drill/exec/record/AbstractRecordBatch.java | 2 +-
.../exec/record/AbstractSingleRecordBatch.java | 2 +-
.../exec/record/selection/SelectionVector2.java | 4 +-
.../drill/exec/rpc/ProtobufLengthDecoder.java | 4 +-
.../drill/exec/store/AbstractRecordReader.java | 2 +-
.../apache/drill/exec/store/RecordReader.java | 2 +-
.../exec/store/easy/json/JSONRecordReader.java | 2 +-
.../drill/exec/store/mock/MockRecordReader.java | 2 +-
.../exec/store/parquet/ParquetFormatPlugin.java | 3 +-
.../exec/store/parquet/ParquetRecordWriter.java | 3 +-
.../columnreaders/ParquetRecordReader.java | 3 +-
.../exec/store/parquet2/DrillParquetReader.java | 3 +-
.../drill/exec/store/pojo/PojoRecordReader.java | 2 +-
.../apache/drill/exec/util/AssertionUtil.java | 39 -
.../apache/drill/exec/util/DecimalUtility.java | 730 -------------------
.../exec/work/batch/AbstractDataCollector.java | 2 +-
.../exec/work/batch/SpoolingRawBatchBuffer.java | 2 +-
.../apache/drill/exec/work/foreman/Foreman.java | 7 +-
.../exec/work/fragment/FragmentExecutor.java | 4 +-
.../parquet/hadoop/ColumnChunkIncReadStore.java | 4 +-
.../java/org/apache/drill/QueryTestUtil.java | 2 +-
.../apache/drill/TestAllocationException.java | 4 +-
.../drill/exec/memory/TestAllocators.java | 4 +-
.../partitionsender/TestPartitionSender.java | 2 +-
.../drill/exec/server/DrillClientFactory.java | 2 +-
exec/vector/pom.xml | 6 +
.../src/main/codegen/includes/vv_imports.ftl | 2 +-
.../main/codegen/templates/BasicTypeHelper.java | 544 ++++++++++++++
.../codegen/templates/FixedValueVectors.java | 6 +-
.../codegen/templates/NullableValueVectors.java | 4 +-
.../codegen/templates/RepeatedValueVectors.java | 6 +-
.../src/main/codegen/templates/TypeHelper.java | 600 ---------------
.../src/main/codegen/templates/UnionVector.java | 2 +-
.../templates/VariableLengthVectors.java | 6 +-
.../src/main/java/io/netty/buffer/DrillBuf.java | 18 +-
.../java/io/netty/buffer/FakeAllocator.java | 45 +-
.../netty/buffer/PooledByteBufAllocatorL.java | 9 +-
.../exec/exception/OutOfMemoryException.java | 51 ++
.../exception/OversizedAllocationException.java | 1 -
.../exec/expr/fn/impl/ByteFunctionHelpers.java | 236 ++++++
.../drill/exec/expr/fn/impl/DateUtility.java | 682 +++++++++++++++++
.../org/apache/drill/exec/memory/Accountor.java | 43 ++
.../drill/exec/memory/BufferAllocator.java | 30 +-
.../apache/drill/exec/memory/LimitConsumer.java | 28 +
.../memory/OutOfMemoryRuntimeException.java | 49 --
.../apache/drill/exec/ops/BufferManager.java | 66 ++
.../drill/exec/record/MaterializedField.java | 4 +-
.../apache/drill/exec/record/TypedFieldId.java | 5 +-
.../apache/drill/exec/util/AssertionUtil.java | 39 +
.../apache/drill/exec/util/DecimalUtility.java | 730 +++++++++++++++++++
.../drill/exec/vector/AllocationHelper.java | 2 +-
.../org/apache/drill/exec/vector/BitVector.java | 6 +-
.../apache/drill/exec/vector/ObjectVector.java | 6 +-
.../drill/exec/vector/ValueHolderHelper.java | 10 +-
.../apache/drill/exec/vector/ValueVector.java | 6 +-
.../apache/drill/exec/vector/ZeroVector.java | 4 +-
.../vector/complex/AbstractContainerVector.java | 6 +-
.../exec/vector/complex/AbstractMapVector.java | 18 +-
.../vector/complex/BaseRepeatedValueVector.java | 12 +-
.../drill/exec/vector/complex/ListVector.java | 4 +-
.../drill/exec/vector/complex/MapVector.java | 9 +-
.../exec/vector/complex/RepeatedListVector.java | 8 +-
.../exec/vector/complex/RepeatedMapVector.java | 8 +-
.../vector/complex/impl/ComplexWriterImpl.java | 25 -
.../vector/complex/impl/PromotableWriter.java | 12 +-
106 files changed, 3309 insertions(+), 3241 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/4524fdbe/exec/java-exec/src/main/codegen/includes/vv_imports.ftl
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/codegen/includes/vv_imports.ftl b/exec/java-exec/src/main/codegen/includes/vv_imports.ftl
index 2fedfd0..5865ece 100644
--- a/exec/java-exec/src/main/codegen/includes/vv_imports.ftl
+++ b/exec/java-exec/src/main/codegen/includes/vv_imports.ftl
@@ -46,7 +46,7 @@ import org.apache.drill.exec.vector.complex.writer.BaseWriter.MapWriter;
import org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter;
import org.apache.drill.exec.util.JsonStringArrayList;
-import org.apache.drill.exec.memory.OutOfMemoryRuntimeException;
+import org.apache.drill.exec.exception.OutOfMemoryException;
import com.sun.codemodel.JType;
import com.sun.codemodel.JCodeModel;
http://git-wip-us.apache.org/repos/asf/drill/blob/4524fdbe/exec/java-exec/src/main/codegen/templates/TypeHelper.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/codegen/templates/TypeHelper.java b/exec/java-exec/src/main/codegen/templates/TypeHelper.java
new file mode 100644
index 0000000..133ed3b
--- /dev/null
+++ b/exec/java-exec/src/main/codegen/templates/TypeHelper.java
@@ -0,0 +1,95 @@
+/**
+ * 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.
+ */
+
+import org.apache.drill.exec.vector.complex.UnionVector;
+
+<@pp.dropOutputFile />
+<@pp.changeOutputFile name="/org/apache/drill/exec/expr/TypeHelper.java" />
+
+<#include "/@includes/license.ftl" />
+
+package org.apache.drill.exec.expr;
+
+<#include "/@includes/vv_imports.ftl" />
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.types.TypeProtos.DataMode;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.common.types.TypeProtos.MajorType;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.vector.accessor.*;
+import org.apache.drill.exec.vector.complex.RepeatedMapVector;
+import org.apache.drill.exec.util.CallBack;
+
+public class TypeHelper extends BasicTypeHelper{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TypeHelper.class);
+
+ public static SqlAccessor getSqlAccessor(ValueVector vector){
+ final MajorType type = vector.getField().getType();
+ switch(type.getMinorType()){
+ case UNION:
+ return new UnionSqlAccessor((UnionVector) vector);
+ <#list vv.types as type>
+ <#list type.minor as minor>
+ case ${minor.class?upper_case}:
+ switch (type.getMode()) {
+ case REQUIRED:
+ return new ${minor.class}Accessor((${minor.class}Vector) vector);
+ case OPTIONAL:
+ return new Nullable${minor.class}Accessor((Nullable${minor.class}Vector) vector);
+ case REPEATED:
+ return new GenericAccessor(vector);
+ }
+ </#list>
+ </#list>
+ case MAP:
+ case LIST:
+ return new GenericAccessor(vector);
+ }
+ throw new UnsupportedOperationException(buildErrorMessage("find sql accessor", type));
+ }
+
+ public static JType getHolderType(JCodeModel model, MinorType type, DataMode mode){
+ switch (type) {
+ case UNION:
+ return model._ref(UnionHolder.class);
+ case MAP:
+ case LIST:
+ return model._ref(ComplexHolder.class);
+
+<#list vv.types as type>
+ <#list type.minor as minor>
+ case ${minor.class?upper_case}:
+ switch (mode) {
+ case REQUIRED:
+ return model._ref(${minor.class}Holder.class);
+ case OPTIONAL:
+ return model._ref(Nullable${minor.class}Holder.class);
+ case REPEATED:
+ return model._ref(Repeated${minor.class}Holder.class);
+ }
+ </#list>
+</#list>
+ case GENERIC_OBJECT:
+ return model._ref(ObjectHolder.class);
+ default:
+ break;
+ }
+ throw new UnsupportedOperationException(buildErrorMessage("get holder type", type, mode));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/4524fdbe/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
index 08663eb..e476fff 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
@@ -41,8 +41,8 @@ import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.coord.ClusterCoordinator;
import org.apache.drill.exec.coord.zk.ZKClusterCoordinator;
+import org.apache.drill.exec.exception.OutOfMemoryException;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
http://git-wip-us.apache.org/repos/asf/drill/blob/4524fdbe/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ByteFunctionHelpers.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ByteFunctionHelpers.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ByteFunctionHelpers.java
deleted file mode 100644
index 191b459..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ByteFunctionHelpers.java
+++ /dev/null
@@ -1,236 +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.drill.exec.expr.fn.impl;
-
-import io.netty.buffer.DrillBuf;
-import io.netty.util.internal.PlatformDependent;
-
-import org.apache.drill.exec.util.AssertionUtil;
-import org.apache.drill.exec.util.DecimalUtility;
-
-import com.google.common.primitives.UnsignedLongs;
-
-public class ByteFunctionHelpers {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ByteFunctionHelpers.class);
-
- private static final boolean BOUNDS_CHECKING_ENABLED = AssertionUtil.BOUNDS_CHECKING_ENABLED;
-
- /**
- * Helper function to check for equality of bytes in two DrillBuffers
- *
- * @param left Left DrillBuf for comparison
- * @param lStart start offset in the buffer
- * @param lEnd end offset in the buffer
- * @param right Right DrillBuf for comparison
- * @param rStart start offset in the buffer
- * @param rEnd end offset in the buffer
- * @return 1 if left input is greater, -1 if left input is smaller, 0 otherwise
- */
- public static final int equal(final DrillBuf left, int lStart, int lEnd, final DrillBuf right, int rStart, int rEnd){
- if(BOUNDS_CHECKING_ENABLED){
- left.checkBytes(lStart, lEnd);
- right.checkBytes(rStart, rEnd);
- }
- return memEqual(left.memoryAddress(), lStart, lEnd, right.memoryAddress(), rStart, rEnd);
- }
-
- private static final int memEqual(final long laddr, int lStart, int lEnd, final long raddr, int rStart,
- final int rEnd) {
-
- int n = lEnd - lStart;
- if (n == rEnd - rStart) {
- long lPos = laddr + lStart;
- long rPos = raddr + rStart;
-
- while (n > 7) {
- long leftLong = PlatformDependent.getLong(lPos);
- long rightLong = PlatformDependent.getLong(rPos);
- if (leftLong != rightLong) {
- return 0;
- }
- lPos += 8;
- rPos += 8;
- n -= 8;
- }
- while (n-- != 0) {
- byte leftByte = PlatformDependent.getByte(lPos);
- byte rightByte = PlatformDependent.getByte(rPos);
- if (leftByte != rightByte) {
- return 0;
- }
- lPos++;
- rPos++;
- }
- return 1;
- } else {
- return 0;
- }
- }
-
- /**
- * Helper function to compare a set of bytes in two DrillBuffers.
- *
- * Function will check data before completing in the case that
- *
- * @param left Left DrillBuf to compare
- * @param lStart start offset in the buffer
- * @param lEnd end offset in the buffer
- * @param right Right DrillBuf to compare
- * @param rStart start offset in the buffer
- * @param rEnd end offset in the buffer
- * @return 1 if left input is greater, -1 if left input is smaller, 0 otherwise
- */
- public static final int compare(final DrillBuf left, int lStart, int lEnd, final DrillBuf right, int rStart, int rEnd){
- if(BOUNDS_CHECKING_ENABLED){
- left.checkBytes(lStart, lEnd);
- right.checkBytes(rStart, rEnd);
- }
- return memcmp(left.memoryAddress(), lStart, lEnd, right.memoryAddress(), rStart, rEnd);
- }
-
- private static final int memcmp(final long laddr, int lStart, int lEnd, final long raddr, int rStart, final int rEnd) {
- int lLen = lEnd - lStart;
- int rLen = rEnd - rStart;
- int n = Math.min(rLen, lLen);
- long lPos = laddr + lStart;
- long rPos = raddr + rStart;
-
- while (n > 7) {
- long leftLong = PlatformDependent.getLong(lPos);
- long rightLong = PlatformDependent.getLong(rPos);
- if (leftLong != rightLong) {
- return UnsignedLongs.compare(Long.reverseBytes(leftLong), Long.reverseBytes(rightLong));
- }
- lPos += 8;
- rPos += 8;
- n -= 8;
- }
-
- while (n-- != 0) {
- byte leftByte = PlatformDependent.getByte(lPos);
- byte rightByte = PlatformDependent.getByte(rPos);
- if (leftByte != rightByte) {
- return ((leftByte & 0xFF) - (rightByte & 0xFF)) > 0 ? 1 : -1;
- }
- lPos++;
- rPos++;
- }
-
- if (lLen == rLen) {
- return 0;
- }
-
- return lLen > rLen ? 1 : -1;
-
- }
-
- /**
- * Helper function to compare a set of bytes in DrillBuf to a ByteArray.
- *
- * @param left Left DrillBuf for comparison purposes
- * @param lStart start offset in the buffer
- * @param lEnd end offset in the buffer
- * @param right second input to be compared
- * @param rStart start offset in the byte array
- * @param rEnd end offset in the byte array
- * @return 1 if left input is greater, -1 if left input is smaller, 0 otherwise
- */
- public static final int compare(final DrillBuf left, int lStart, int lEnd, final byte[] right, int rStart, final int rEnd) {
- if(BOUNDS_CHECKING_ENABLED){
- left.checkBytes(lStart, lEnd);
- }
- return memcmp(left.memoryAddress(), lStart, lEnd, right, rStart, rEnd);
- }
-
-
- private static final int memcmp(final long laddr, int lStart, int lEnd, final byte[] right, int rStart, final int rEnd) {
- int lLen = lEnd - lStart;
- int rLen = rEnd - rStart;
- int n = Math.min(rLen, lLen);
- long lPos = laddr + lStart;
- int rPos = rStart;
-
- while (n-- != 0) {
- byte leftByte = PlatformDependent.getByte(lPos);
- byte rightByte = right[rPos];
- if (leftByte != rightByte) {
- return ((leftByte & 0xFF) - (rightByte & 0xFF)) > 0 ? 1 : -1;
- }
- lPos++;
- rPos++;
- }
-
- if (lLen == rLen) {
- return 0;
- }
-
- return lLen > rLen ? 1 : -1;
- }
-
- /*
- * Following are helper functions to interact with sparse decimal represented in a byte array.
- */
-
- // Get the integer ignore the sign
- public static int getInteger(byte[] b, int index) {
- return getInteger(b, index, true);
- }
- // Get the integer, ignore the sign
- public static int getInteger(byte[] b, int index, boolean ignoreSign) {
- int startIndex = index * DecimalUtility.integerSize;
-
- if (index == 0 && ignoreSign == true) {
- return (b[startIndex + 3] & 0xFF) |
- (b[startIndex + 2] & 0xFF) << 8 |
- (b[startIndex + 1] & 0xFF) << 16 |
- (b[startIndex] & 0x7F) << 24;
- }
-
- return ((b[startIndex + 3] & 0xFF) |
- (b[startIndex + 2] & 0xFF) << 8 |
- (b[startIndex + 1] & 0xFF) << 16 |
- (b[startIndex] & 0xFF) << 24);
-
- }
-
- // Set integer in the byte array
- public static void setInteger(byte[] b, int index, int value) {
- int startIndex = index * DecimalUtility.integerSize;
- b[startIndex] = (byte) ((value >> 24) & 0xFF);
- b[startIndex + 1] = (byte) ((value >> 16) & 0xFF);
- b[startIndex + 2] = (byte) ((value >> 8) & 0xFF);
- b[startIndex + 3] = (byte) ((value) & 0xFF);
- }
-
- // Set the sign in a sparse decimal representation
- public static void setSign(byte[] b, boolean sign) {
- int value = getInteger(b, 0);
- if (sign == true) {
- setInteger(b, 0, value | 0x80000000);
- } else {
- setInteger(b, 0, value & 0x7FFFFFFF);
- }
- }
-
- // Get the sign
- public static boolean getSign(byte[] b) {
- return ((getInteger(b, 0, false) & 0x80000000) != 0);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/drill/blob/4524fdbe/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/DateUtility.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/DateUtility.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/DateUtility.java
deleted file mode 100644
index 98a0b61..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/DateUtility.java
+++ /dev/null
@@ -1,682 +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.drill.exec.expr.fn.impl;
-
-import org.joda.time.Period;
-import org.joda.time.format.DateTimeFormat;
-import org.joda.time.format.DateTimeFormatter;
-import org.joda.time.format.DateTimeFormatterBuilder;
-import org.joda.time.format.DateTimeParser;
-
-import com.carrotsearch.hppc.ObjectIntOpenHashMap;
-
-// Utility class for Date, DateTime, TimeStamp, Interval data types
-public class DateUtility {
-
-
- /* We have a hashmap that stores the timezone as the key and an index as the value
- * While storing the timezone in value vectors, holders we only use this index. As we
- * reconstruct the timestamp, we use this index to index through the array timezoneList
- * and get the corresponding timezone and pass it to joda-time
- */
- public static ObjectIntOpenHashMap<String> timezoneMap = new ObjectIntOpenHashMap<String>();
-
- public static String[] timezoneList = {"Africa/Abidjan",
- "Africa/Accra",
- "Africa/Addis_Ababa",
- "Africa/Algiers",
- "Africa/Asmara",
- "Africa/Asmera",
- "Africa/Bamako",
- "Africa/Bangui",
- "Africa/Banjul",
- "Africa/Bissau",
- "Africa/Blantyre",
- "Africa/Brazzaville",
- "Africa/Bujumbura",
- "Africa/Cairo",
- "Africa/Casablanca",
- "Africa/Ceuta",
- "Africa/Conakry",
- "Africa/Dakar",
- "Africa/Dar_es_Salaam",
- "Africa/Djibouti",
- "Africa/Douala",
- "Africa/El_Aaiun",
- "Africa/Freetown",
- "Africa/Gaborone",
- "Africa/Harare",
- "Africa/Johannesburg",
- "Africa/Juba",
- "Africa/Kampala",
- "Africa/Khartoum",
- "Africa/Kigali",
- "Africa/Kinshasa",
- "Africa/Lagos",
- "Africa/Libreville",
- "Africa/Lome",
- "Africa/Luanda",
- "Africa/Lubumbashi",
- "Africa/Lusaka",
- "Africa/Malabo",
- "Africa/Maputo",
- "Africa/Maseru",
- "Africa/Mbabane",
- "Africa/Mogadishu",
- "Africa/Monrovia",
- "Africa/Nairobi",
- "Africa/Ndjamena",
- "Africa/Niamey",
- "Africa/Nouakchott",
- "Africa/Ouagadougou",
- "Africa/Porto-Novo",
- "Africa/Sao_Tome",
- "Africa/Timbuktu",
- "Africa/Tripoli",
- "Africa/Tunis",
- "Africa/Windhoek",
- "America/Adak",
- "America/Anchorage",
- "America/Anguilla",
- "America/Antigua",
- "America/Araguaina",
- "America/Argentina/Buenos_Aires",
- "America/Argentina/Catamarca",
- "America/Argentina/ComodRivadavia",
- "America/Argentina/Cordoba",
- "America/Argentina/Jujuy",
- "America/Argentina/La_Rioja",
- "America/Argentina/Mendoza",
- "America/Argentina/Rio_Gallegos",
- "America/Argentina/Salta",
- "America/Argentina/San_Juan",
- "America/Argentina/San_Luis",
- "America/Argentina/Tucuman",
- "America/Argentina/Ushuaia",
- "America/Aruba",
- "America/Asuncion",
- "America/Atikokan",
- "America/Atka",
- "America/Bahia",
- "America/Bahia_Banderas",
- "America/Barbados",
- "America/Belem",
- "America/Belize",
- "America/Blanc-Sablon",
- "America/Boa_Vista",
- "America/Bogota",
- "America/Boise",
- "America/Buenos_Aires",
- "America/Cambridge_Bay",
- "America/Campo_Grande",
- "America/Cancun",
- "America/Caracas",
- "America/Catamarca",
- "America/Cayenne",
- "America/Cayman",
- "America/Chicago",
- "America/Chihuahua",
- "America/Coral_Harbour",
- "America/Cordoba",
- "America/Costa_Rica",
- "America/Cuiaba",
- "America/Curacao",
- "America/Danmarkshavn",
- "America/Dawson",
- "America/Dawson_Creek",
- "America/Denver",
- "America/Detroit",
- "America/Dominica",
- "America/Edmonton",
- "America/Eirunepe",
- "America/El_Salvador",
- "America/Ensenada",
- "America/Fort_Wayne",
- "America/Fortaleza",
- "America/Glace_Bay",
- "America/Godthab",
- "America/Goose_Bay",
- "America/Grand_Turk",
- "America/Grenada",
- "America/Guadeloupe",
- "America/Guatemala",
- "America/Guayaquil",
- "America/Guyana",
- "America/Halifax",
- "America/Havana",
- "America/Hermosillo",
- "America/Indiana/Indianapolis",
- "America/Indiana/Knox",
- "America/Indiana/Marengo",
- "America/Indiana/Petersburg",
- "America/Indiana/Tell_City",
- "America/Indiana/Vevay",
- "America/Indiana/Vincennes",
- "America/Indiana/Winamac",
- "America/Indianapolis",
- "America/Inuvik",
- "America/Iqaluit",
- "America/Jamaica",
- "America/Jujuy",
- "America/Juneau",
- "America/Kentucky/Louisville",
- "America/Kentucky/Monticello",
- "America/Knox_IN",
- "America/Kralendijk",
- "America/La_Paz",
- "America/Lima",
- "America/Los_Angeles",
- "America/Louisville",
- "America/Lower_Princes",
- "America/Maceio",
- "America/Managua",
- "America/Manaus",
- "America/Marigot",
- "America/Martinique",
- "America/Matamoros",
- "America/Mazatlan",
- "America/Mendoza",
- "America/Menominee",
- "America/Merida",
- "America/Metlakatla",
- "America/Mexico_City",
- "America/Miquelon",
- "America/Moncton",
- "America/Monterrey",
- "America/Montevideo",
- "America/Montreal",
- "America/Montserrat",
- "America/Nassau",
- "America/New_York",
- "America/Nipigon",
- "America/Nome",
- "America/Noronha",
- "America/North_Dakota/Beulah",
- "America/North_Dakota/Center",
- "America/North_Dakota/New_Salem",
- "America/Ojinaga",
- "America/Panama",
- "America/Pangnirtung",
- "America/Paramaribo",
- "America/Phoenix",
- "America/Port-au-Prince",
- "America/Port_of_Spain",
- "America/Porto_Acre",
- "America/Porto_Velho",
- "America/Puerto_Rico",
- "America/Rainy_River",
- "America/Rankin_Inlet",
- "America/Recife",
- "America/Regina",
- "America/Resolute",
- "America/Rio_Branco",
- "America/Rosario",
- "America/Santa_Isabel",
- "America/Santarem",
- "America/Santiago",
- "America/Santo_Domingo",
- "America/Sao_Paulo",
- "America/Scoresbysund",
- "America/Shiprock",
- "America/Sitka",
- "America/St_Barthelemy",
- "America/St_Johns",
- "America/St_Kitts",
- "America/St_Lucia",
- "America/St_Thomas",
- "America/St_Vincent",
- "America/Swift_Current",
- "America/Tegucigalpa",
- "America/Thule",
- "America/Thunder_Bay",
- "America/Tijuana",
- "America/Toronto",
- "America/Tortola",
- "America/Vancouver",
- "America/Virgin",
- "America/Whitehorse",
- "America/Winnipeg",
- "America/Yakutat",
- "America/Yellowknife",
- "Antarctica/Casey",
- "Antarctica/Davis",
- "Antarctica/DumontDUrville",
- "Antarctica/Macquarie",
- "Antarctica/Mawson",
- "Antarctica/McMurdo",
- "Antarctica/Palmer",
- "Antarctica/Rothera",
- "Antarctica/South_Pole",
- "Antarctica/Syowa",
- "Antarctica/Vostok",
- "Arctic/Longyearbyen",
- "Asia/Aden",
- "Asia/Almaty",
- "Asia/Amman",
- "Asia/Anadyr",
- "Asia/Aqtau",
- "Asia/Aqtobe",
- "Asia/Ashgabat",
- "Asia/Ashkhabad",
- "Asia/Baghdad",
- "Asia/Bahrain",
- "Asia/Baku",
- "Asia/Bangkok",
- "Asia/Beirut",
- "Asia/Bishkek",
- "Asia/Brunei",
- "Asia/Calcutta",
- "Asia/Choibalsan",
- "Asia/Chongqing",
- "Asia/Chungking",
- "Asia/Colombo",
- "Asia/Dacca",
- "Asia/Damascus",
- "Asia/Dhaka",
- "Asia/Dili",
- "Asia/Dubai",
- "Asia/Dushanbe",
- "Asia/Gaza",
- "Asia/Harbin",
- "Asia/Hebron",
- "Asia/Ho_Chi_Minh",
- "Asia/Hong_Kong",
- "Asia/Hovd",
- "Asia/Irkutsk",
- "Asia/Istanbul",
- "Asia/Jakarta",
- "Asia/Jayapura",
- "Asia/Jerusalem",
- "Asia/Kabul",
- "Asia/Kamchatka",
- "Asia/Karachi",
- "Asia/Kashgar",
- "Asia/Kathmandu",
- "Asia/Katmandu",
- "Asia/Kolkata",
- "Asia/Krasnoyarsk",
- "Asia/Kuala_Lumpur",
- "Asia/Kuching",
- "Asia/Kuwait",
- "Asia/Macao",
- "Asia/Macau",
- "Asia/Magadan",
- "Asia/Makassar",
- "Asia/Manila",
- "Asia/Muscat",
- "Asia/Nicosia",
- "Asia/Novokuznetsk",
- "Asia/Novosibirsk",
- "Asia/Omsk",
- "Asia/Oral",
- "Asia/Phnom_Penh",
- "Asia/Pontianak",
- "Asia/Pyongyang",
- "Asia/Qatar",
- "Asia/Qyzylorda",
- "Asia/Rangoon",
- "Asia/Riyadh",
- "Asia/Saigon",
- "Asia/Sakhalin",
- "Asia/Samarkand",
- "Asia/Seoul",
- "Asia/Shanghai",
- "Asia/Singapore",
- "Asia/Taipei",
- "Asia/Tashkent",
- "Asia/Tbilisi",
- "Asia/Tehran",
- "Asia/Tel_Aviv",
- "Asia/Thimbu",
- "Asia/Thimphu",
- "Asia/Tokyo",
- "Asia/Ujung_Pandang",
- "Asia/Ulaanbaatar",
- "Asia/Ulan_Bator",
- "Asia/Urumqi",
- "Asia/Vientiane",
- "Asia/Vladivostok",
- "Asia/Yakutsk",
- "Asia/Yekaterinburg",
- "Asia/Yerevan",
- "Atlantic/Azores",
- "Atlantic/Bermuda",
- "Atlantic/Canary",
- "Atlantic/Cape_Verde",
- "Atlantic/Faeroe",
- "Atlantic/Faroe",
- "Atlantic/Jan_Mayen",
- "Atlantic/Madeira",
- "Atlantic/Reykjavik",
- "Atlantic/South_Georgia",
- "Atlantic/St_Helena",
- "Atlantic/Stanley",
- "Australia/ACT",
- "Australia/Adelaide",
- "Australia/Brisbane",
- "Australia/Broken_Hill",
- "Australia/Canberra",
- "Australia/Currie",
- "Australia/Darwin",
- "Australia/Eucla",
- "Australia/Hobart",
- "Australia/LHI",
- "Australia/Lindeman",
- "Australia/Lord_Howe",
- "Australia/Melbourne",
- "Australia/NSW",
- "Australia/North",
- "Australia/Perth",
- "Australia/Queensland",
- "Australia/South",
- "Australia/Sydney",
- "Australia/Tasmania",
- "Australia/Victoria",
- "Australia/West",
- "Australia/Yancowinna",
- "Brazil/Acre",
- "Brazil/DeNoronha",
- "Brazil/East",
- "Brazil/West",
- "CET",
- "CST6CDT",
- "Canada/Atlantic",
- "Canada/Central",
- "Canada/East-Saskatchewan",
- "Canada/Eastern",
- "Canada/Mountain",
- "Canada/Newfoundland",
- "Canada/Pacific",
- "Canada/Saskatchewan",
- "Canada/Yukon",
- "Chile/Continental",
- "Chile/EasterIsland",
- "Cuba",
- "EET",
- "EST",
- "EST5EDT",
- "Egypt",
- "Eire",
- "Etc/GMT",
- "Etc/GMT+0",
- "Etc/GMT+1",
- "Etc/GMT+10",
- "Etc/GMT+11",
- "Etc/GMT+12",
- "Etc/GMT+2",
- "Etc/GMT+3",
- "Etc/GMT+4",
- "Etc/GMT+5",
- "Etc/GMT+6",
- "Etc/GMT+7",
- "Etc/GMT+8",
- "Etc/GMT+9",
- "Etc/GMT-0",
- "Etc/GMT-1",
- "Etc/GMT-10",
- "Etc/GMT-11",
- "Etc/GMT-12",
- "Etc/GMT-13",
- "Etc/GMT-14",
- "Etc/GMT-2",
- "Etc/GMT-3",
- "Etc/GMT-4",
- "Etc/GMT-5",
- "Etc/GMT-6",
- "Etc/GMT-7",
- "Etc/GMT-8",
- "Etc/GMT-9",
- "Etc/GMT0",
- "Etc/Greenwich",
- "Etc/UCT",
- "Etc/UTC",
- "Etc/Universal",
- "Etc/Zulu",
- "Europe/Amsterdam",
- "Europe/Andorra",
- "Europe/Athens",
- "Europe/Belfast",
- "Europe/Belgrade",
- "Europe/Berlin",
- "Europe/Bratislava",
- "Europe/Brussels",
- "Europe/Bucharest",
- "Europe/Budapest",
- "Europe/Chisinau",
- "Europe/Copenhagen",
- "Europe/Dublin",
- "Europe/Gibraltar",
- "Europe/Guernsey",
- "Europe/Helsinki",
- "Europe/Isle_of_Man",
- "Europe/Istanbul",
- "Europe/Jersey",
- "Europe/Kaliningrad",
- "Europe/Kiev",
- "Europe/Lisbon",
- "Europe/Ljubljana",
- "Europe/London",
- "Europe/Luxembourg",
- "Europe/Madrid",
- "Europe/Malta",
- "Europe/Mariehamn",
- "Europe/Minsk",
- "Europe/Monaco",
- "Europe/Moscow",
- "Europe/Nicosia",
- "Europe/Oslo",
- "Europe/Paris",
- "Europe/Podgorica",
- "Europe/Prague",
- "Europe/Riga",
- "Europe/Rome",
- "Europe/Samara",
- "Europe/San_Marino",
- "Europe/Sarajevo",
- "Europe/Simferopol",
- "Europe/Skopje",
- "Europe/Sofia",
- "Europe/Stockholm",
- "Europe/Tallinn",
- "Europe/Tirane",
- "Europe/Tiraspol",
- "Europe/Uzhgorod",
- "Europe/Vaduz",
- "Europe/Vatican",
- "Europe/Vienna",
- "Europe/Vilnius",
- "Europe/Volgograd",
- "Europe/Warsaw",
- "Europe/Zagreb",
- "Europe/Zaporozhye",
- "Europe/Zurich",
- "GB",
- "GB-Eire",
- "GMT",
- "GMT+0",
- "GMT-0",
- "GMT0",
- "Greenwich",
- "HST",
- "Hongkong",
- "Iceland",
- "Indian/Antananarivo",
- "Indian/Chagos",
- "Indian/Christmas",
- "Indian/Cocos",
- "Indian/Comoro",
- "Indian/Kerguelen",
- "Indian/Mahe",
- "Indian/Maldives",
- "Indian/Mauritius",
- "Indian/Mayotte",
- "Indian/Reunion",
- "Iran",
- "Israel",
- "Jamaica",
- "Japan",
- "Kwajalein",
- "Libya",
- "MET",
- "MST",
- "MST7MDT",
- "Mexico/BajaNorte",
- "Mexico/BajaSur",
- "Mexico/General",
- "NZ",
- "NZ-CHAT",
- "Navajo",
- "PRC",
- "PST8PDT",
- "Pacific/Apia",
- "Pacific/Auckland",
- "Pacific/Chatham",
- "Pacific/Chuuk",
- "Pacific/Easter",
- "Pacific/Efate",
- "Pacific/Enderbury",
- "Pacific/Fakaofo",
- "Pacific/Fiji",
- "Pacific/Funafuti",
- "Pacific/Galapagos",
- "Pacific/Gambier",
- "Pacific/Guadalcanal",
- "Pacific/Guam",
- "Pacific/Honolulu",
- "Pacific/Johnston",
- "Pacific/Kiritimati",
- "Pacific/Kosrae",
- "Pacific/Kwajalein",
- "Pacific/Majuro",
- "Pacific/Marquesas",
- "Pacific/Midway",
- "Pacific/Nauru",
- "Pacific/Niue",
- "Pacific/Norfolk",
- "Pacific/Noumea",
- "Pacific/Pago_Pago",
- "Pacific/Palau",
- "Pacific/Pitcairn",
- "Pacific/Pohnpei",
- "Pacific/Ponape",
- "Pacific/Port_Moresby",
- "Pacific/Rarotonga",
- "Pacific/Saipan",
- "Pacific/Samoa",
- "Pacific/Tahiti",
- "Pacific/Tarawa",
- "Pacific/Tongatapu",
- "Pacific/Truk",
- "Pacific/Wake",
- "Pacific/Wallis",
- "Pacific/Yap",
- "Poland",
- "Portugal",
- "ROC",
- "ROK",
- "Singapore",
- "Turkey",
- "UCT",
- "US/Alaska",
- "US/Aleutian",
- "US/Arizona",
- "US/Central",
- "US/East-Indiana",
- "US/Eastern",
- "US/Hawaii",
- "US/Indiana-Starke",
- "US/Michigan",
- "US/Mountain",
- "US/Pacific",
- "US/Pacific-New",
- "US/Samoa",
- "UTC",
- "Universal",
- "W-SU",
- "WET",
- "Zulu"};
-
- static {
- for (int i = 0; i < timezoneList.length; i++) {
- timezoneMap.put(timezoneList[i], i);
- }
- }
-
- public static final DateTimeFormatter formatDate = DateTimeFormat.forPattern("yyyy-MM-dd");
- public static final DateTimeFormatter formatTimeStamp = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS");
- public static final DateTimeFormatter formatTimeStampTZ = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS ZZZ");
- public static final DateTimeFormatter formatTime = DateTimeFormat.forPattern("HH:mm:ss.SSS");
-
- public static DateTimeFormatter dateTimeTZFormat = null;
- public static DateTimeFormatter timeFormat = null;
-
- public static final int yearsToMonths = 12;
- public static final int hoursToMillis = 60 * 60 * 1000;
- public static final int minutesToMillis = 60 * 1000;
- public static final int secondsToMillis = 1000;
- public static final int monthToStandardDays = 30;
- public static final long monthsToMillis = 2592000000L; // 30 * 24 * 60 * 60 * 1000
- public static final int daysToStandardMillis = 24 * 60 * 60 * 1000;
-
-
- public static int getIndex(String timezone) {
- return timezoneMap.get(timezone);
- }
-
- public static String getTimeZone(int index) {
- return timezoneList[index];
- }
-
- // Function returns the date time formatter used to parse date strings
- public static DateTimeFormatter getDateTimeFormatter() {
-
- if (dateTimeTZFormat == null) {
- DateTimeFormatter dateFormatter = DateTimeFormat.forPattern("yyyy-MM-dd");
- DateTimeParser optionalTime = DateTimeFormat.forPattern(" HH:mm:ss").getParser();
- DateTimeParser optionalSec = DateTimeFormat.forPattern(".SSS").getParser();
- DateTimeParser optionalZone = DateTimeFormat.forPattern(" ZZZ").getParser();
-
- dateTimeTZFormat = new DateTimeFormatterBuilder().append(dateFormatter).appendOptional(optionalTime).appendOptional(optionalSec).appendOptional(optionalZone).toFormatter();
- }
-
- return dateTimeTZFormat;
- }
-
- // Function returns time formatter used to parse time strings
- public static DateTimeFormatter getTimeFormatter() {
- if (timeFormat == null) {
- DateTimeFormatter timeFormatter = DateTimeFormat.forPattern("HH:mm:ss");
- DateTimeParser optionalSec = DateTimeFormat.forPattern(".SSS").getParser();
- timeFormat = new DateTimeFormatterBuilder().append(timeFormatter).appendOptional(optionalSec).toFormatter();
- }
- return timeFormat;
- }
-
- public static int monthsFromPeriod(Period period){
- return (period.getYears() * yearsToMonths) + period.getMonths();
- }
-
- public static int millisFromPeriod(final Period period){
- return (period.getHours() * hoursToMillis) +
- (period.getMinutes() * minutesToMillis) +
- (period.getSeconds() * secondsToMillis) +
- (period.getMillis());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/drill/blob/4524fdbe/exec/java-exec/src/main/java/org/apache/drill/exec/memory/Accountor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/Accountor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/Accountor.java
deleted file mode 100644
index ad6a787..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/Accountor.java
+++ /dev/null
@@ -1,462 +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.drill.exec.memory;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.DrillBuf;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.concurrent.ConcurrentMap;
-
-import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.exec.ExecConstants;
-import org.apache.drill.exec.ops.FragmentContext;
-import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
-import org.apache.drill.exec.proto.helper.QueryIdHelper;
-import org.apache.drill.exec.util.AssertionUtil;
-
-import com.google.common.collect.LinkedListMultimap;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Multimap;
-
-public class Accountor {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Accountor.class);
-
- private static final boolean ENABLE_ACCOUNTING = AssertionUtil.isAssertionsEnabled();
- private final AtomicRemainder remainder;
- private final long total;
- private ConcurrentMap<ByteBuf, DebugStackTrace> buffers = Maps.newConcurrentMap();
- private final FragmentHandle handle;
- private String fragmentStr;
- private Accountor parent;
-
- private final boolean errorOnLeak;
- // some operators are no subject to the fragment limit. They set the applyFragmentLimit to false
-
- private final boolean enableFragmentLimit;
- private final double fragmentMemOvercommitFactor;
-
- private final boolean DEFAULT_ENABLE_FRAGMENT_LIMIT=false;
- private final double DEFAULT_FRAGMENT_MEM_OVERCOMMIT_FACTOR=1.5;
-
- private final boolean applyFragmentLimit;
-
- private final FragmentContext fragmentContext;
- long fragmentLimit;
-
- private long peakMemoryAllocation = 0;
-
- // The top level Allocator has an accountor that keeps track of all the FragmentContexts currently executing.
- // This enables the top level accountor to calculate a new fragment limit whenever necessary.
- private final List<FragmentContext> fragmentContexts;
-
- public Accountor(DrillConfig config, boolean errorOnLeak, FragmentContext context, Accountor parent, long max, long preAllocated, boolean applyFragLimit) {
- // TODO: fix preallocation stuff
- this.errorOnLeak = errorOnLeak;
- AtomicRemainder parentRemainder = parent != null ? parent.remainder : null;
- this.parent = parent;
-
- boolean enableFragmentLimit;
- double fragmentMemOvercommitFactor;
-
- try {
- enableFragmentLimit = config.getBoolean(ExecConstants.ENABLE_FRAGMENT_MEMORY_LIMIT);
- fragmentMemOvercommitFactor = config.getDouble(ExecConstants.FRAGMENT_MEM_OVERCOMMIT_FACTOR);
- }catch(Exception e){
- enableFragmentLimit = DEFAULT_ENABLE_FRAGMENT_LIMIT;
- fragmentMemOvercommitFactor = DEFAULT_FRAGMENT_MEM_OVERCOMMIT_FACTOR;
- }
- this.enableFragmentLimit = enableFragmentLimit;
- this.fragmentMemOvercommitFactor = fragmentMemOvercommitFactor;
-
-
- this.applyFragmentLimit=applyFragLimit;
-
- this.remainder = new AtomicRemainder(errorOnLeak, parentRemainder, max, preAllocated, applyFragmentLimit);
- this.total = max;
- this.fragmentContext=context;
- this.handle = (context!=null) ? context.getHandle() : null;
- this.fragmentStr= (handle!=null) ? ( handle.getMajorFragmentId()+":"+handle.getMinorFragmentId() ) : "0:0";
- this.fragmentLimit=this.total; // Allow as much as possible to start with;
- if (ENABLE_ACCOUNTING) {
- buffers = Maps.newConcurrentMap();
- } else {
- buffers = null;
- }
- this.fragmentContexts = new ArrayList<FragmentContext>();
- if(parent!=null && parent.parent==null){ // Only add the fragment context to the fragment level accountor
- synchronized(this) {
- addFragmentContext(this.fragmentContext);
- }
- }
- }
-
- public boolean transferTo(Accountor target, DrillBuf buf, long size) {
- return transfer(target, buf, size, true);
- }
-
- public boolean transferIn(DrillBuf buf, long size) {
- return transfer(this, buf, size, false);
- }
-
- private boolean transfer(Accountor target, DrillBuf buf, long size, boolean release) {
- boolean withinLimit = target.forceAdditionalReservation(size);
- if(release){
- release(buf, size);
- }
-
- if (ENABLE_ACCOUNTING) {
- target.buffers.put(buf, new DebugStackTrace(buf.capacity(), Thread.currentThread().getStackTrace()));
- }
- return withinLimit;
- }
-
- public long getAvailable() {
- if (parent != null) {
- return Math.min(parent.getAvailable(), getCapacity() - getAllocation());
- }
- return getCapacity() - getAllocation();
- }
-
- public long getCapacity() {
- return fragmentLimit;
- }
-
- public long getAllocation() {
- return remainder.getUsed();
- }
-
- public long getPeakMemoryAllocation() {
- return peakMemoryAllocation;
- }
-
- public boolean reserve(long size) {
- logger.trace("Fragment:"+fragmentStr+" Reserved "+size+" bytes. Total Allocated: "+getAllocation());
- boolean status = remainder.get(size, this.applyFragmentLimit);
- peakMemoryAllocation = Math.max(peakMemoryAllocation, getAllocation());
- return status;
- }
-
- public boolean forceAdditionalReservation(long size) {
- if (size > 0) {
- boolean status = remainder.forceGet(size);
- peakMemoryAllocation = Math.max(peakMemoryAllocation, getAllocation());
- return status;
- } else {
- return true;
- }
- }
-
- public void reserved(long expected, DrillBuf buf) {
- // make sure to take away the additional memory that happened due to rounding.
-
- long additional = buf.capacity() - expected;
- if (additional > 0) {
- remainder.forceGet(additional);
- }
-
- if (ENABLE_ACCOUNTING) {
- buffers.put(buf, new DebugStackTrace(buf.capacity(), Thread.currentThread().getStackTrace()));
- }
-
- peakMemoryAllocation = Math.max(peakMemoryAllocation, getAllocation());
- }
-
-
- public void releasePartial(DrillBuf buf, long size) {
- remainder.returnAllocation(size);
- if (ENABLE_ACCOUNTING) {
- if (buf != null) {
- DebugStackTrace dst = buffers.get(buf);
- if (dst == null) {
- throw new IllegalStateException("Partially releasing a buffer that has already been released. Buffer: " + buf);
- }
- dst.size -= size;
- if (dst.size < 0) {
- throw new IllegalStateException("Partially releasing a buffer that has already been released. Buffer: " + buf);
- }
- }
- }
- }
-
- void release(long size) {
- remainder.returnAllocation(size);
- }
-
- public void release(DrillBuf buf, long size) {
- remainder.returnAllocation(size);
- if (ENABLE_ACCOUNTING) {
- if (buf != null && buffers.remove(buf) == null) {
- throw new IllegalStateException("Releasing a buffer that has already been released. Buffer: " + buf);
- }
- }
- }
-
- private void addFragmentContext(FragmentContext c) {
- if (parent != null){
- parent.addFragmentContext(c);
- }else {
- if(logger.isTraceEnabled()) {
- FragmentHandle hndle;
- String fragStr;
- if(c!=null) {
- hndle = c.getHandle();
- fragStr = (hndle != null) ? (hndle.getMajorFragmentId() + ":" + hndle.getMinorFragmentId()) : "[Null Fragment Handle]";
- }else{
- fragStr = "[Null Context]";
- }
- fragStr+=" (Object Id: "+System.identityHashCode(c)+")";
- StackTraceElement[] ste = (new Throwable()).getStackTrace();
- StringBuffer sb = new StringBuffer();
- for (StackTraceElement s : ste) {
- sb.append(s.toString());
- sb.append("\n");
- }
-
- logger.trace("Fragment " + fragStr + " added to root accountor.\n"+sb.toString());
- }
- synchronized(this) {
- fragmentContexts.add(c);
- }
- }
- }
-
- private void removeFragmentContext(FragmentContext c) {
- if (parent != null){
- if (parent.parent==null){
- // only fragment level allocators will have the fragment context saved
- parent.removeFragmentContext(c);
- }
- }else{
- if(logger.isDebugEnabled()) {
- FragmentHandle hndle;
- String fragStr;
- if (c != null) {
- hndle = c.getHandle();
- fragStr = (hndle != null) ? (hndle.getMajorFragmentId() + ":" + hndle.getMinorFragmentId()) : "[Null Fragment Handle]";
- } else {
- fragStr = "[Null Context]";
- }
- fragStr += " (Object Id: " + System.identityHashCode(c) + ")";
- logger.trace("Fragment " + fragStr + " removed from root accountor");
- }
- synchronized(this) {
- fragmentContexts.remove(c);
- }
- }
- }
-
- public long resetFragmentLimits(){
- // returns the new capacity
- if(!this.enableFragmentLimit){
- return getCapacity();
- }
-
- if(parent!=null){
- parent.resetFragmentLimits();
- }else {
- //Get remaining memory available per fragment and distribute it EQUALLY among all the fragments.
- //Fragments get the memory limit added to the amount already allocated.
- //This favours fragments that are already running which will get a limit greater than newly started fragments.
- //If the already running fragments end quickly, their limits will be assigned back to the remaining fragments
- //quickly. If they are long running, then we want to favour them with larger limits anyway.
- synchronized (this) {
- int nFragments=fragmentContexts.size();
- long allocatedMemory=0;
- for(FragmentContext fragment: fragmentContexts){
- BufferAllocator a = fragment.getAllocator();
- if(a!=null) {
- allocatedMemory += fragment.getAllocator().getAllocatedMemory();
- }
- }
- if(logger.isTraceEnabled()) {
- logger.trace("Resetting Fragment Memory Limit: total Available memory== "+total
- +" Total Allocated Memory :"+allocatedMemory
- +" Number of fragments: "+nFragments
- + " fragmentMemOvercommitFactor: "+fragmentMemOvercommitFactor
- + " Root fragment limit: "+this.fragmentLimit + "(Root obj: "+System.identityHashCode(this)+")"
- );
- }
- if(nFragments>0) {
- long rem = (total - allocatedMemory) / nFragments;
- for (FragmentContext fragment : fragmentContexts) {
- fragment.setFragmentLimit((long) (rem * fragmentMemOvercommitFactor));
- }
- }
- if(logger.isTraceEnabled() && false){
- StringBuffer sb= new StringBuffer();
- sb.append("[root](0:0)");
- sb.append("Allocated memory: ");
- sb.append(this.getAllocation());
- sb.append(" Fragment Limit: ");
- sb.append(this.getFragmentLimit());
- logger.trace(sb.toString());
- for(FragmentContext fragment: fragmentContexts){
- sb= new StringBuffer();
- if (handle != null) {
- sb.append("[");
- sb.append(QueryIdHelper.getQueryId(handle.getQueryId()));
- sb.append("](");
- sb.append(handle.getMajorFragmentId());
- sb.append(":");
- sb.append(handle.getMinorFragmentId());
- sb.append(")");
- }else{
- sb.append("[fragment](0:0)");
- }
- sb.append("Allocated memory: ");
- sb.append(fragment.getAllocator().getAllocatedMemory());
- sb.append(" Fragment Limit: ");
- sb.append(fragment.getAllocator().getFragmentLimit());
- logger.trace(sb.toString());
- }
- logger.trace("Resetting Complete");
- }
- }
- }
- return getCapacity();
- }
-
- public void close() {
- // remove the fragment context and reset fragment limits whenever an allocator closes
- if(parent!=null && parent.parent==null) {
- logger.debug("Fragment " + fragmentStr + " accountor being closed");
- removeFragmentContext(fragmentContext);
- }
- resetFragmentLimits();
-
- if (ENABLE_ACCOUNTING && !buffers.isEmpty()) {
- StringBuffer sb = new StringBuffer();
- sb.append("Attempted to close accountor with ");
- sb.append(buffers.size());
- sb.append(" buffer(s) still allocated");
- if (handle != null) {
- sb.append("for QueryId: ");
- sb.append(QueryIdHelper.getQueryId(handle.getQueryId()));
- sb.append(", MajorFragmentId: ");
- sb.append(handle.getMajorFragmentId());
- sb.append(", MinorFragmentId: ");
- sb.append(handle.getMinorFragmentId());
- }
- sb.append(".\n");
-
- Multimap<DebugStackTrace, DebugStackTrace> multi = LinkedListMultimap.create();
- for (DebugStackTrace t : buffers.values()) {
- multi.put(t, t);
- }
-
- for (DebugStackTrace entry : multi.keySet()) {
- Collection<DebugStackTrace> allocs = multi.get(entry);
-
- sb.append("\n\n\tTotal ");
- sb.append(allocs.size());
- sb.append(" allocation(s) of byte size(s): ");
- for (DebugStackTrace alloc : allocs) {
- sb.append(alloc.size);
- sb.append(", ");
- }
-
- sb.append("at stack location:\n");
- entry.addToString(sb);
- }
- if (!buffers.isEmpty()) {
- IllegalStateException e = new IllegalStateException(sb.toString());
- if (errorOnLeak) {
- throw e;
- } else {
- logger.warn("Memory leaked.", e);
- }
- }
- }
-
- remainder.close();
-
- }
-
- public void setFragmentLimit(long add) {
- // We ADD the limit to the current allocation. If none has been allocated, this
- // sets a new limit. If memory has already been allocated, the fragment gets its
- // limit based on the allocation, though this might still result in reducing the
- // limit.
-
- if (parent != null && parent.parent==null) { // This is a fragment level accountor
- this.fragmentLimit=getAllocation()+add;
- this.remainder.setLimit(this.fragmentLimit);
- logger.trace("Fragment "+fragmentStr+" memory limit set to "+this.fragmentLimit);
- }
- }
-
- public long getFragmentLimit(){
- return this.fragmentLimit;
- }
-
- public class DebugStackTrace {
-
- private StackTraceElement[] elements;
- private long size;
-
- public DebugStackTrace(long size, StackTraceElement[] elements) {
- super();
- this.elements = elements;
- this.size = size;
- }
-
- public void addToString(StringBuffer sb) {
- for (int i = 3; i < elements.length; i++) {
- sb.append("\t\t");
- sb.append(elements[i]);
- sb.append("\n");
- }
- }
-
- @Override
- public int hashCode() {
- final int prime = 31;
- int result = 1;
- result = prime * result + Arrays.hashCode(elements);
-// result = prime * result + (int) (size ^ (size >>> 32));
- return result;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null) {
- return false;
- }
- if (getClass() != obj.getClass()) {
- return false;
- }
- DebugStackTrace other = (DebugStackTrace) obj;
- if (!Arrays.equals(elements, other.elements)) {
- return false;
- }
- // weird equal where size doesn't matter for multimap purposes.
-// if (size != other.size)
-// return false;
- return true;
- }
-
- }
-
-}