You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by su...@apache.org on 2016/11/05 01:01:30 UTC
[4/4] drill git commit: DRILL-4823: Fix OOM while trying to prune
partitions with reasonable data size
DRILL-4823: Fix OOM while trying to prune partitions with reasonable data size
closes #560
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/010aa6aa
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/010aa6aa
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/010aa6aa
Branch: refs/heads/master
Commit: 010aa6aac62ece002889a94aaec122e2adf5759f
Parents: 0672d49
Author: Arina Ielchiieva <ar...@gmail.com>
Authored: Wed Aug 3 11:56:43 2016 +0000
Committer: Sudheesh Katkam <sk...@maprtech.com>
Committed: Fri Nov 4 17:17:06 2016 -0700
----------------------------------------------------------------------
.../fn/interpreter/InterpreterEvaluator.java | 42 +++++++++++++++-----
.../apache/drill/exec/ops/FragmentContext.java | 21 ++++++++++
.../org/apache/drill/exec/ops/QueryContext.java | 23 +++++++++++
.../org/apache/drill/exec/ops/UdfUtilities.java | 11 +++++
4 files changed, 87 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/010aa6aa/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/interpreter/InterpreterEvaluator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/interpreter/InterpreterEvaluator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/interpreter/InterpreterEvaluator.java
index d106887..2b9ac0a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/interpreter/InterpreterEvaluator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/interpreter/InterpreterEvaluator.java
@@ -20,8 +20,10 @@ package org.apache.drill.exec.expr.fn.interpreter;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
+import javax.annotation.Nullable;
import javax.inject.Inject;
+import com.google.common.base.Function;
import org.apache.drill.common.exceptions.DrillRuntimeException;
import org.apache.drill.common.expression.BooleanOperator;
import org.apache.drill.common.expression.ConvertExpression;
@@ -35,6 +37,7 @@ import org.apache.drill.common.expression.TypedNullConstant;
import org.apache.drill.common.expression.ValueExpressions;
import org.apache.drill.common.expression.visitors.AbstractExprVisitor;
import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.common.types.TypeProtos.MinorType;
import org.apache.drill.exec.expr.DrillFuncHolderExpr;
import org.apache.drill.exec.expr.DrillSimpleFunc;
import org.apache.drill.exec.expr.TypeHelper;
@@ -197,10 +200,6 @@ public class InterpreterEvaluator {
this.udfUtilities = udfUtilities;
}
- public DrillBuf getManagedBufferIfAvailable() {
- return udfUtilities.getManagedBuffer();
- }
-
@Override
public ValueHolder visitFunctionCall(FunctionCall call, Integer value) throws RuntimeException {
return visitUnknown(call, value);
@@ -222,13 +221,25 @@ public class InterpreterEvaluator {
}
@Override
- public ValueHolder visitDecimal28Constant(ValueExpressions.Decimal28Expression decExpr,Integer value) throws RuntimeException {
- return ValueHolderHelper.getDecimal28Holder(getManagedBufferIfAvailable(), decExpr.getBigDecimal().toString());
+ public ValueHolder visitDecimal28Constant(final ValueExpressions.Decimal28Expression decExpr,Integer value) throws RuntimeException {
+ return getConstantValueHolder(decExpr.getBigDecimal().toString(), decExpr.getMajorType().getMinorType(), new Function<DrillBuf, ValueHolder>() {
+ @Nullable
+ @Override
+ public ValueHolder apply(DrillBuf buffer) {
+ return ValueHolderHelper.getDecimal28Holder(buffer, decExpr.getBigDecimal().toString());
+ }
+ });
}
@Override
- public ValueHolder visitDecimal38Constant(ValueExpressions.Decimal38Expression decExpr,Integer value) throws RuntimeException {
- return ValueHolderHelper.getDecimal38Holder(getManagedBufferIfAvailable(), decExpr.getBigDecimal().toString());
+ public ValueHolder visitDecimal38Constant(final ValueExpressions.Decimal38Expression decExpr,Integer value) throws RuntimeException {
+ return getConstantValueHolder(decExpr.getBigDecimal().toString(), decExpr.getMajorType().getMinorType(), new Function<DrillBuf, ValueHolder>() {
+ @Nullable
+ @Override
+ public ValueHolder apply(DrillBuf buffer) {
+ return ValueHolderHelper.getDecimal38Holder(buffer, decExpr.getBigDecimal().toString());
+ }
+ });
}
@Override
@@ -385,8 +396,14 @@ public class InterpreterEvaluator {
}
@Override
- public ValueHolder visitQuotedStringConstant(ValueExpressions.QuotedString e, Integer value) throws RuntimeException {
- return ValueHolderHelper.getVarCharHolder(getManagedBufferIfAvailable(), e.value);
+ public ValueHolder visitQuotedStringConstant(final ValueExpressions.QuotedString e, Integer value) throws RuntimeException {
+ return getConstantValueHolder(e.value, e.getMajorType().getMinorType(), new Function<DrillBuf, ValueHolder>() {
+ @Nullable
+ @Override
+ public ValueHolder apply(DrillBuf buffer) {
+ return ValueHolderHelper.getVarCharHolder(buffer, e.value);
+ }
+ });
}
@@ -507,6 +524,11 @@ public class InterpreterEvaluator {
return Trivalent.FALSE;
}
}
+
+ private ValueHolder getConstantValueHolder(String value, MinorType type, Function<DrillBuf, ValueHolder> holderInitializer) {
+ return udfUtilities.getConstantValueHolder(value, type, holderInitializer);
+ }
+
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/010aa6aa/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
index 8506b91..8229b5b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.ops;
+import com.google.common.base.Function;
import io.netty.buffer.DrillBuf;
import java.io.IOException;
@@ -28,12 +29,14 @@ import org.apache.calcite.schema.SchemaPlus;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.types.TypeProtos.MinorType;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.exception.ClassTransformationException;
import org.apache.drill.exec.exception.OutOfMemoryException;
import org.apache.drill.exec.expr.ClassGenerator;
import org.apache.drill.exec.expr.CodeGenerator;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.expr.holders.ValueHolder;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.planner.physical.PlannerSettings;
@@ -103,6 +106,8 @@ public class FragmentContext implements AutoCloseable, UdfUtilities {
private final RpcOutcomeListener<Ack> statusHandler = new StatusHandler(exceptionConsumer, sendingAccountor);
private final AccountingUserConnection accountingUserConnection;
+ /** Stores constants and their holders by type */
+ private final Map<String, Map<MinorType, ValueHolder>> constantValueHolderCache;
/**
* Create a FragmentContext instance for non-root fragment.
@@ -173,6 +178,7 @@ public class FragmentContext implements AutoCloseable, UdfUtilities {
stats = new FragmentStats(allocator, fragment.getAssignment());
bufferManager = new BufferManagerImpl(this.allocator);
+ constantValueHolderCache = Maps.newHashMap();
}
/**
@@ -440,6 +446,21 @@ public class FragmentContext implements AutoCloseable, UdfUtilities {
"option is set to true.", PlannerSettings.CONSTANT_FOLDING.getOptionName()));
}
+ @Override
+ public ValueHolder getConstantValueHolder(String value, MinorType type, Function<DrillBuf, ValueHolder> holderInitializer) {
+ if (!constantValueHolderCache.containsKey(value)) {
+ constantValueHolderCache.put(value, Maps.<MinorType, ValueHolder>newHashMap());
+ }
+
+ Map<MinorType, ValueHolder> holdersByType = constantValueHolderCache.get(value);
+ ValueHolder valueHolder = holdersByType.get(type);
+ if (valueHolder == null) {
+ valueHolder = holderInitializer.apply(getManagedBuffer());
+ holdersByType.put(type, valueHolder);
+ }
+ return valueHolder;
+ }
+
public Executor getExecutor(){
return context.getExecutor();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/010aa6aa/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java
index ceb1224..918fe01 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java
@@ -17,18 +17,23 @@
*/
package org.apache.drill.exec.ops;
+import com.google.common.base.Function;
+import com.google.common.collect.Maps;
import io.netty.buffer.DrillBuf;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.drill.common.AutoCloseables;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.config.LogicalPlanPersistence;
+import org.apache.drill.common.types.TypeProtos.MinorType;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
import org.apache.drill.exec.expr.fn.registry.RemoteFunctionRegistry;
+import org.apache.drill.exec.expr.holders.ValueHolder;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.planner.sql.DrillOperatorTable;
@@ -70,6 +75,8 @@ public class QueryContext implements AutoCloseable, OptimizerRulesContext, Schem
private final QueryContextInformation queryContextInfo;
private final ViewExpansionContext viewExpansionContext;
private final SchemaTreeProvider schemaTreeProvider;
+ /** Stores constants and their holders by type */
+ private final Map<String, Map<MinorType, ValueHolder>> constantValueHolderCache;
/*
* Flag to indicate if close has been called, after calling close the first
@@ -96,6 +103,7 @@ public class QueryContext implements AutoCloseable, OptimizerRulesContext, Schem
bufferManager = new BufferManagerImpl(this.allocator);
viewExpansionContext = new ViewExpansionContext(this);
schemaTreeProvider = new SchemaTreeProvider(drillbitContext);
+ constantValueHolderCache = Maps.newHashMap();
}
@Override
@@ -243,6 +251,21 @@ public class QueryContext implements AutoCloseable, OptimizerRulesContext, Schem
}
@Override
+ public ValueHolder getConstantValueHolder(String value, MinorType type, Function<DrillBuf, ValueHolder> holderInitializer) {
+ if (!constantValueHolderCache.containsKey(value)) {
+ constantValueHolderCache.put(value, Maps.<MinorType, ValueHolder>newHashMap());
+ }
+
+ Map<MinorType, ValueHolder> holdersByType = constantValueHolderCache.get(value);
+ ValueHolder valueHolder = holdersByType.get(type);
+ if (valueHolder == null) {
+ valueHolder = holderInitializer.apply(getManagedBuffer());
+ holdersByType.put(type, valueHolder);
+ }
+ return valueHolder;
+ }
+
+ @Override
public void close() throws Exception {
try {
if (!closed) {
http://git-wip-us.apache.org/repos/asf/drill/blob/010aa6aa/exec/java-exec/src/main/java/org/apache/drill/exec/ops/UdfUtilities.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/UdfUtilities.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/UdfUtilities.java
index 5df2b1b..6752d76 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/UdfUtilities.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/UdfUtilities.java
@@ -17,6 +17,9 @@
******************************************************************************/
package org.apache.drill.exec.ops;
+import com.google.common.base.Function;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.exec.expr.holders.ValueHolder;
import org.apache.drill.exec.store.PartitionExplorer;
import com.google.common.collect.ImmutableMap;
@@ -84,4 +87,12 @@ public interface UdfUtilities {
* @return - an object for exploring partitions of all available schemas
*/
PartitionExplorer getPartitionExplorer();
+
+ /**
+ * Works with value holders cache which holds constant value and its wrapper by type.
+ * If value is absent uses holderInitializer to create holder and adds it to cache.
+ *
+ * @return - a wrapper object for an constant value.
+ */
+ ValueHolder getConstantValueHolder(String value, MinorType type, Function<DrillBuf, ValueHolder> holderInitializer);
}