You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by ji...@apache.org on 2015/04/18 04:57:54 UTC
[1/4] tajo git commit: TAJO-1344: Python UDF support. (jihoon)
Repository: tajo
Updated Branches:
refs/heads/master 14a1e536c -> a74538539
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/LogicalPlanEqualityTester.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/LogicalPlanEqualityTester.java b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/LogicalPlanEqualityTester.java
index 8a24add..43a8618 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/LogicalPlanEqualityTester.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/LogicalPlanEqualityTester.java
@@ -48,7 +48,7 @@ public class LogicalPlanEqualityTester implements LogicalPlanRewriteRule {
public LogicalPlan rewrite(OverridableConf queryContext, LogicalPlan plan) throws PlanningException {
LogicalNode root = plan.getRootBlock().getRoot();
PlanProto.LogicalNodeTree serialized = LogicalNodeSerializer.serialize(plan.getRootBlock().getRoot());
- LogicalNode deserialized = LogicalNodeDeserializer.deserialize(queryContext, serialized);
+ LogicalNode deserialized = LogicalNodeDeserializer.deserialize(queryContext, null, serialized);
assert root.deepEquals(deserialized);
return plan;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java
index c49d51b..d8b6380 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java
@@ -86,7 +86,7 @@ public class PartitionedTableRewriter implements LogicalPlanRewriteRule {
public PartitionPathFilter(Schema schema, EvalNode partitionFilter) {
this.schema = schema;
this.partitionFilter = partitionFilter;
- partitionFilter.bind(schema);
+ partitionFilter.bind(null, schema);
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java
index 5a96054..3ca76ee 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeDeserializer.java
@@ -35,10 +35,11 @@ import org.apache.tajo.datum.*;
import org.apache.tajo.exception.InternalException;
import org.apache.tajo.plan.expr.*;
import org.apache.tajo.plan.function.AggFunction;
-import org.apache.tajo.plan.function.GeneralFunction;
+import org.apache.tajo.plan.function.python.PythonScriptEngine;
import org.apache.tajo.plan.logical.WindowSpec;
import org.apache.tajo.plan.serder.PlanProto.WinFunctionEvalSpec;
+import java.io.IOException;
import java.util.*;
/**
@@ -52,7 +53,7 @@ import java.util.*;
*/
public class EvalNodeDeserializer {
- public static EvalNode deserialize(OverridableConf context, PlanProto.EvalNodeTree tree) {
+ public static EvalNode deserialize(OverridableConf context, EvalContext evalContext, PlanProto.EvalNodeTree tree) {
Map<Integer, EvalNode> evalNodeMap = Maps.newHashMap();
// sort serialized eval nodes in an ascending order of their IDs.
@@ -180,9 +181,10 @@ public class EvalNodeDeserializer {
try {
funcDesc = new FunctionDesc(funcProto.getFuncion());
if (type == EvalType.FUNCTION) {
- GeneralFunction instance = (GeneralFunction) funcDesc.newInstance();
- current = new GeneralFunctionEval(context, new FunctionDesc(funcProto.getFuncion()), instance, params);
-
+ current = new GeneralFunctionEval(context, funcDesc, params);
+ if (evalContext != null && funcDesc.getInvocation().hasPython()) {
+ evalContext.addScriptEngine(current, new PythonScriptEngine(funcDesc));
+ }
} else if (type == EvalType.AGG_FUNCTION || type == EvalType.WINDOW_FUNCTION) {
AggFunction instance = (AggFunction) funcDesc.newInstance();
if (type == EvalType.AGG_FUNCTION) {
@@ -228,6 +230,8 @@ public class EvalNodeDeserializer {
throw new NoSuchFunctionException(functionName, parameterTypes);
} catch (InternalException ie) {
throw new NoSuchFunctionException(funcDesc.getFunctionName(), funcDesc.getParamTypes());
+ } catch (IOException e) {
+ throw new NoSuchFunctionException(e.getMessage());
}
} else {
throw new RuntimeException("Unknown EvalType: " + type.name());
@@ -309,6 +313,8 @@ public class EvalNodeDeserializer {
return new IntervalDatum(datum.getInterval().getMonth(), datum.getInterval().getMsec());
case NULL_TYPE:
return NullDatum.get();
+ case ANY:
+ return DatumFactory.createAny(deserialize(datum.getActual()));
default:
throw new RuntimeException("Unknown data type: " + datum.getType().name());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java
index c7702c5..e47d620 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeSerializer.java
@@ -24,6 +24,7 @@ import com.google.protobuf.ByteString;
import org.apache.tajo.algebra.WindowSpec.WindowFrameEndBoundType;
import org.apache.tajo.algebra.WindowSpec.WindowFrameStartBoundType;
import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.datum.AnyDatum;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.datum.IntervalDatum;
import org.apache.tajo.plan.expr.*;
@@ -181,6 +182,7 @@ public class EvalNodeSerializer
return rowConst;
}
+ @Override
public EvalNode visitField(EvalTreeProtoBuilderContext context, Stack<EvalNode> stack, FieldEval field) {
PlanProto.EvalNode.Builder builder = createEvalBuilder(context, field);
builder.setField(field.getColumnRef().getProto());
@@ -188,6 +190,7 @@ public class EvalNodeSerializer
return field;
}
+ @Override
public EvalNode visitBetween(EvalTreeProtoBuilderContext context, BetweenPredicateEval between,
Stack<EvalNode> stack) {
// visiting and registering childs
@@ -211,6 +214,7 @@ public class EvalNodeSerializer
return between;
}
+ @Override
public EvalNode visitCaseWhen(EvalTreeProtoBuilderContext context, CaseWhenEval caseWhen, Stack<EvalNode> stack) {
// visiting and registering childs
super.visitCaseWhen(context, caseWhen, stack);
@@ -235,6 +239,7 @@ public class EvalNodeSerializer
return caseWhen;
}
+ @Override
public EvalNode visitIfThen(EvalTreeProtoBuilderContext context, CaseWhenEval.IfThenEval ifCond,
Stack<EvalNode> stack) {
// visiting and registering childs
@@ -254,6 +259,7 @@ public class EvalNodeSerializer
return ifCond;
}
+ @Override
public EvalNode visitFuncCall(EvalTreeProtoBuilderContext context, FunctionEval function, Stack<EvalNode> stack) {
// visiting and registering childs
super.visitFuncCall(context, function, stack);
@@ -297,7 +303,6 @@ public class EvalNodeSerializer
builder.setWinFunction(windowFuncBuilder);
}
-
context.treeBuilder.addNodes(builder);
return function;
}
@@ -388,6 +393,9 @@ public class EvalNodeSerializer
intervalBuilder.setMsec(interval.getMilliSeconds());
builder.setInterval(intervalBuilder);
break;
+ case ANY:
+ builder.setActual(serialize(((AnyDatum)datum).getActual()));
+ break;
default:
throw new RuntimeException("Unknown data type: " + datum.type().name());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/serder/LogicalNodeDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/LogicalNodeDeserializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/LogicalNodeDeserializer.java
index f96626b..84991bb 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/LogicalNodeDeserializer.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/LogicalNodeDeserializer.java
@@ -23,6 +23,7 @@ import com.google.common.collect.Maps;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.OverridableConf;
import org.apache.tajo.algebra.JoinType;
+import org.apache.tajo.annotation.Nullable;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.SortSpec;
@@ -31,10 +32,7 @@ import org.apache.tajo.catalog.partition.PartitionMethodDesc;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.exception.UnimplementedException;
import org.apache.tajo.plan.Target;
-import org.apache.tajo.plan.expr.AggregationFunctionCallEval;
-import org.apache.tajo.plan.expr.EvalNode;
-import org.apache.tajo.plan.expr.FieldEval;
-import org.apache.tajo.plan.expr.WindowFunctionEval;
+import org.apache.tajo.plan.expr.*;
import org.apache.tajo.plan.logical.*;
import org.apache.tajo.util.KeyValueSet;
import org.apache.tajo.util.TUtil;
@@ -58,7 +56,8 @@ public class LogicalNodeDeserializer {
* @param tree LogicalNodeTree which contains a list of serialized logical nodes.
* @return A logical node tree
*/
- public static LogicalNode deserialize(OverridableConf context, PlanProto.LogicalNodeTree tree) {
+ public static LogicalNode deserialize(OverridableConf context, @Nullable EvalContext evalContext,
+ PlanProto.LogicalNodeTree tree) {
Map<Integer, LogicalNode> nodeMap = Maps.newHashMap();
// sort serialized logical nodes in an ascending order of their sids
@@ -88,10 +87,10 @@ public class LogicalNodeDeserializer {
current = convertSetSession(protoNode);
break;
case EXPRS:
- current = convertEvalExpr(context, protoNode);
+ current = convertEvalExpr(context, evalContext, protoNode);
break;
case PROJECTION:
- current = convertProjection(context, nodeMap, protoNode);
+ current = convertProjection(context, evalContext, nodeMap, protoNode);
break;
case LIMIT:
current = convertLimit(nodeMap, protoNode);
@@ -100,34 +99,34 @@ public class LogicalNodeDeserializer {
current = convertSort(nodeMap, protoNode);
break;
case WINDOW_AGG:
- current = convertWindowAgg(context, nodeMap, protoNode);
+ current = convertWindowAgg(context, evalContext, nodeMap, protoNode);
break;
case HAVING:
- current = convertHaving(context, nodeMap, protoNode);
+ current = convertHaving(context, evalContext, nodeMap, protoNode);
break;
case GROUP_BY:
- current = convertGroupby(context, nodeMap, protoNode);
+ current = convertGroupby(context, evalContext, nodeMap, protoNode);
break;
case DISTINCT_GROUP_BY:
- current = convertDistinctGroupby(context, nodeMap, protoNode);
+ current = convertDistinctGroupby(context, evalContext, nodeMap, protoNode);
break;
case SELECTION:
- current = convertFilter(context, nodeMap, protoNode);
+ current = convertFilter(context, evalContext, nodeMap, protoNode);
break;
case JOIN:
- current = convertJoin(context, nodeMap, protoNode);
+ current = convertJoin(context, evalContext, nodeMap, protoNode);
break;
case TABLE_SUBQUERY:
- current = convertTableSubQuery(context, nodeMap, protoNode);
+ current = convertTableSubQuery(context, evalContext, nodeMap, protoNode);
break;
case UNION:
current = convertUnion(nodeMap, protoNode);
break;
case PARTITIONS_SCAN:
- current = convertPartitionScan(context, protoNode);
+ current = convertPartitionScan(context, evalContext, protoNode);
break;
case SCAN:
- current = convertScan(context, protoNode);
+ current = convertScan(context, evalContext, protoNode);
break;
case CREATE_TABLE:
@@ -192,22 +191,25 @@ public class LogicalNodeDeserializer {
return setSession;
}
- private static EvalExprNode convertEvalExpr(OverridableConf context, PlanProto.LogicalNode protoNode) {
+ private static EvalExprNode convertEvalExpr(OverridableConf context, EvalContext evalContext,
+ PlanProto.LogicalNode protoNode) {
PlanProto.EvalExprNode evalExprProto = protoNode.getExprEval();
EvalExprNode evalExpr = new EvalExprNode(protoNode.getNodeId());
evalExpr.setInSchema(convertSchema(protoNode.getInSchema()));
- evalExpr.setTargets(convertTargets(context, evalExprProto.getTargetsList()));
+ evalExpr.setTargets(convertTargets(context, evalContext, evalExprProto.getTargetsList()));
return evalExpr;
}
- private static ProjectionNode convertProjection(OverridableConf context, Map<Integer, LogicalNode> nodeMap,
- PlanProto.LogicalNode protoNode) {
+ private static ProjectionNode convertProjection(OverridableConf context, EvalContext evalContext,
+ Map<Integer, LogicalNode> nodeMap,
+ PlanProto.LogicalNode protoNode) {
PlanProto.ProjectionNode projectionProto = protoNode.getProjection();
ProjectionNode projectionNode = new ProjectionNode(protoNode.getNodeId());
- projectionNode.init(projectionProto.getDistinct(), convertTargets(context, projectionProto.getTargetsList()));
+ projectionNode.init(projectionProto.getDistinct(), convertTargets(context, evalContext,
+ projectionProto.getTargetsList()));
projectionNode.setChild(nodeMap.get(projectionProto.getChildSeq()));
projectionNode.setInSchema(convertSchema(protoNode.getInSchema()));
projectionNode.setOutSchema(convertSchema(protoNode.getOutSchema()));
@@ -239,21 +241,22 @@ public class LogicalNodeDeserializer {
return sortNode;
}
- private static HavingNode convertHaving(OverridableConf context, Map<Integer, LogicalNode> nodeMap,
- PlanProto.LogicalNode protoNode) {
+ private static HavingNode convertHaving(OverridableConf context, EvalContext evalContext,
+ Map<Integer, LogicalNode> nodeMap, PlanProto.LogicalNode protoNode) {
PlanProto.FilterNode havingProto = protoNode.getFilter();
HavingNode having = new HavingNode(protoNode.getNodeId());
having.setChild(nodeMap.get(havingProto.getChildSeq()));
- having.setQual(EvalNodeDeserializer.deserialize(context, havingProto.getQual()));
+ having.setQual(EvalNodeDeserializer.deserialize(context, evalContext, havingProto.getQual()));
having.setInSchema(convertSchema(protoNode.getInSchema()));
having.setOutSchema(convertSchema(protoNode.getOutSchema()));
return having;
}
- private static WindowAggNode convertWindowAgg(OverridableConf context, Map<Integer, LogicalNode> nodeMap,
- PlanProto.LogicalNode protoNode) {
+ private static WindowAggNode convertWindowAgg(OverridableConf context, EvalContext evalContext,
+ Map<Integer, LogicalNode> nodeMap,
+ PlanProto.LogicalNode protoNode) {
PlanProto.WindowAggNode windowAggProto = protoNode.getWindowAgg();
WindowAggNode windowAgg = new WindowAggNode(protoNode.getNodeId());
@@ -264,7 +267,8 @@ public class LogicalNodeDeserializer {
}
if (windowAggProto.getWindowFunctionsCount() > 0) {
- windowAgg.setWindowFunctions(convertWindowFunccEvals(context, windowAggProto.getWindowFunctionsList()));
+ windowAgg.setWindowFunctions(convertWindowFunccEvals(context, evalContext,
+ windowAggProto.getWindowFunctionsList()));
}
windowAgg.setDistinct(windowAggProto.getDistinct());
@@ -274,7 +278,7 @@ public class LogicalNodeDeserializer {
}
if (windowAggProto.getTargetsCount() > 0) {
- windowAgg.setTargets(convertTargets(context, windowAggProto.getTargetsList()));
+ windowAgg.setTargets(convertTargets(context, evalContext, windowAggProto.getTargetsList()));
}
windowAgg.setInSchema(convertSchema(protoNode.getInSchema()));
@@ -283,8 +287,8 @@ public class LogicalNodeDeserializer {
return windowAgg;
}
- private static GroupbyNode convertGroupby(OverridableConf context, Map<Integer, LogicalNode> nodeMap,
- PlanProto.LogicalNode protoNode) {
+ private static GroupbyNode convertGroupby(OverridableConf context, EvalContext evalContext,
+ Map<Integer, LogicalNode> nodeMap, PlanProto.LogicalNode protoNode) {
PlanProto.GroupbyNode groupbyProto = protoNode.getGroupby();
GroupbyNode groupby = new GroupbyNode(protoNode.getNodeId());
@@ -295,10 +299,10 @@ public class LogicalNodeDeserializer {
groupby.setGroupingColumns(convertColumns(groupbyProto.getGroupingKeysList()));
}
if (groupbyProto.getAggFunctionsCount() > 0) {
- groupby.setAggFunctions(convertAggFuncCallEvals(context, groupbyProto.getAggFunctionsList()));
+ groupby.setAggFunctions(convertAggFuncCallEvals(context, evalContext, groupbyProto.getAggFunctionsList()));
}
if (groupbyProto.getTargetsCount() > 0) {
- groupby.setTargets(convertTargets(context, groupbyProto.getTargetsList()));
+ groupby.setTargets(convertTargets(context, evalContext, groupbyProto.getTargetsList()));
}
groupby.setInSchema(convertSchema(protoNode.getInSchema()));
@@ -307,21 +311,23 @@ public class LogicalNodeDeserializer {
return groupby;
}
- private static DistinctGroupbyNode convertDistinctGroupby(OverridableConf context, Map<Integer, LogicalNode> nodeMap,
- PlanProto.LogicalNode protoNode) {
+ private static DistinctGroupbyNode convertDistinctGroupby(OverridableConf context, EvalContext evalContext,
+ Map<Integer, LogicalNode> nodeMap,
+ PlanProto.LogicalNode protoNode) {
PlanProto.DistinctGroupbyNode distinctGroupbyProto = protoNode.getDistinctGroupby();
DistinctGroupbyNode distinctGroupby = new DistinctGroupbyNode(protoNode.getNodeId());
distinctGroupby.setChild(nodeMap.get(distinctGroupbyProto.getChildSeq()));
if (distinctGroupbyProto.hasGroupbyNode()) {
- distinctGroupby.setGroupbyPlan(convertGroupby(context, nodeMap, distinctGroupbyProto.getGroupbyNode()));
+ distinctGroupby.setGroupbyPlan(convertGroupby(context, evalContext, nodeMap,
+ distinctGroupbyProto.getGroupbyNode()));
}
if (distinctGroupbyProto.getSubPlansCount() > 0) {
List<GroupbyNode> subPlans = TUtil.newList();
for (int i = 0; i < distinctGroupbyProto.getSubPlansCount(); i++) {
- subPlans.add(convertGroupby(context, nodeMap, distinctGroupbyProto.getSubPlans(i)));
+ subPlans.add(convertGroupby(context, evalContext, nodeMap, distinctGroupbyProto.getSubPlans(i)));
}
distinctGroupby.setSubPlans(subPlans);
}
@@ -330,10 +336,11 @@ public class LogicalNodeDeserializer {
distinctGroupby.setGroupingColumns(convertColumns(distinctGroupbyProto.getGroupingKeysList()));
}
if (distinctGroupbyProto.getAggFunctionsCount() > 0) {
- distinctGroupby.setAggFunctions(convertAggFuncCallEvals(context, distinctGroupbyProto.getAggFunctionsList()));
+ distinctGroupby.setAggFunctions(convertAggFuncCallEvals(context, evalContext,
+ distinctGroupbyProto.getAggFunctionsList()));
}
if (distinctGroupbyProto.getTargetsCount() > 0) {
- distinctGroupby.setTargets(convertTargets(context, distinctGroupbyProto.getTargetsList()));
+ distinctGroupby.setTargets(convertTargets(context, evalContext, distinctGroupbyProto.getTargetsList()));
}
int [] resultColumnIds = new int[distinctGroupbyProto.getResultIdCount()];
for (int i = 0; i < distinctGroupbyProto.getResultIdCount(); i++) {
@@ -348,8 +355,8 @@ public class LogicalNodeDeserializer {
return distinctGroupby;
}
- private static JoinNode convertJoin(OverridableConf context, Map<Integer, LogicalNode> nodeMap,
- PlanProto.LogicalNode protoNode) {
+ private static JoinNode convertJoin(OverridableConf context, EvalContext evalContext,
+ Map<Integer, LogicalNode> nodeMap, PlanProto.LogicalNode protoNode) {
PlanProto.JoinNode joinProto = protoNode.getJoin();
JoinNode join = new JoinNode(protoNode.getNodeId());
@@ -359,24 +366,24 @@ public class LogicalNodeDeserializer {
join.setInSchema(convertSchema(protoNode.getInSchema()));
join.setOutSchema(convertSchema(protoNode.getOutSchema()));
if (joinProto.hasJoinQual()) {
- join.setJoinQual(EvalNodeDeserializer.deserialize(context, joinProto.getJoinQual()));
+ join.setJoinQual(EvalNodeDeserializer.deserialize(context, evalContext, joinProto.getJoinQual()));
}
if (joinProto.getExistsTargets()) {
- join.setTargets(convertTargets(context, joinProto.getTargetsList()));
+ join.setTargets(convertTargets(context, evalContext, joinProto.getTargetsList()));
}
return join;
}
- private static SelectionNode convertFilter(OverridableConf context, Map<Integer, LogicalNode> nodeMap,
- PlanProto.LogicalNode protoNode) {
+ private static SelectionNode convertFilter(OverridableConf context, EvalContext evalContext,
+ Map<Integer, LogicalNode> nodeMap, PlanProto.LogicalNode protoNode) {
PlanProto.FilterNode filterProto = protoNode.getFilter();
SelectionNode selection = new SelectionNode(protoNode.getNodeId());
selection.setInSchema(convertSchema(protoNode.getInSchema()));
selection.setOutSchema(convertSchema(protoNode.getOutSchema()));
selection.setChild(nodeMap.get(filterProto.getChildSeq()));
- selection.setQual(EvalNodeDeserializer.deserialize(context, filterProto.getQual()));
+ selection.setQual(EvalNodeDeserializer.deserialize(context, evalContext, filterProto.getQual()));
return selection;
}
@@ -393,14 +400,15 @@ public class LogicalNodeDeserializer {
return union;
}
- private static ScanNode convertScan(OverridableConf context, PlanProto.LogicalNode protoNode) {
+ private static ScanNode convertScan(OverridableConf context, EvalContext evalContext, PlanProto.LogicalNode protoNode) {
ScanNode scan = new ScanNode(protoNode.getNodeId());
- fillScanNode(context, protoNode, scan);
+ fillScanNode(context, evalContext, protoNode, scan);
return scan;
}
- private static void fillScanNode(OverridableConf context, PlanProto.LogicalNode protoNode, ScanNode scan) {
+ private static void fillScanNode(OverridableConf context, EvalContext evalContext, PlanProto.LogicalNode protoNode,
+ ScanNode scan) {
PlanProto.ScanNode scanProto = protoNode.getScan();
if (scanProto.hasAlias()) {
scan.init(new TableDesc(scanProto.getTable()), scanProto.getAlias());
@@ -409,11 +417,11 @@ public class LogicalNodeDeserializer {
}
if (scanProto.getExistTargets()) {
- scan.setTargets(convertTargets(context, scanProto.getTargetsList()));
+ scan.setTargets(convertTargets(context, evalContext, scanProto.getTargetsList()));
}
if (scanProto.hasQual()) {
- scan.setQual(EvalNodeDeserializer.deserialize(context, scanProto.getQual()));
+ scan.setQual(EvalNodeDeserializer.deserialize(context, evalContext, scanProto.getQual()));
}
if(scanProto.hasBroadcast()){
@@ -423,9 +431,10 @@ public class LogicalNodeDeserializer {
scan.setOutSchema(convertSchema(protoNode.getOutSchema()));
}
- private static PartitionedTableScanNode convertPartitionScan(OverridableConf context, PlanProto.LogicalNode protoNode) {
+ private static PartitionedTableScanNode convertPartitionScan(OverridableConf context, EvalContext evalContext,
+ PlanProto.LogicalNode protoNode) {
PartitionedTableScanNode partitionedScan = new PartitionedTableScanNode(protoNode.getNodeId());
- fillScanNode(context, protoNode, partitionedScan);
+ fillScanNode(context, evalContext, protoNode, partitionedScan);
PlanProto.PartitionScanSpec partitionScanProto = protoNode.getPartitionScan();
Path [] paths = new Path[partitionScanProto.getPathsCount()];
@@ -436,16 +445,16 @@ public class LogicalNodeDeserializer {
return partitionedScan;
}
- private static TableSubQueryNode convertTableSubQuery(OverridableConf context,
- Map<Integer, LogicalNode> nodeMap,
- PlanProto.LogicalNode protoNode) {
+ private static TableSubQueryNode convertTableSubQuery(OverridableConf context, EvalContext evalContext,
+ Map<Integer, LogicalNode> nodeMap,
+ PlanProto.LogicalNode protoNode) {
PlanProto.TableSubQueryNode proto = protoNode.getTableSubQuery();
TableSubQueryNode tableSubQuery = new TableSubQueryNode(protoNode.getNodeId());
tableSubQuery.init(proto.getTableName(), nodeMap.get(proto.getChildSeq()));
tableSubQuery.setInSchema(convertSchema(protoNode.getInSchema()));
if (proto.getTargetsCount() > 0) {
- tableSubQuery.setTargets(convertTargets(context, proto.getTargetsList()));
+ tableSubQuery.setTargets(convertTargets(context, evalContext, proto.getTargetsList()));
}
return tableSubQuery;
@@ -602,20 +611,21 @@ public class LogicalNodeDeserializer {
return truncateTable;
}
- private static AggregationFunctionCallEval [] convertAggFuncCallEvals(OverridableConf context,
+ private static AggregationFunctionCallEval [] convertAggFuncCallEvals(OverridableConf context, EvalContext evalContext,
List<PlanProto.EvalNodeTree> evalTrees) {
AggregationFunctionCallEval [] aggFuncs = new AggregationFunctionCallEval[evalTrees.size()];
for (int i = 0; i < aggFuncs.length; i++) {
- aggFuncs[i] = (AggregationFunctionCallEval) EvalNodeDeserializer.deserialize(context, evalTrees.get(i));
+ aggFuncs[i] = (AggregationFunctionCallEval) EvalNodeDeserializer.deserialize(context, evalContext,
+ evalTrees.get(i));
}
return aggFuncs;
}
- private static WindowFunctionEval[] convertWindowFunccEvals(OverridableConf context,
- List<PlanProto.EvalNodeTree> evalTrees) {
+ private static WindowFunctionEval[] convertWindowFunccEvals(OverridableConf context, EvalContext evalContext,
+ List<PlanProto.EvalNodeTree> evalTrees) {
WindowFunctionEval[] winFuncEvals = new WindowFunctionEval[evalTrees.size()];
for (int i = 0; i < winFuncEvals.length; i++) {
- winFuncEvals[i] = (WindowFunctionEval) EvalNodeDeserializer.deserialize(context, evalTrees.get(i));
+ winFuncEvals[i] = (WindowFunctionEval) EvalNodeDeserializer.deserialize(context, evalContext, evalTrees.get(i));
}
return winFuncEvals;
}
@@ -632,11 +642,12 @@ public class LogicalNodeDeserializer {
return columns;
}
- public static Target[] convertTargets(OverridableConf context, List<PlanProto.Target> targetsProto) {
+ public static Target[] convertTargets(OverridableConf context, EvalContext evalContext,
+ List<PlanProto.Target> targetsProto) {
Target [] targets = new Target[targetsProto.size()];
for (int i = 0; i < targets.length; i++) {
PlanProto.Target targetProto = targetsProto.get(i);
- EvalNode evalNode = EvalNodeDeserializer.deserialize(context, targetProto.getExpr());
+ EvalNode evalNode = EvalNodeDeserializer.deserialize(context, evalContext, targetProto.getExpr());
if (targetProto.hasAlias()) {
targets[i] = new Target(evalNode, targetProto.getAlias());
} else {
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/proto/Plan.proto
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/proto/Plan.proto b/tajo-plan/src/main/proto/Plan.proto
index c555a1a..de949d5 100644
--- a/tajo-plan/src/main/proto/Plan.proto
+++ b/tajo-plan/src/main/proto/Plan.proto
@@ -330,25 +330,26 @@ enum EvalType {
WINDOW_FUNCTION = 18;
AGG_FUNCTION = 19;
FUNCTION = 20;
+ PYTHON_FUNCTION = 21;
// String operator or pattern matching predicates
- LIKE = 21;
- SIMILAR_TO = 22;
- REGEX = 23;
- CONCATENATE = 24;
+ LIKE = 22;
+ SIMILAR_TO = 23;
+ REGEX = 24;
+ CONCATENATE = 25;
// Other predicates
- BETWEEN = 25;
- CASE = 26;
- IF_THEN = 27;
- IN = 28;
+ BETWEEN = 26;
+ CASE = 27;
+ IF_THEN = 28;
+ IN = 29;
// Value or Reference
- SIGNED = 29;
- CAST = 30;
- ROW_CONSTANT = 31;
- FIELD = 32;
- CONST = 33;
+ SIGNED = 30;
+ CAST = 31;
+ ROW_CONSTANT = 32;
+ FIELD = 33;
+ CONST = 34;
}
message EvalNodeTree {
@@ -476,6 +477,7 @@ message Datum {
optional string text = 7;
optional bytes blob = 8;
optional Interval interval = 12;
+ optional Datum actual = 13; // for ANY type datum
}
message Interval {
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-storage/tajo-storage-common/src/test/java/org/apache/tajo/storage/TestLazyTuple.java
----------------------------------------------------------------------
diff --git a/tajo-storage/tajo-storage-common/src/test/java/org/apache/tajo/storage/TestLazyTuple.java b/tajo-storage/tajo-storage-common/src/test/java/org/apache/tajo/storage/TestLazyTuple.java
index fccaf2a..9e7f334 100644
--- a/tajo-storage/tajo-storage-common/src/test/java/org/apache/tajo/storage/TestLazyTuple.java
+++ b/tajo-storage/tajo-storage-common/src/test/java/org/apache/tajo/storage/TestLazyTuple.java
@@ -65,7 +65,7 @@ public class TestLazyTuple {
sb.append(DatumFactory.createFloat4(77.9f)).append('|');
sb.append(DatumFactory.createFloat8(271.9f)).append('|');
sb.append(DatumFactory.createText("str2")).append('|');
- sb.append(DatumFactory.createBlob("jinho".getBytes())).append('|');
+ sb.append(DatumFactory.createBlob("jinho")).append('|');
sb.append(DatumFactory.createInet4("192.168.0.1")).append('|');
sb.append(new String(nullbytes)).append('|');
sb.append(NullDatum.get());
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-storage/tajo-storage-hdfs/src/test/resources/dataset/TestDelimitedTextFile/testErrorTolerance1.json
----------------------------------------------------------------------
diff --git a/tajo-storage/tajo-storage-hdfs/src/test/resources/dataset/TestDelimitedTextFile/testErrorTolerance1.json b/tajo-storage/tajo-storage-hdfs/src/test/resources/dataset/TestDelimitedTextFile/testErrorTolerance1.json
index 739dfe7..d3aee33 100644
--- a/tajo-storage/tajo-storage-hdfs/src/test/resources/dataset/TestDelimitedTextFile/testErrorTolerance1.json
+++ b/tajo-storage/tajo-storage-hdfs/src/test/resources/dataset/TestDelimitedTextFile/testErrorTolerance1.json
@@ -1,6 +1,6 @@
-{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "hyunsik", "col10": "192.168.0.1"}
-{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "hyunsik", "col10": "192.168.0.1"
-{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "hyunsik", "col10": "192.168.0.1"}
-{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "hyunsik", "col10": "192.168.0.1"
-{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "hyunsik", "col10": "192.168.0.1"
-{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "hyunsik", "col10": "192.168.0.1"}
\ No newline at end of file
+{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "aHl1bnNpaw==", "col10": "192.168.0.1"}
+{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "aHl1bnNpaw==", "col10": "192.168.0.1"
+{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "aHl1bnNpaw==", "col10": "192.168.0.1"}
+{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "aHl1bnNpaw==", "col10": "192.168.0.1"
+{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "aHl1bnNpaw==", "col10": "192.168.0.1"
+{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "aHl1bnNpaw==", "col10": "192.168.0.1"}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-storage/tajo-storage-hdfs/src/test/resources/dataset/TestJsonSerDe/testVariousType.json
----------------------------------------------------------------------
diff --git a/tajo-storage/tajo-storage-hdfs/src/test/resources/dataset/TestJsonSerDe/testVariousType.json b/tajo-storage/tajo-storage-hdfs/src/test/resources/dataset/TestJsonSerDe/testVariousType.json
index 8ee3408..ec31982 100644
--- a/tajo-storage/tajo-storage-hdfs/src/test/resources/dataset/TestJsonSerDe/testVariousType.json
+++ b/tajo-storage/tajo-storage-hdfs/src/test/resources/dataset/TestJsonSerDe/testVariousType.json
@@ -1 +1 @@
-{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "hyunsik", "col10": "192.168.0.1"}
+{"col1": "true", "col2": "hyunsik", "col3": 17, "col4": 59, "col5": 23, "col6": 77.9, "col7": 271.9, "col8": "hyunsik", "col9": "aHl1bnNpaw==", "col10": "192.168.0.1"}
[2/4] tajo git commit: TAJO-1344: Python UDF support. (jihoon)
Posted by ji...@apache.org.
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/FunctionInvoke.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/FunctionInvoke.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/FunctionInvoke.java
new file mode 100644
index 0000000..728ae10
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/FunctionInvoke.java
@@ -0,0 +1,90 @@
+/**
+ * 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.tajo.plan.function;
+
+import com.google.gson.annotations.Expose;
+import org.apache.tajo.catalog.FunctionDesc;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.exception.InternalException;
+import org.apache.tajo.exception.UnsupportedException;
+import org.apache.tajo.plan.expr.EvalContext;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * An abstract class for actual function invocation.
+ * The metadata for function invocation are stored in the {@link org.apache.tajo.function.FunctionInvocation} class.
+ */
+public abstract class FunctionInvoke implements Cloneable {
+ @Expose protected FunctionDesc functionDesc;
+
+ public FunctionInvoke() {
+
+ }
+
+ public FunctionInvoke(FunctionDesc functionDesc) {
+ this.functionDesc = functionDesc;
+ }
+
+ public static FunctionInvoke newInstance(FunctionDesc desc) throws InternalException {
+ if (desc.getInvocation().hasLegacy()) {
+ return new LegacyScalarFunctionInvoke(desc);
+ } else if (desc.getInvocation().hasPython()) {
+ return new PythonFunctionInvoke(desc);
+ } else {
+ throw new UnsupportedException(desc.getInvocation() + " is not supported");
+ }
+ }
+
+ public void setFunctionDesc(FunctionDesc functionDesc) throws InternalException {
+ this.functionDesc = functionDesc;
+ }
+
+ public abstract void init(FunctionInvokeContext context) throws IOException;
+
+ /**
+ * Evaluate the given tuple with a function
+ * @param tuple a tuple evaluated with parameters
+ * @return a result of a fuction execution
+ */
+ public abstract Datum eval(Tuple tuple);
+
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof FunctionInvoke) {
+ FunctionInvoke other = (FunctionInvoke) o;
+ return this.functionDesc.equals(other.functionDesc);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return functionDesc.hashCode();
+ }
+
+ @Override
+ public Object clone() throws CloneNotSupportedException {
+ FunctionInvoke clone = (FunctionInvoke) super.clone();
+ clone.functionDesc = (FunctionDesc) this.functionDesc.clone();
+ return clone;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/FunctionInvokeContext.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/FunctionInvokeContext.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/FunctionInvokeContext.java
new file mode 100644
index 0000000..b938072
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/FunctionInvokeContext.java
@@ -0,0 +1,74 @@
+/*
+ * 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.tajo.plan.function;
+
+import com.google.common.base.Objects;
+import org.apache.tajo.OverridableConf;
+import org.apache.tajo.plan.expr.FunctionEval;
+import org.apache.tajo.plan.function.python.TajoScriptEngine;
+
+import java.util.Arrays;
+
+/**
+ * This class contains some metadata need to execute functions.
+ */
+public class FunctionInvokeContext {
+ private final OverridableConf queryContext;
+ private final FunctionEval.ParamType[] paramTypes;
+ private TajoScriptEngine scriptEngine;
+
+ public FunctionInvokeContext(OverridableConf queryContext, FunctionEval.ParamType[] paramTypes) {
+ this.queryContext = queryContext;
+ this.paramTypes = paramTypes;
+ }
+
+ public OverridableConf getQueryContext() {
+ return queryContext;
+ }
+
+ public FunctionEval.ParamType[] getParamTypes() {
+ return paramTypes;
+ }
+
+ public void setScriptEngine(TajoScriptEngine scriptEngine) {
+ this.scriptEngine = scriptEngine;
+ }
+
+ public boolean hasScriptExecutor() {
+ return scriptEngine != null;
+ }
+
+ public TajoScriptEngine getScriptEngine() {
+ return scriptEngine;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(queryContext, Arrays.hashCode(paramTypes));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof FunctionInvokeContext) {
+ FunctionInvokeContext other = (FunctionInvokeContext) o;
+ return queryContext.equals(other.queryContext) && Arrays.equals(paramTypes, other.paramTypes);
+ }
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/LegacyScalarFunctionInvoke.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/LegacyScalarFunctionInvoke.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/LegacyScalarFunctionInvoke.java
new file mode 100644
index 0000000..6b2c116
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/LegacyScalarFunctionInvoke.java
@@ -0,0 +1,81 @@
+/**
+ * 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.tajo.plan.function;
+
+import com.google.gson.annotations.Expose;
+import org.apache.tajo.catalog.FunctionDesc;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.exception.InternalException;
+import org.apache.tajo.plan.expr.EvalContext;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.util.TUtil;
+
+/**
+ * This class invokes the legacy scala functions.
+ */
+public class LegacyScalarFunctionInvoke extends FunctionInvoke implements Cloneable {
+ @Expose private GeneralFunction function;
+
+ public LegacyScalarFunctionInvoke() {
+
+ }
+
+ public LegacyScalarFunctionInvoke(FunctionDesc funcDesc) throws InternalException {
+ super(funcDesc);
+ function = (GeneralFunction) funcDesc.newInstance();
+ }
+
+ @Override
+ public void setFunctionDesc(FunctionDesc desc) throws InternalException {
+ super.setFunctionDesc(desc);
+ function = (GeneralFunction) functionDesc.newInstance();
+ }
+
+ @Override
+ public void init(FunctionInvokeContext context) {
+ function.init(context.getQueryContext(), context.getParamTypes());
+ }
+
+ @Override
+ public Datum eval(Tuple tuple) {
+ return function.eval(tuple);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof LegacyScalarFunctionInvoke) {
+ LegacyScalarFunctionInvoke other = (LegacyScalarFunctionInvoke) o;
+ return super.equals(other) &&
+ TUtil.checkEquals(function, other.function);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return function.hashCode();
+ }
+
+ @Override
+ public Object clone() throws CloneNotSupportedException {
+ LegacyScalarFunctionInvoke clone = (LegacyScalarFunctionInvoke) super.clone();
+ clone.function = (GeneralFunction) function.clone();
+ return clone;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/PythonFunctionInvoke.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/PythonFunctionInvoke.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/PythonFunctionInvoke.java
new file mode 100644
index 0000000..1019c60
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/PythonFunctionInvoke.java
@@ -0,0 +1,59 @@
+/*
+ * Lisensed 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.tajo.plan.function;
+
+import org.apache.tajo.catalog.FunctionDesc;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.plan.function.python.PythonScriptEngine;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+
+/**
+ * This class invokes the python functions.
+ */
+public class PythonFunctionInvoke extends FunctionInvoke implements Cloneable {
+
+ private transient PythonScriptEngine scriptEngine;
+
+ public PythonFunctionInvoke() {
+
+ }
+
+ public PythonFunctionInvoke(FunctionDesc functionDesc) {
+ super(functionDesc);
+ }
+
+ @Override
+ public void init(FunctionInvokeContext context) throws IOException {
+ this.scriptEngine = (PythonScriptEngine) context.getScriptEngine();
+ }
+
+ @Override
+ public Datum eval(Tuple tuple) {
+ Datum res = scriptEngine.eval(tuple);
+ return res;
+ }
+
+ @Override
+ public Object clone() throws CloneNotSupportedException {
+ // nothing to do
+ return super.clone();
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/python/PythonScriptEngine.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/python/PythonScriptEngine.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/python/PythonScriptEngine.java
new file mode 100644
index 0000000..0da30f1
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/python/PythonScriptEngine.java
@@ -0,0 +1,368 @@
+/*
+ * 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.tajo.plan.function.python;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.tajo.catalog.*;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.function.FunctionInvocation;
+import org.apache.tajo.function.FunctionSignature;
+import org.apache.tajo.function.FunctionSupplement;
+import org.apache.tajo.function.PythonInvocationDesc;
+import org.apache.tajo.plan.function.stream.*;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+import org.apache.tajo.util.FileUtil;
+import org.apache.tajo.util.TUtil;
+
+import java.io.*;
+import java.net.URI;
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+/**
+ * {@link PythonScriptEngine} is responsible for registering python functions and maintaining the controller process.
+ * The controller is a python process that executes the python UDFs.
+ * (Please refer to 'tajo-core/src/main/resources/python/controller.py')
+ * Data are exchanged via standard I/O between PythonScriptEngine and the controller.
+ */
+public class PythonScriptEngine extends TajoScriptEngine {
+
+ private static final Log LOG = LogFactory.getLog(PythonScriptEngine.class);
+
+ public static final String FILE_EXTENSION = ".py";
+
+ /**
+ * Register functions defined in a python script
+ *
+ * @param path path to the python script file
+ * @param namespace namespace where the functions will be defined
+ * @return set of function descriptions
+ * @throws IOException
+ */
+ public static Set<FunctionDesc> registerFunctions(URI path, String namespace) throws IOException {
+ // TODO: we should support the namespace for python functions.
+
+ Set<FunctionDesc> functionDescs = TUtil.newHashSet();
+
+ InputStream in = getScriptAsStream(path);
+ List<FuncInfo> functions = null;
+ try {
+ functions = getFunctions(in);
+ } finally {
+ in.close();
+ }
+ for(FuncInfo funcInfo : functions) {
+ TajoDataTypes.DataType returnType = CatalogUtil.newSimpleDataType(TajoDataTypes.Type.valueOf(funcInfo.returnType));
+ FunctionSignature signature = new FunctionSignature(CatalogProtos.FunctionType.UDF, funcInfo.funcName,
+ returnType, createParamTypes(funcInfo.paramNum));
+ FunctionInvocation invocation = new FunctionInvocation();
+ PythonInvocationDesc invocationDesc = new PythonInvocationDesc(funcInfo.funcName, path.getPath());
+ invocation.setPython(invocationDesc);
+ FunctionSupplement supplement = new FunctionSupplement();
+ functionDescs.add(new FunctionDesc(signature, invocation, supplement));
+ }
+ return functionDescs;
+ }
+
+ private static TajoDataTypes.DataType[] createParamTypes(int paramNum) {
+ TajoDataTypes.DataType[] paramTypes = new TajoDataTypes.DataType[paramNum];
+ for (int i = 0; i < paramNum; i++) {
+ paramTypes[i] = TajoDataTypes.DataType.newBuilder().setType(TajoDataTypes.Type.ANY).build();
+ }
+ return paramTypes;
+ }
+
+ private static final Pattern pSchema = Pattern.compile("^\\s*\\W+outputType.*");
+ private static final Pattern pDef = Pattern.compile("^\\s*def\\s+(\\w+)\\s*.+");
+
+ private static class FuncInfo {
+ String returnType;
+ String funcName;
+ int paramNum;
+
+ public FuncInfo(String returnType, String funcName, int paramNum) {
+ this.returnType = returnType.toUpperCase();
+ this.funcName = funcName;
+ this.paramNum = paramNum;
+ }
+ }
+
+ // TODO: python parser must be improved.
+ private static List<FuncInfo> getFunctions(InputStream is) throws IOException {
+ List<FuncInfo> functions = TUtil.newList();
+ InputStreamReader in = new InputStreamReader(is, Charset.defaultCharset());
+ BufferedReader br = new BufferedReader(in);
+ String line = br.readLine();
+ String schemaString = null;
+ while (line != null) {
+ if (pSchema.matcher(line).matches()) {
+ int start = line.indexOf("(") + 2; //drop brackets/quotes
+ int end = line.lastIndexOf(")") - 1;
+ schemaString = line.substring(start,end).trim();
+ } else if (pDef.matcher(line).matches()) {
+ int nameStart = line.indexOf("def ") + "def ".length();
+ int nameEnd = line.indexOf('(');
+ int signatureEnd = line.indexOf(')');
+ String[] params = line.substring(nameEnd+1, signatureEnd).split(",");
+ int paramNum;
+ if (params.length == 1) {
+ paramNum = params[0].equals("") ? 0 : 1;
+ } else {
+ paramNum = params.length;
+ }
+
+ String functionName = line.substring(nameStart, nameEnd).trim();
+ schemaString = schemaString == null ? "blob" : schemaString;
+ functions.add(new FuncInfo(schemaString, functionName, paramNum));
+ schemaString = null;
+ }
+ line = br.readLine();
+ }
+ br.close();
+ in.close();
+ return functions;
+ }
+
+
+ private static final String PYTHON_LANGUAGE = "python";
+ private static final String PYTHON_ROOT_PATH = "/python";
+ private static final String TAJO_UTIL_NAME = "tajo_util.py";
+ private static final String CONTROLLER_NAME = "controller.py";
+ private static final String PYTHON_CONTROLLER_JAR_PATH = PYTHON_ROOT_PATH + File.separator + CONTROLLER_NAME; // Relative to root of tajo jar.
+ private static final String PYTHON_TAJO_UTIL_PATH = PYTHON_ROOT_PATH + File.separator + TAJO_UTIL_NAME; // Relative to root of tajo jar.
+ private static final String DEFAULT_LOG_DIR = "/tmp/tajo-" + System.getProperty("user.name") + "/python";
+
+ // Indexes for arguments being passed to external process
+ private static final int UDF_LANGUAGE = 0;
+ private static final int PATH_TO_CONTROLLER_FILE = 1;
+ private static final int UDF_FILE_NAME = 2; // Name of file where UDF function is defined
+ private static final int UDF_FILE_PATH = 3; // Path to directory containing file where UDF function is defined
+ private static final int UDF_NAME = 4; // Name of UDF function being called.
+ private static final int PATH_TO_FILE_CACHE = 5; // Directory where required files (like tajo_util) are cached on cluster nodes.
+ private static final int STD_OUT_OUTPUT_PATH = 6; // File for output from when user writes to standard output.
+ private static final int STD_ERR_OUTPUT_PATH = 7; // File for output from when user writes to standard error.
+ private static final int CONTROLLER_LOG_FILE_PATH = 8; // Controller log file logs progress through the controller script not user code.
+ private static final int OUT_SCHEMA = 9; // the schema of the output column
+
+ private Configuration systemConf;
+
+ private Process process; // Handle to the external execution of python functions
+
+ private InputHandler inputHandler;
+ private OutputHandler outputHandler;
+
+ private DataOutputStream stdin; // stdin of the process
+ private InputStream stdout; // stdout of the process
+ private InputStream stderr; // stderr of the process
+
+ private final FunctionSignature functionSignature;
+ private final PythonInvocationDesc invocationDesc;
+ private final Schema inSchema;
+ private final Schema outSchema;
+ private final int [] projectionCols = new int[]{0};
+
+ private final CSVLineSerDe lineSerDe = new CSVLineSerDe();
+ private final TableMeta pipeMeta = CatalogUtil.newTableMeta(CatalogProtos.StoreType.TEXTFILE);
+
+ private static final Tuple EMPTY_INPUT = new VTuple(0);
+
+ public PythonScriptEngine(FunctionDesc functionDesc) {
+ if (!functionDesc.getInvocation().hasPython()) {
+ throw new IllegalStateException("Function type must be 'python'");
+ }
+ functionSignature = functionDesc.getSignature();
+ invocationDesc = functionDesc.getInvocation().getPython();
+
+ TajoDataTypes.DataType[] paramTypes = functionSignature.getParamTypes();
+ inSchema = new Schema();
+ for (int i = 0; i < paramTypes.length; i++) {
+ inSchema.addColumn(new Column("in_" + i, paramTypes[i]));
+ }
+ outSchema = new Schema(new Column[]{new Column("out", functionSignature.getReturnType())});
+ }
+
+ @Override
+ public void start(Configuration systemConf) throws IOException {
+ this.systemConf = systemConf;
+ startUdfController();
+ createInputHandlers();
+ setStreams();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("PythonScriptExecutor starts up");
+ }
+ }
+
+ @Override
+ public void shutdown() {
+ process.destroy();
+ FileUtil.cleanup(LOG, stdin);
+ FileUtil.cleanup(LOG, stdout);
+ FileUtil.cleanup(LOG, stderr);
+ FileUtil.cleanup(LOG, inputHandler);
+ FileUtil.cleanup(LOG, outputHandler);
+ stdin = null;
+ stdout = stderr = null;
+ inputHandler = null;
+ outputHandler = null;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("PythonScriptExecutor shuts down");
+ }
+ }
+
+ private void startUdfController() throws IOException {
+ ProcessBuilder processBuilder = StreamingUtil.createProcess(buildCommand());
+ process = processBuilder.start();
+ }
+
+ /**
+ * Build a command to execute an external process.
+ * @return
+ * @throws IOException
+ */
+ private String[] buildCommand() throws IOException {
+ String[] command = new String[10];
+
+ // TODO: support controller logging
+ String standardOutputRootWriteLocation = systemConf.get(TajoConf.ConfVars.PYTHON_CONTROLLER_LOG_DIR.keyname(),
+ DEFAULT_LOG_DIR);
+ if (!standardOutputRootWriteLocation.equals(DEFAULT_LOG_DIR)) {
+ LOG.warn("Currently, logging is not supported for the python controller.");
+ }
+ String controllerLogFileName, outFileName, errOutFileName;
+
+ String funcName = invocationDesc.getName();
+ String filePath = invocationDesc.getPath();
+
+ controllerLogFileName = standardOutputRootWriteLocation + funcName + "_controller.log";
+ outFileName = standardOutputRootWriteLocation + funcName + ".out";
+ errOutFileName = standardOutputRootWriteLocation + funcName + ".err";
+
+ command[UDF_LANGUAGE] = PYTHON_LANGUAGE;
+ command[PATH_TO_CONTROLLER_FILE] = getControllerPath();
+ int lastSeparator = filePath.lastIndexOf(File.separator) + 1;
+ String fileName = filePath.substring(lastSeparator);
+ fileName = fileName.endsWith(FILE_EXTENSION) ? fileName.substring(0, fileName.length()-3) : fileName;
+ command[UDF_FILE_NAME] = fileName;
+ command[UDF_FILE_PATH] = lastSeparator <= 0 ? "." : filePath.substring(0, lastSeparator - 1);
+ command[UDF_NAME] = funcName;
+ String fileCachePath = systemConf.get(TajoConf.ConfVars.PYTHON_CODE_DIR.keyname());
+ if (fileCachePath == null) {
+ throw new IOException(TajoConf.ConfVars.PYTHON_CODE_DIR.keyname() + " must be set.");
+ }
+ command[PATH_TO_FILE_CACHE] = "'" + fileCachePath + "'";
+ command[STD_OUT_OUTPUT_PATH] = outFileName;
+ command[STD_ERR_OUTPUT_PATH] = errOutFileName;
+ command[CONTROLLER_LOG_FILE_PATH] = controllerLogFileName;
+ command[OUT_SCHEMA] = outSchema.getColumn(0).getDataType().getType().name().toLowerCase();
+
+ return command;
+ }
+
+ private void createInputHandlers() {
+ TextLineSerializer serializer = lineSerDe.createSerializer(inSchema, pipeMeta);
+ serializer.init();
+ this.inputHandler = new InputHandler(serializer);
+ TextLineDeserializer deserializer = lineSerDe.createDeserializer(outSchema, pipeMeta, projectionCols);
+ deserializer.init();
+ this.outputHandler = new OutputHandler(deserializer);
+ }
+
+ /**
+ * Get the standard input, output, and error streams of the external process
+ *
+ * @throws IOException
+ */
+ private void setStreams() throws IOException {
+ stdout = new DataInputStream(new BufferedInputStream(process.getInputStream()));
+ outputHandler.bindTo(stdout);
+
+ stdin = new DataOutputStream(new BufferedOutputStream(process.getOutputStream()));
+ inputHandler.bindTo(stdin);
+
+ stderr = new DataInputStream(new BufferedInputStream(process.getErrorStream()));
+ }
+
+ /**
+ * Find the path to the controller file for the streaming language.
+ *
+ * First check path to job jar and if the file is not found (like in the
+ * case of running hadoop in standalone mode) write the necessary files
+ * to temporary files and return that path.
+ *
+ * @return
+ * @throws IOException
+ */
+ private String getControllerPath() throws IOException {
+ String controllerPath = PYTHON_CONTROLLER_JAR_PATH;
+ File controller = new File(PYTHON_CONTROLLER_JAR_PATH);
+ if (!controller.exists()) {
+ File controllerFile = File.createTempFile("controller", FILE_EXTENSION);
+ InputStream pythonControllerStream = this.getClass().getResourceAsStream(PYTHON_CONTROLLER_JAR_PATH);
+ try {
+ FileUtils.copyInputStreamToFile(pythonControllerStream, controllerFile);
+ } finally {
+ pythonControllerStream.close();
+ }
+ controllerFile.deleteOnExit();
+ File tajoUtilFile = new File(controllerFile.getParent() + File.separator + TAJO_UTIL_NAME);
+ tajoUtilFile.deleteOnExit();
+ InputStream pythonUtilStream = this.getClass().getResourceAsStream(PYTHON_TAJO_UTIL_PATH);
+ try {
+ FileUtils.copyInputStreamToFile(pythonUtilStream, tajoUtilFile);
+ } finally {
+ pythonUtilStream.close();
+ }
+ controllerPath = controllerFile.getAbsolutePath();
+ }
+ return controllerPath;
+ }
+
+ public Datum eval(Tuple input) {
+ try {
+ if (input == null) {
+ // When nothing is passed into the UDF the tuple
+ // being sent is the full tuple for the relation.
+ // We want it to be nothing (since that's what the user wrote).
+ input = EMPTY_INPUT;
+ }
+
+ inputHandler.putNext(input);
+ stdin.flush();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed adding input to inputQueue", e);
+ }
+ Datum result;
+ try {
+ result = outputHandler.getNext().get(0);
+ } catch (Exception e) {
+ throw new RuntimeException("Problem getting output", e);
+ }
+
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/python/TajoScriptEngine.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/python/TajoScriptEngine.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/python/TajoScriptEngine.java
new file mode 100644
index 0000000..726ec2f
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/python/TajoScriptEngine.java
@@ -0,0 +1,83 @@
+/*
+ * 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.tajo.plan.function.python;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.*;
+import java.net.URI;
+
+/**
+ * Abstract class of script engine
+ */
+public abstract class TajoScriptEngine {
+
+ /**
+ * Open a stream load a script locally or in the classpath
+ * @param scriptPath the path of the script
+ * @return a stream (it is the responsibility of the caller to close it)
+ * @throws IllegalStateException if we could not open a stream
+ */
+ protected static InputStream getScriptAsStream(URI scriptPath) {
+ InputStream is = null;
+ File file;
+ try {
+ file = new File(scriptPath);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("path: " + scriptPath, e);
+ }
+ if (file.exists()) {
+ try {
+ is = new FileInputStream(file);
+ } catch (FileNotFoundException e) {
+ throw new IllegalStateException("could not find existing file "+scriptPath, e);
+ }
+ }
+
+ if (is == null) {
+ throw new IllegalStateException(
+ "Could not initialize interpreter (from file system or classpath) with " + scriptPath);
+ }
+ return is;
+ }
+
+ /**
+ * Start TajoScriptEngine.
+ *
+ * @param systemConf
+ * @throws IOException
+ */
+ public abstract void start(Configuration systemConf) throws IOException;
+
+ /**
+ * Shutdown TajoScriptEngine.
+ * @throws IOException
+ */
+ public abstract void shutdown();
+
+ /**
+ * Evaluate the input tuple.
+ *
+ * @param input
+ * @return
+ */
+ public abstract Datum eval(Tuple input);
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/BufferPool.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/BufferPool.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/BufferPool.java
new file mode 100644
index 0000000..0b3c625
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/BufferPool.java
@@ -0,0 +1,74 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.util.internal.PlatformDependent;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/* this class is PooledBuffer holder */
+public class BufferPool {
+
+ private static final PooledByteBufAllocator allocator;
+
+ private BufferPool() {
+ }
+
+ static {
+ //TODO we need determine the default params
+ allocator = new PooledByteBufAllocator(PlatformDependent.directBufferPreferred());
+
+ /* if you are finding memory leak, please enable this line */
+ //ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);
+ }
+
+ public static long maxDirectMemory() {
+ return PlatformDependent.maxDirectMemory();
+ }
+
+
+ public static ByteBuf directBuffer(int size) {
+ return allocator.directBuffer(size);
+ }
+
+ /**
+ *
+ * @param size the initial capacity
+ * @param max the max capacity
+ * @return allocated ByteBuf from pool
+ */
+ public static ByteBuf directBuffer(int size, int max) {
+ return allocator.directBuffer(size, max);
+ }
+
+ @InterfaceStability.Unstable
+ public static void forceRelease(ByteBuf buf) {
+ buf.release(buf.refCnt());
+ }
+
+ /**
+ * the ByteBuf will increase to writable size
+ * @param buf
+ * @param minWritableBytes required minimum writable size
+ */
+ public static void ensureWritable(ByteBuf buf, int minWritableBytes) {
+ buf.ensureWritable(minWritableBytes);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/ByteBufInputChannel.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/ByteBufInputChannel.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/ByteBufInputChannel.java
new file mode 100644
index 0000000..daf2357
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/ByteBufInputChannel.java
@@ -0,0 +1,71 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import org.apache.hadoop.fs.ByteBufferReadable;
+import org.apache.tajo.util.FileUtil;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.ScatteringByteChannel;
+import java.nio.channels.spi.AbstractInterruptibleChannel;
+
+public class ByteBufInputChannel extends AbstractInterruptibleChannel implements ScatteringByteChannel {
+
+ ByteBufferReadable byteBufferReadable;
+ ReadableByteChannel channel;
+ InputStream inputStream;
+
+ public ByteBufInputChannel(InputStream inputStream) {
+ if (inputStream instanceof ByteBufferReadable) {
+ this.byteBufferReadable = (ByteBufferReadable) inputStream;
+ } else {
+ this.channel = Channels.newChannel(inputStream);
+ }
+
+ this.inputStream = inputStream;
+ }
+
+ @Override
+ public long read(ByteBuffer[] dsts, int offset, int length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long read(ByteBuffer[] dsts) {
+ return read(dsts, 0, dsts.length);
+ }
+
+ @Override
+ public int read(ByteBuffer dst) throws IOException {
+ if (byteBufferReadable != null) {
+ return byteBufferReadable.read(dst);
+ } else {
+ return channel.read(dst);
+ }
+ }
+
+ @Override
+ protected void implCloseChannel() throws IOException {
+ FileUtil.cleanup(null, channel, inputStream);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/ByteBufLineReader.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/ByteBufLineReader.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/ByteBufLineReader.java
new file mode 100644
index 0000000..f100931
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/ByteBufLineReader.java
@@ -0,0 +1,176 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.CharsetUtil;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class ByteBufLineReader implements Closeable {
+ private static int DEFAULT_BUFFER = 64 * 1024;
+
+ private int bufferSize;
+ private long readBytes;
+ private int startIndex;
+ private boolean eof = false;
+ private ByteBuf buffer;
+ private final ByteBufInputChannel channel;
+ private final AtomicInteger lineReadBytes = new AtomicInteger();
+ private final LineSplitProcessor processor = new LineSplitProcessor();
+
+ public ByteBufLineReader(ByteBufInputChannel channel) {
+ this(channel, BufferPool.directBuffer(DEFAULT_BUFFER));
+ }
+
+ public ByteBufLineReader(ByteBufInputChannel channel, ByteBuf buf) {
+ this.readBytes = 0;
+ this.channel = channel;
+ this.buffer = buf;
+ this.bufferSize = buf.capacity();
+ }
+
+ public long readBytes() {
+ return readBytes - buffer.readableBytes();
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (this.buffer.refCnt() > 0) {
+ this.buffer.release();
+ }
+ this.channel.close();
+ }
+
+ public String readLine() throws IOException {
+ ByteBuf buf = readLineBuf(lineReadBytes);
+ if (buf != null) {
+ return buf.toString(CharsetUtil.UTF_8);
+ }
+ return null;
+ }
+
+ private void fillBuffer() throws IOException {
+
+ int tailBytes = 0;
+ if (this.readBytes > 0) {
+ //startIndex = 0, readIndex = tailBytes length, writable = (buffer capacity - tailBytes)
+ this.buffer.markReaderIndex();
+ this.buffer.discardReadBytes(); // compact the buffer
+ tailBytes = this.buffer.writerIndex();
+ if (!this.buffer.isWritable()) {
+ // a line bytes is large than the buffer
+ BufferPool.ensureWritable(buffer, bufferSize * 2);
+ this.bufferSize = buffer.capacity();
+ }
+ this.startIndex = 0;
+ }
+
+ boolean release = true;
+ try {
+ int readBytes = tailBytes;
+ // read only once
+ int localReadBytes = buffer.writeBytes(channel, this.bufferSize - readBytes);
+ if (localReadBytes < 0) {
+ if (buffer.isWritable()) {
+ //if read bytes is less than the buffer capacity, there is no more bytes in the channel
+ eof = true;
+ }
+ }
+ readBytes += localReadBytes;
+
+ this.readBytes += (readBytes - tailBytes);
+ release = false;
+
+ this.buffer.readerIndex(this.buffer.readerIndex() + tailBytes); //skip past buffer (tail)
+ } finally {
+ if (release) {
+ buffer.release();
+ }
+ }
+ }
+
+ /**
+ * Read a line terminated by one of CR, LF, or CRLF.
+ */
+ public ByteBuf readLineBuf(AtomicInteger reads) throws IOException {
+ int readBytes = 0; // newline + text line bytes
+ int newlineLength = 0; //length of terminating newline
+ int readable;
+
+ this.startIndex = buffer.readerIndex();
+
+ loop:
+ while (true) {
+ readable = buffer.readableBytes();
+ if (readable <= 0) {
+ buffer.readerIndex(this.startIndex);
+ fillBuffer(); //compact and fill buffer
+
+ //if buffer.writerIndex() is zero, there is no bytes in buffer
+ if (!buffer.isReadable() && buffer.writerIndex() == 0) {
+ reads.set(0);
+ return null;
+ } else {
+ //skip first newLine
+ if (processor.isPrevCharCR() && buffer.getByte(buffer.readerIndex()) == LineSplitProcessor.LF) {
+ buffer.skipBytes(1);
+ if(eof && !buffer.isReadable()) {
+ reads.set(1);
+ return null;
+ }
+
+ newlineLength++;
+ readBytes++;
+ startIndex = buffer.readerIndex();
+ }
+ }
+ readable = buffer.readableBytes();
+ }
+
+ int endIndex = buffer.forEachByte(buffer.readerIndex(), readable, processor);
+ if (endIndex < 0) {
+ //does not appeared terminating newline
+ buffer.readerIndex(buffer.writerIndex()); // set to end buffer
+ if(eof){
+ readBytes += (buffer.readerIndex() - startIndex);
+ break loop;
+ }
+ } else {
+ buffer.readerIndex(endIndex + 1);
+ readBytes += (buffer.readerIndex() - startIndex); //past newline + text line
+
+ //appeared terminating CRLF
+ if (processor.isPrevCharCR() && buffer.isReadable()
+ && buffer.getByte(buffer.readerIndex()) == LineSplitProcessor.LF) {
+ buffer.skipBytes(1);
+ readBytes++;
+ newlineLength += 2;
+ } else {
+ newlineLength += 1;
+ }
+ break loop;
+ }
+ }
+ reads.set(readBytes);
+ return buffer.slice(startIndex, readBytes - newlineLength);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineDeserializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineDeserializer.java
new file mode 100644
index 0000000..1969d90
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineDeserializer.java
@@ -0,0 +1,99 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufProcessor;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+
+public class CSVLineDeserializer extends TextLineDeserializer {
+ private ByteBufProcessor processor;
+ private FieldSerializerDeserializer fieldSerDer;
+ private ByteBuf nullChars;
+ private int delimiterCompensation;
+
+ public CSVLineDeserializer(Schema schema, TableMeta meta, int[] targetColumnIndexes) {
+ super(schema, meta, targetColumnIndexes);
+ }
+
+ @Override
+ public void init() {
+ byte[] delimiter = CSVLineSerDe.getFieldDelimiter(meta);
+ this.processor = new FieldSplitProcessor(delimiter[0]);
+ this.delimiterCompensation = delimiter.length - 1;
+
+ if (nullChars != null) {
+ nullChars.release();
+ }
+ nullChars = TextLineSerDe.getNullChars(meta);
+
+ fieldSerDer = new TextFieldSerializerDeserializer(meta);
+ }
+
+ public void deserialize(final ByteBuf lineBuf, Tuple output) throws IOException, TextLineParsingError {
+ int[] projection = targetColumnIndexes;
+ if (lineBuf == null || targetColumnIndexes == null || targetColumnIndexes.length == 0) {
+ return;
+ }
+
+ final int rowLength = lineBuf.readableBytes();
+ int start = 0, fieldLength = 0, end = 0;
+
+ //Projection
+ int currentTarget = 0;
+ int currentIndex = 0;
+
+ while (end != -1) {
+ end = lineBuf.forEachByte(start, rowLength - start, processor);
+
+ if (end < 0) {
+ fieldLength = rowLength - start;
+ } else {
+ fieldLength = end - start - delimiterCompensation;
+ }
+
+ if (projection.length > currentTarget && currentIndex == projection[currentTarget]) {
+ lineBuf.setIndex(start, start + fieldLength);
+ Datum datum = fieldSerDer.deserialize(lineBuf, schema.getColumn(currentIndex), currentIndex, nullChars);
+ output.put(currentIndex, datum);
+ currentTarget++;
+ }
+
+ if (projection.length == currentTarget) {
+ break;
+ }
+
+ start = end + 1;
+ currentIndex++;
+ }
+ }
+
+ @Override
+ public void release() {
+ if (nullChars != null) {
+ nullChars.release();
+ nullChars = null;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineSerDe.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineSerDe.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineSerDe.java
new file mode 100644
index 0000000..566e5c9
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineSerDe.java
@@ -0,0 +1,42 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import org.apache.commons.lang.StringEscapeUtils;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.storage.StorageConstants;
+import org.apache.tajo.util.Bytes;
+
+public class CSVLineSerDe extends TextLineSerDe {
+ @Override
+ public TextLineDeserializer createDeserializer(Schema schema, TableMeta meta, int[] targetColumnIndexes) {
+ return new CSVLineDeserializer(schema, meta, targetColumnIndexes);
+ }
+
+ @Override
+ public TextLineSerializer createSerializer(Schema schema, TableMeta meta) {
+ return new CSVLineSerializer(schema, meta);
+ }
+
+ public static byte[] getFieldDelimiter(TableMeta meta) {
+ return StringEscapeUtils.unescapeJava(meta.getOption(StorageConstants.TEXT_DELIMITER,
+ StorageConstants.DEFAULT_FIELD_DELIMITER)).getBytes(Bytes.UTF8_CHARSET);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineSerializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineSerializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineSerializer.java
new file mode 100644
index 0000000..e1c7375
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/CSVLineSerializer.java
@@ -0,0 +1,118 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.datum.AnyDatum;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.exception.UnsupportedException;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+public class CSVLineSerializer extends TextLineSerializer {
+ private FieldSerializerDeserializer serde;
+
+ private byte[] nullChars;
+ private byte[] delimiter;
+ private int columnNum;
+
+ private final static String PARAM_DELIM = "|\t_";
+
+ public CSVLineSerializer(Schema schema, TableMeta meta) {
+ super(schema, meta);
+ }
+
+ @Override
+ public void init() {
+ nullChars = TextLineSerDe.getNullCharsAsBytes(meta);
+ delimiter = "|,_".getBytes();
+ columnNum = schema.size();
+
+ serde = new TextFieldSerializerDeserializer(meta);
+ }
+
+ @Override
+ public int serialize(OutputStream out, Tuple input) throws IOException {
+ int writtenBytes = 0;
+
+ for (int i = 0; i < columnNum; i++) {
+ Datum datum = input.get(i);
+ String typeStr;
+ if (datum.type() == TajoDataTypes.Type.ANY) {
+ typeStr = getTypeString(((AnyDatum)datum).getActual());
+ } else {
+ typeStr = getTypeString(datum);
+ }
+ out.write(typeStr.getBytes());
+ out.write(PARAM_DELIM.getBytes());
+
+ writtenBytes += serde.serialize(out, datum, schema.getColumn(i), i, nullChars);
+
+ if (columnNum - 1 > i) {
+ out.write(delimiter);
+ writtenBytes += delimiter.length;
+ }
+ }
+
+ return writtenBytes;
+ }
+
+ @Override
+ public void release() {
+
+ }
+
+ private static String getTypeString(Datum val) {
+ switch (val.type()) {
+ case NULL_TYPE:
+ return "-";
+ case BOOLEAN:
+ return "B";
+ case INT1:
+ case INT2:
+ case INT4:
+ return "I";
+ case INT8:
+ return "L";
+ case FLOAT4:
+ return "F";
+ case FLOAT8:
+ return "D";
+ case NUMERIC:
+ return "E";
+ case CHAR:
+ case TEXT:
+ return "C";
+ case DATE:
+ case TIME:
+ case TIMESTAMP:
+ return "T";
+ case BLOB:
+ case INET4:
+ case INET6:
+ return "A";
+ default:
+ throw new UnsupportedException(val.type().name());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/FieldSerializerDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/FieldSerializerDeserializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/FieldSerializerDeserializer.java
new file mode 100644
index 0000000..29a5aa2
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/FieldSerializerDeserializer.java
@@ -0,0 +1,36 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.datum.Datum;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+
+public interface FieldSerializerDeserializer {
+
+ int serialize(OutputStream out, Datum datum, Column col, int columnIndex, byte[] nullChars) throws IOException;
+
+ Datum deserialize(ByteBuf buf, Column col, int columnIndex, ByteBuf nullChars)
+ throws IOException, TextLineParsingError;
+
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/FieldSplitProcessor.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/FieldSplitProcessor.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/FieldSplitProcessor.java
new file mode 100644
index 0000000..2e89d39
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/FieldSplitProcessor.java
@@ -0,0 +1,34 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import io.netty.buffer.ByteBufProcessor;
+
+public class FieldSplitProcessor implements ByteBufProcessor {
+ private byte delimiter; //the ascii separate character
+
+ public FieldSplitProcessor(byte recordDelimiterByte) {
+ this.delimiter = recordDelimiterByte;
+ }
+
+ @Override
+ public boolean process(byte value) throws Exception {
+ return delimiter != value;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/InputHandler.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/InputHandler.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/InputHandler.java
new file mode 100644
index 0000000..dcc53c1
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/InputHandler.java
@@ -0,0 +1,78 @@
+/*
+ * Lisensed 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.tajo.plan.function.stream;
+
+import org.apache.tajo.storage.Tuple;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * {@link InputHandler} is responsible for handling the input to the Tajo-Streaming external command.
+ *
+ */
+public class InputHandler implements Closeable {
+
+ private final static byte[] END_OF_RECORD_DELIM = "|_\n".getBytes();
+ private final static byte[] END_OF_STREAM = ("C" + "\\x04" + "|_\n").getBytes();
+
+ private final TextLineSerializer serializer;
+
+ private OutputStream out;
+
+ // flag to mark if close() has already been called
+ private boolean alreadyClosed = false;
+
+ public InputHandler(TextLineSerializer serializer) {
+ this.serializer = serializer;
+ }
+
+ /**
+ * Send the given input <code>Tuple</code> to the managed executable.
+ *
+ * @param t input <code>Tuple</code>
+ * @throws IOException
+ */
+ public void putNext(Tuple t) throws IOException {
+ serializer.serialize(out, t);
+ out.write(END_OF_RECORD_DELIM);
+ }
+
+ public void close() throws IOException {
+ if (!alreadyClosed) {
+ out.flush();
+ out.close();
+ out = null;
+ alreadyClosed = true;
+ }
+ }
+
+ /**
+ * Bind the <code>InputHandler</code> to the <code>OutputStream</code>
+ * from which it reads input and sends it to the managed process.
+ *
+ * @param os <code>OutputStream</code> from which to read input data for the
+ * managed process
+ * @throws IOException
+ */
+ public void bindTo(OutputStream os) throws IOException {
+ out = os;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/LineSplitProcessor.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/LineSplitProcessor.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/LineSplitProcessor.java
new file mode 100644
index 0000000..0415141
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/LineSplitProcessor.java
@@ -0,0 +1,45 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import io.netty.buffer.ByteBufProcessor;
+
+public class LineSplitProcessor implements ByteBufProcessor {
+ public static final byte CR = '\r';
+ public static final byte LF = '\n';
+ private boolean prevCharCR = false; //true of prev char was CR
+
+ @Override
+ public boolean process(byte value) throws Exception {
+ switch (value) {
+ case LF:
+ return false;
+ case CR:
+ prevCharCR = true;
+ return false;
+ default:
+ prevCharCR = false;
+ return true;
+ }
+ }
+
+ public boolean isPrevCharCR() {
+ return prevCharCR;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/OutputHandler.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/OutputHandler.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/OutputHandler.java
new file mode 100644
index 0000000..c5eb419
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/OutputHandler.java
@@ -0,0 +1,156 @@
+/*
+ * Lisensed 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.tajo.plan.function.stream;
+
+import com.google.common.base.Charsets;
+import io.netty.buffer.ByteBuf;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * {@link OutputHandler} is responsible for handling the output of the
+ * Tajo-Streaming external command.
+ */
+public class OutputHandler implements Closeable {
+ private static int DEFAULT_BUFFER = 64 * 1024;
+ private final static byte[] END_OF_RECORD_DELIM = "|_\n".getBytes();
+
+ private final TextLineDeserializer deserializer;
+
+ private ByteBufLineReader in = null;
+
+ private String currValue = null;
+
+ private InputStream istream;
+
+ private final ByteBuf buf = BufferPool.directBuffer(DEFAULT_BUFFER);
+
+ // Both of these ignore the trailing "\n". So if the default delimiter is "\n", recordDelimStr is "".
+ private String recordDelimStr = null;
+ private int recordDelimLength = 0;
+ private final Tuple tuple = new VTuple(1);
+
+ // flag to mark if close() has already been called
+ private boolean alreadyClosed = false;
+
+ public OutputHandler(TextLineDeserializer deserializer) {
+ this.deserializer = deserializer;
+ }
+
+ /**
+ * Bind the <code>OutputHandler</code> to the <code>InputStream</code>
+ * from which to read the output data of the managed process.
+ *
+ * @param is <code>InputStream</code> from which to read the output data
+ * of the managed process
+ * @throws IOException
+ */
+ public void bindTo(InputStream is) throws IOException {
+ this.istream = is;
+ this.in = new ByteBufLineReader(new ByteBufInputChannel(istream));
+ }
+
+ /**
+ * Get the next output <code>Tuple</code> of the managed process.
+ *
+ * @return the next output <code>Tuple</code> of the managed process
+ * @throws IOException
+ */
+ public Tuple getNext() throws IOException {
+ if (in == null) {
+ return null;
+ }
+
+ currValue = null;
+ if (!readValue()) {
+ return null;
+ }
+ buf.clear();
+ buf.writeBytes(currValue.getBytes());
+ try {
+ deserializer.deserialize(buf, tuple);
+ } catch (TextLineParsingError textLineParsingError) {
+ throw new IOException(textLineParsingError);
+ }
+ return tuple;
+ }
+
+ private boolean readValue() throws IOException {
+ currValue = in.readLine();
+ if (currValue == null) {
+ return false;
+ }
+
+ while(!isEndOfRow()) {
+ // Need to add back the newline character we ate.
+ currValue += '\n';
+
+ byte[] lineBytes = readNextLine();
+ if (lineBytes == null) {
+ // We have no more input, so just break;
+ break;
+ }
+ currValue += new String(lineBytes);
+ }
+
+ if (currValue.contains("|_")) {
+ int pos = currValue.lastIndexOf("|_");
+ currValue = currValue.substring(0, pos);
+ }
+
+ return true;
+ }
+
+ private byte[] readNextLine() throws IOException {
+ String line = in.readLine();
+ if (line == null) {
+ return null;
+ }
+
+ return line.getBytes();
+ }
+
+ private boolean isEndOfRow() {
+ if (recordDelimStr == null) {
+ byte[] recordDelimBa = END_OF_RECORD_DELIM;
+ recordDelimLength = recordDelimBa.length - 1; //Ignore trailing \n
+ recordDelimStr = new String(recordDelimBa, 0, recordDelimLength, Charsets.UTF_8);
+ }
+ if (recordDelimLength == 0 || currValue.length() < recordDelimLength) {
+ return true;
+ }
+ return currValue.contains(recordDelimStr);
+ }
+
+ /**
+ * Close the <code>OutputHandler</code>.
+ * @throws IOException
+ */
+ public void close() throws IOException {
+ if(!alreadyClosed) {
+ istream.close();
+ istream = null;
+ alreadyClosed = true;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/StreamingUtil.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/StreamingUtil.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/StreamingUtil.java
new file mode 100644
index 0000000..9bb60dd
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/StreamingUtil.java
@@ -0,0 +1,91 @@
+/*
+ * Lisensed 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.tajo.plan.function.stream;
+
+import org.apache.tajo.util.TUtil;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+public class StreamingUtil {
+
+ private static final String BASH = "bash";
+ private static final String PATH = "PATH";
+
+ /**
+ * Create an external process for StreamingCommand command.
+ *
+ * @param argv process arguments
+ * @return
+ */
+ public static ProcessBuilder createProcess(String[] argv) {
+ // Set the actual command to run with 'bash -c exec ...'
+ List<String> cmdArgs = TUtil.newList();
+
+ StringBuffer argBuffer = new StringBuffer();
+ for (String arg : argv) {
+ argBuffer.append(arg);
+ argBuffer.append(" ");
+ }
+ String argvAsString = argBuffer.toString();
+
+ if (System.getProperty("os.name").toUpperCase().startsWith("WINDOWS")) {
+ cmdArgs.add("cmd");
+ cmdArgs.add("/c");
+ cmdArgs.add(argvAsString);
+ } else {
+ cmdArgs.add(BASH);
+ cmdArgs.add("-c");
+ StringBuffer sb = new StringBuffer();
+ sb.append("exec ");
+ sb.append(argvAsString);
+ cmdArgs.add(sb.toString());
+ }
+
+ // Start the external process
+ ProcessBuilder processBuilder = new ProcessBuilder(cmdArgs
+ .toArray(new String[cmdArgs.size()]));
+ setupEnvironment(processBuilder);
+ return processBuilder;
+ }
+
+ /**
+ * Set up the run-time environment of the managed process.
+ *
+ * @param pb {@link ProcessBuilder} used to exec the process
+ */
+ private static void setupEnvironment(ProcessBuilder pb) {
+ String separator = ":";
+ Map<String, String> env = pb.environment();
+
+ // Add the current-working-directory to the $PATH
+ File dir = pb.directory();
+ String cwd = (dir != null) ? dir.getAbsolutePath() : System
+ .getProperty("user.dir");
+
+ String envPath = env.get(PATH);
+ if (envPath == null) {
+ envPath = cwd;
+ } else {
+ envPath = envPath + separator + cwd;
+ }
+ env.put(PATH, envPath);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextFieldSerializerDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextFieldSerializerDeserializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextFieldSerializerDeserializer.java
new file mode 100644
index 0000000..cd9518b
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextFieldSerializerDeserializer.java
@@ -0,0 +1,257 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import com.google.protobuf.Message;
+import io.netty.buffer.ByteBuf;
+import io.netty.util.CharsetUtil;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.tajo.TajoConstants;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.datum.*;
+import org.apache.tajo.datum.protobuf.ProtobufJsonFormat;
+import org.apache.tajo.exception.UnsupportedException;
+import org.apache.tajo.storage.StorageConstants;
+import org.apache.tajo.util.NumberUtil;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.CharsetDecoder;
+import java.util.TimeZone;
+
+public class TextFieldSerializerDeserializer implements FieldSerializerDeserializer {
+ public static final byte[] trueBytes = "true".getBytes();
+ public static final byte[] falseBytes = "false".getBytes();
+ private static ProtobufJsonFormat protobufJsonFormat = ProtobufJsonFormat.getInstance();
+ private final CharsetDecoder decoder = CharsetUtil.getDecoder(CharsetUtil.UTF_8);
+
+ private final boolean hasTimezone;
+ private final TimeZone timezone;
+
+ public TextFieldSerializerDeserializer(TableMeta meta) {
+ hasTimezone = meta.containsOption(StorageConstants.TIMEZONE);
+ timezone = TimeZone.getTimeZone(meta.getOption(StorageConstants.TIMEZONE, TajoConstants.DEFAULT_SYSTEM_TIMEZONE));
+ }
+
+ private static boolean isNull(ByteBuf val, ByteBuf nullBytes) {
+ return !val.isReadable() || nullBytes.equals(val);
+ }
+
+ private static boolean isNullText(ByteBuf val, ByteBuf nullBytes) {
+ return val.readableBytes() > 0 && nullBytes.equals(val);
+ }
+
+ @Override
+ public int serialize(OutputStream out, Datum datum, Column col, int columnIndex, byte[] nullChars)
+ throws IOException {
+ byte[] bytes;
+ int length = 0;
+ TajoDataTypes.DataType dataType = col.getDataType();
+
+ if (datum == null || datum instanceof NullDatum) {
+ switch (dataType.getType()) {
+ case CHAR:
+ case TEXT:
+ length = nullChars.length;
+ out.write(nullChars);
+ break;
+ default:
+ break;
+ }
+ return length;
+ }
+
+ switch (dataType.getType()) {
+ case BOOLEAN:
+ out.write(datum.asBool() ? trueBytes : falseBytes);
+ length = trueBytes.length;
+ break;
+ case CHAR:
+ byte[] pad = new byte[dataType.getLength() - datum.size()];
+ bytes = datum.asTextBytes();
+ out.write(bytes);
+ out.write(pad);
+ length = bytes.length + pad.length;
+ break;
+ case TEXT:
+ case BIT:
+ case INT2:
+ case INT4:
+ case INT8:
+ case FLOAT4:
+ case FLOAT8:
+ case INET4:
+ case DATE:
+ case INTERVAL:
+ bytes = datum.asTextBytes();
+ length = bytes.length;
+ out.write(bytes);
+ break;
+ case TIME:
+ if (hasTimezone) {
+ bytes = ((TimeDatum) datum).asChars(timezone, true).getBytes();
+ } else {
+ bytes = datum.asTextBytes();
+ }
+ length = bytes.length;
+ out.write(bytes);
+ break;
+ case TIMESTAMP:
+ if (hasTimezone) {
+ bytes = ((TimestampDatum) datum).asChars(timezone, true).getBytes();
+ } else {
+ bytes = datum.asTextBytes();
+ }
+ length = bytes.length;
+ out.write(bytes);
+ break;
+ case INET6:
+ case BLOB:
+ bytes = Base64.encodeBase64(datum.asByteArray(), false);
+ length = bytes.length;
+ out.write(bytes, 0, length);
+ break;
+ case PROTOBUF:
+ ProtobufDatum protobuf = (ProtobufDatum) datum;
+ byte[] protoBytes = protobufJsonFormat.printToString(protobuf.get()).getBytes();
+ length = protoBytes.length;
+ out.write(protoBytes, 0, protoBytes.length);
+ break;
+ case NULL_TYPE:
+ break;
+ case ANY:
+ AnyDatum anyDatum = (AnyDatum) datum;
+ length = serialize(out, anyDatum.getActual(), new Column("any", anyDatum.getActual().type()), 0, nullChars);
+ break;
+ default:
+ throw new UnsupportedException(dataType.getType().name());
+ }
+ return length;
+ }
+
+ @Override
+ public Datum deserialize(ByteBuf buf, Column col, int columnIndex, ByteBuf nullChars) throws IOException {
+ Datum datum;
+ TajoDataTypes.Type type = col.getDataType().getType();
+ boolean nullField;
+ if (type == TajoDataTypes.Type.TEXT || type == TajoDataTypes.Type.CHAR) {
+ nullField = isNullText(buf, nullChars);
+ } else {
+ nullField = isNull(buf, nullChars);
+ }
+
+ if (nullField) {
+ datum = NullDatum.get();
+ } else {
+ switch (type) {
+ case BOOLEAN:
+ byte bool = buf.readByte();
+ datum = DatumFactory.createBool(bool == 't' || bool == 'T');
+ break;
+ case BIT:
+ datum = DatumFactory.createBit(Byte.parseByte(
+ decoder.decode(buf.nioBuffer(buf.readerIndex(), buf.readableBytes())).toString()));
+ break;
+ case CHAR:
+ datum = DatumFactory.createChar(
+ decoder.decode(buf.nioBuffer(buf.readerIndex(), buf.readableBytes())).toString().trim());
+ break;
+ case INT1:
+ case INT2:
+ datum = DatumFactory.createInt2((short) NumberUtil.parseInt(buf));
+ break;
+ case INT4:
+ datum = DatumFactory.createInt4(NumberUtil.parseInt(buf));
+ break;
+ case INT8:
+ datum = DatumFactory.createInt8(NumberUtil.parseLong(buf));
+ break;
+ case FLOAT4:
+ datum = DatumFactory.createFloat4(
+ decoder.decode(buf.nioBuffer(buf.readerIndex(), buf.readableBytes())).toString());
+ break;
+ case FLOAT8:
+ datum = DatumFactory.createFloat8(NumberUtil.parseDouble(buf));
+ break;
+ case TEXT: {
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.readBytes(bytes);
+ datum = DatumFactory.createText(bytes);
+ break;
+ }
+ case DATE:
+ datum = DatumFactory.createDate(
+ decoder.decode(buf.nioBuffer(buf.readerIndex(), buf.readableBytes())).toString());
+ break;
+ case TIME:
+ if (hasTimezone) {
+ datum = DatumFactory.createTime(
+ decoder.decode(buf.nioBuffer(buf.readerIndex(), buf.readableBytes())).toString(), timezone);
+ } else {
+ datum = DatumFactory.createTime(
+ decoder.decode(buf.nioBuffer(buf.readerIndex(), buf.readableBytes())).toString());
+ }
+ break;
+ case TIMESTAMP:
+ if (hasTimezone) {
+ datum = DatumFactory.createTimestamp(
+ decoder.decode(buf.nioBuffer(buf.readerIndex(), buf.readableBytes())).toString(), timezone);
+ } else {
+ datum = DatumFactory.createTimestamp(
+ decoder.decode(buf.nioBuffer(buf.readerIndex(), buf.readableBytes())).toString());
+ }
+ break;
+ case INTERVAL:
+ datum = DatumFactory.createInterval(
+ decoder.decode(buf.nioBuffer(buf.readerIndex(), buf.readableBytes())).toString());
+ break;
+ case PROTOBUF: {
+ ProtobufDatumFactory factory = ProtobufDatumFactory.get(col.getDataType());
+ Message.Builder builder = factory.newBuilder();
+ try {
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.readBytes(bytes);
+ protobufJsonFormat.merge(bytes, builder);
+ datum = factory.createDatum(builder.build());
+ } catch (IOException e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ break;
+ }
+ case INET4:
+ datum = DatumFactory.createInet4(
+ decoder.decode(buf.nioBuffer(buf.readerIndex(), buf.readableBytes())).toString());
+ break;
+ case BLOB: {
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.readBytes(bytes);
+ datum = DatumFactory.createBlob(Base64.decodeBase64(bytes));
+ break;
+ }
+ default:
+ datum = NullDatum.get();
+ break;
+ }
+ }
+ return datum;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineDeserializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineDeserializer.java
new file mode 100644
index 0000000..c2ea30c
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineDeserializer.java
@@ -0,0 +1,60 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+
+/**
+ * Reads a text line and fills a Tuple with values
+ */
+public abstract class TextLineDeserializer {
+ protected final Schema schema;
+ protected final TableMeta meta;
+ protected final int[] targetColumnIndexes;
+
+ public TextLineDeserializer(Schema schema, TableMeta meta, int[] targetColumnIndexes) {
+ this.schema = schema;
+ this.meta = meta;
+ this.targetColumnIndexes = targetColumnIndexes;
+ }
+
+ /**
+ * Initialize SerDe
+ */
+ public abstract void init();
+
+ /**
+ * It fills a tuple with a read fields in a given line.
+ *
+ * @param buf Read line
+ * @param output Tuple to be filled with read fields
+ * @throws IOException
+ */
+ public abstract void deserialize(final ByteBuf buf, Tuple output) throws IOException, TextLineParsingError;
+
+ /**
+ * Release external resources
+ */
+ public abstract void release();
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineParsingError.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineParsingError.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineParsingError.java
new file mode 100644
index 0000000..9048e27
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineParsingError.java
@@ -0,0 +1,31 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+public class TextLineParsingError extends Exception {
+
+ public TextLineParsingError(Throwable t) {
+ super(t);
+ }
+
+ public TextLineParsingError(String message, Throwable t) {
+ super(t.getMessage() + ", Error line: " + message);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineSerDe.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineSerDe.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineSerDe.java
new file mode 100644
index 0000000..89b169c
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineSerDe.java
@@ -0,0 +1,65 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.commons.lang.StringEscapeUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.storage.StorageConstants;
+import org.apache.tajo.util.Bytes;
+
+/**
+ * Pluggable Text Line SerDe class
+ */
+public abstract class TextLineSerDe {
+
+ public TextLineSerDe() {
+ }
+
+ public abstract TextLineDeserializer createDeserializer(Schema schema, TableMeta meta, int [] targetColumnIndexes);
+
+ public abstract TextLineSerializer createSerializer(Schema schema, TableMeta meta);
+
+ public static ByteBuf getNullChars(TableMeta meta) {
+ byte[] nullCharByteArray = getNullCharsAsBytes(meta);
+
+ ByteBuf nullChars = BufferPool.directBuffer(nullCharByteArray.length, nullCharByteArray.length);
+ nullChars.writeBytes(nullCharByteArray);
+
+ return nullChars;
+ }
+
+ public static byte [] getNullCharsAsBytes(TableMeta meta) {
+ byte [] nullChars;
+
+ String nullCharacters = StringEscapeUtils.unescapeJava(meta.getOption(StorageConstants.TEXT_NULL,
+ NullDatum.DEFAULT_TEXT));
+ if (StringUtils.isEmpty(nullCharacters)) {
+ nullChars = NullDatum.get().asTextBytes();
+ } else {
+ nullChars = nullCharacters.getBytes(Bytes.UTF8_CHARSET);
+ }
+
+ return nullChars;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineSerializer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineSerializer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineSerializer.java
new file mode 100644
index 0000000..ea8576c
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/TextLineSerializer.java
@@ -0,0 +1,45 @@
+/**
+ * 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.tajo.plan.function.stream;
+
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * Write a Tuple into single text formatted line
+ */
+public abstract class TextLineSerializer {
+ protected Schema schema;
+ protected TableMeta meta;
+
+ public TextLineSerializer(Schema schema, TableMeta meta) {
+ this.schema = schema;
+ this.meta = meta;
+ }
+
+ public abstract void init();
+
+ public abstract int serialize(OutputStream out, Tuple input) throws IOException;
+
+ public abstract void release();
+}
[3/4] tajo git commit: TAJO-1344: Python UDF support. (jihoon)
Posted by ji...@apache.org.
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/resources/python/controller.py
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/resources/python/controller.py b/tajo-core/src/main/resources/python/controller.py
new file mode 100644
index 0000000..d969b34
--- /dev/null
+++ b/tajo-core/src/main/resources/python/controller.py
@@ -0,0 +1,330 @@
+#! /usr/bin/env python
+############################################################################
+# 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 sys
+import os
+import logging
+import base64
+
+from datetime import datetime
+try:
+ from dateutil import parser
+ USE_DATEUTIL = True
+except ImportError:
+ USE_DATEUTIL = False
+
+from tajo_util import write_user_exception, udf_logging
+
+FIELD_DELIMITER = ','
+TUPLE_START = '('
+TUPLE_END = ')'
+BAG_START = '{'
+BAG_END = '}'
+MAP_START = '['
+MAP_END = ']'
+MAP_KEY = '#'
+PARAMETER_DELIMITER = '\t'
+PRE_WRAP_DELIM = '|'
+POST_WRAP_DELIM = '_'
+NULL_BYTE = "-"
+END_RECORD_DELIM = '|_\n'
+END_RECORD_DELIM_LENGTH = len(END_RECORD_DELIM)
+
+WRAPPED_FIELD_DELIMITER = PRE_WRAP_DELIM + FIELD_DELIMITER + POST_WRAP_DELIM
+WRAPPED_TUPLE_START = PRE_WRAP_DELIM + TUPLE_START + POST_WRAP_DELIM
+WRAPPED_TUPLE_END = PRE_WRAP_DELIM + TUPLE_END + POST_WRAP_DELIM
+WRAPPED_BAG_START = PRE_WRAP_DELIM + BAG_START + POST_WRAP_DELIM
+WRAPPED_BAG_END = PRE_WRAP_DELIM + BAG_END + POST_WRAP_DELIM
+WRAPPED_MAP_START = PRE_WRAP_DELIM + MAP_START + POST_WRAP_DELIM
+WRAPPED_MAP_END = PRE_WRAP_DELIM + MAP_END + POST_WRAP_DELIM
+WRAPPED_PARAMETER_DELIMITER = PRE_WRAP_DELIM + PARAMETER_DELIMITER + POST_WRAP_DELIM
+WRAPPED_NULL_BYTE = PRE_WRAP_DELIM + NULL_BYTE + POST_WRAP_DELIM
+
+TYPE_TUPLE = TUPLE_START
+TYPE_BAG = BAG_START
+TYPE_MAP = MAP_START
+
+TYPE_BOOLEAN = "B"
+TYPE_INTEGER = "I"
+TYPE_LONG = "L"
+TYPE_FLOAT = "F"
+TYPE_DOUBLE = "D"
+TYPE_BYTEARRAY = "A"
+TYPE_CHARARRAY = "C"
+TYPE_DATETIME = "T"
+TYPE_BIGINTEGER = "N"
+TYPE_BIGDECIMAL = "E"
+
+END_OF_STREAM = TYPE_CHARARRAY + "\x04" + END_RECORD_DELIM
+TURN_ON_OUTPUT_CAPTURING = TYPE_CHARARRAY + "TURN_ON_OUTPUT_CAPTURING" + END_RECORD_DELIM
+NUM_LINES_OFFSET_TRACE = int(os.environ.get('PYTHON_TRACE_OFFSET', 0))
+
+class PythonStreamingController:
+ def __init__(self, profiling_mode=False):
+ self.profiling_mode = profiling_mode
+
+ def main(self,
+ module_name, file_path, func_name, cache_path,
+ output_stream_path, error_stream_path, log_file_name, output_schema):
+ sys.stdin = os.fdopen(sys.stdin.fileno(), 'rb', 0)
+
+ # Need to ensure that user functions can't write to the streams we use to communicate with pig.
+ self.stream_output = os.fdopen(sys.stdout.fileno(), 'wb', 0)
+ self.stream_error = os.fdopen(sys.stderr.fileno(), 'wb', 0)
+
+ self.input_stream = sys.stdin
+ # TODO: support controller logging
+ # self.log_stream = open(output_stream_path, 'a')
+ # sys.stderr = open(error_stream_path, 'w')
+
+ sys.path.append(file_path)
+ sys.path.append(cache_path)
+ sys.path.append('.')
+
+ should_log = False
+ if should_log:
+ logging.basicConfig(filename=log_file_name, format="%(asctime)s %(levelname)s %(message)s", level=udf_logging.udf_log_level)
+ logging.info("To reduce the amount of information being logged only a small subset of rows are logged at the "
+ "INFO level. Call udf_logging.set_log_level_debug in tajo_util to see all rows being processed.")
+
+ input_str = self.get_next_input()
+
+ try:
+ func = __import__(module_name, globals(), locals(), [func_name], -1).__dict__[func_name]
+ except:
+ # These errors should always be caused by user code.
+ write_user_exception(module_name, self.stream_error, NUM_LINES_OFFSET_TRACE)
+ self.close_controller(-1)
+
+ log_message = logging.info
+ if udf_logging.udf_log_level == logging.DEBUG:
+ log_message = logging.debug
+
+ while input_str != END_OF_STREAM:
+ try:
+ try:
+ if should_log:
+ log_message("Serialized Input: %s" % (input_str))
+ inputs = deserialize_input(input_str)
+ if should_log:
+ log_message("Deserialized Input: %s" % (unicode(inputs)))
+ except:
+ # Capture errors where the user passes in bad data.
+ write_user_exception(module_name, self.stream_error, NUM_LINES_OFFSET_TRACE)
+ self.close_controller(-3)
+
+ try:
+ func_output = func(*inputs)
+ if should_log:
+ log_message("UDF Output: %s" % (unicode(func_output)))
+ except:
+ # These errors should always be caused by user code.
+ write_user_exception(module_name, self.stream_error, NUM_LINES_OFFSET_TRACE)
+ self.close_controller(-2)
+
+ output = serialize_output(func_output, output_schema)
+ if should_log:
+ log_message("Serialized Output: %s" % (output))
+
+ self.stream_output.write( "%s%s" % (output, END_RECORD_DELIM) )
+ except Exception as e:
+ # This should only catch internal exceptions with the controller
+ # and pig- not with user code.
+ import traceback
+ traceback.print_exc(file=self.stream_error)
+ sys.exit(-3)
+
+ sys.stdout.flush()
+ sys.stderr.flush()
+ self.stream_output.flush()
+ self.stream_error.flush()
+
+ input_str = self.get_next_input()
+
+ def get_next_input(self):
+ input_stream = self.input_stream
+ # log_stream = self.log_stream
+
+ input_str = input_stream.readline()
+
+ while input_str.endswith(END_RECORD_DELIM) == False:
+ line = input_stream.readline()
+ if line == '':
+ input_str = ''
+ break
+ input_str += line
+
+ if input_str == '':
+ return END_OF_STREAM
+
+ if input_str == END_OF_STREAM:
+ return input_str
+
+ return input_str[:-END_RECORD_DELIM_LENGTH]
+
+ def close_controller(self, exit_code):
+ sys.stderr.close()
+ self.stream_error.write("\n")
+ self.stream_error.close()
+ sys.stdout.close()
+ self.stream_output.write("\n")
+ self.stream_output.close()
+ sys.exit(exit_code)
+
+def deserialize_input(input_str):
+ if len(input_str) == 0:
+ return []
+
+ return [_deserialize_input(param, 0, len(param)) for param in input_str.split(WRAPPED_FIELD_DELIMITER)]
+
+def _deserialize_input(input_str, si, ei):
+ len = ei - si + 1
+ if len < 1:
+ # Handle all of the cases where you can have valid empty input.
+ if ei == si:
+ if input_str[si] == TYPE_CHARARRAY:
+ return u""
+ elif input_str[si] == TYPE_BYTEARRAY:
+ return bytearray("")
+ else:
+ raise Exception("Got input type flag %s, but no data to go with it.\nInput string: %s\nSlice: %s" % (input_str[si], input_str, input_str[si:ei+1]))
+ else:
+ raise Exception("Start index %d greater than end index %d.\nInput string: %s\n, Slice: %s" % (si, ei, input_str[si:ei+1]))
+
+ tokens = input_str.split(WRAPPED_PARAMETER_DELIMITER)
+ schema = tokens[0];
+ param = tokens[1];
+
+ if schema == NULL_BYTE:
+ return None
+ elif schema == TYPE_CHARARRAY:
+ return unicode(param, 'utf-8')
+ elif schema == TYPE_BYTEARRAY:
+ return bytearray(param)
+ elif schema == TYPE_INTEGER:
+ return int(param)
+ elif schema == TYPE_LONG or schema == TYPE_BIGINTEGER:
+ return long(param)
+ elif schema == TYPE_FLOAT or schema == TYPE_DOUBLE or schema == TYPE_BIGDECIMAL:
+ return float(param)
+ elif schema == TYPE_BOOLEAN:
+ return param == "true"
+ elif schema == TYPE_DATETIME:
+ # Format is "yyyy-MM-ddTHH:mm:ss.SSS+00:00" or "2013-08-23T18:14:03.123+ZZ"
+ if USE_DATEUTIL:
+ return parser.parse(param)
+ else:
+ # Try to use datetime even though it doesn't handle time zones properly,
+ # We only use the first 3 microsecond digits and drop time zone (first 23 characters)
+ return datetime.strptime(param, "%Y-%m-%dT%H:%M:%S.%f")
+ else:
+ raise Exception("Can't determine type of input: %s" % param)
+
+def _deserialize_collection(input_str, return_type, si, ei):
+ list_result = []
+ append_to_list_result = list_result.append
+ dict_result = {}
+
+ index = si
+ field_start = si
+ depth = 0
+
+ key = None
+
+ # recurse to deserialize elements if the collection is not empty
+ if ei-si+1 > 0:
+ while True:
+ if index >= ei - 2:
+ if return_type == TYPE_MAP:
+ dict_result[key] = _deserialize_input(input_str, value_start, ei)
+ else:
+ append_to_list_result(_deserialize_input(input_str, field_start, ei))
+ break
+
+ if return_type == TYPE_MAP and not key:
+ key_index = input_str.find(MAP_KEY, index)
+ key = unicode(input_str[index+1:key_index], 'utf-8')
+ index = key_index + 1
+ value_start = key_index + 1
+ continue
+
+ if not (input_str[index] == PRE_WRAP_DELIM and input_str[index+2] == POST_WRAP_DELIM):
+ prewrap_index = input_str.find(PRE_WRAP_DELIM, index+1)
+ index = (prewrap_index if prewrap_index != -1 else end_index)
+ continue
+
+ mid = input_str[index+1]
+
+ if mid == BAG_START or mid == TUPLE_START or mid == MAP_START:
+ depth += 1
+ elif mid == BAG_END or mid == TUPLE_END or mid == MAP_END:
+ depth -= 1
+ elif depth == 0 and mid == FIELD_DELIMITER:
+ if return_type == TYPE_MAP:
+ dict_result[key] = _deserialize_input(input_str, value_start, index - 1)
+ key = None
+ else:
+ append_to_list_result(_deserialize_input(input_str, field_start, index - 1))
+ field_start = index + 3
+
+ index += 3
+
+ if return_type == TYPE_MAP:
+ return dict_result
+ elif return_type == TYPE_TUPLE:
+ return tuple(list_result)
+ else:
+ return list_result
+
+def wrap_tuple(o, serialized_item):
+ if type(o) != tuple:
+ return WRAPPED_TUPLE_START + serialized_item + WRAPPED_TUPLE_END
+ else:
+ return serialized_item
+
+def serialize_output(output, out_schema, utfEncodeAllFields=False):
+ """
+ @param utfEncodeStrings - Generally we want to utf encode only strings. But for
+ Maps we utf encode everything because on the Java side we don't know the schema
+ for maps so we wouldn't be able to tell which fields were encoded or not.
+ """
+
+ output_type = type(output)
+
+ if output is None:
+ result = WRAPPED_NULL_BYTE
+ elif output_type == bool:
+ result = ("true" if output else "false")
+ elif output_type == bytearray:
+ result = str(output)
+ elif output_type == datetime:
+ result = output.isoformat()
+ elif utfEncodeAllFields or output_type == str or output_type == unicode:
+ # unicode is necessary in cases where we're encoding non-strings.
+ result = unicode(output).encode('utf-8')
+ else:
+ result = str(output)
+
+ if out_schema == "blob":
+ return base64.b64encode(result)
+ else:
+ return result
+
+if __name__ == '__main__':
+ controller = PythonStreamingController()
+ controller.main(sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4],
+ sys.argv[5], sys.argv[6], sys.argv[7], sys.argv[8])
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/resources/python/tajo_util.py
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/resources/python/tajo_util.py b/tajo-core/src/main/resources/python/tajo_util.py
new file mode 100644
index 0000000..77b28a6
--- /dev/null
+++ b/tajo-core/src/main/resources/python/tajo_util.py
@@ -0,0 +1,103 @@
+#!/usr/bin/python
+
+############################################################################
+# 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 logging
+
+class udf_logging(object):
+ udf_log_level = logging.INFO
+
+ @classmethod
+ def set_log_level_error(cls):
+ cls.udf_log_level = logging.ERROR
+
+ @classmethod
+ def set_log_level_warn(cls):
+ cls.udf_log_level = logging.WARN
+
+ @classmethod
+ def set_log_level_info(cls):
+ cls.udf_log_level = logging.INFO
+
+ @classmethod
+ def set_log_level_debug(cls):
+ cls.udf_log_level = logging.DEBUG
+
+def outputType(type_str):
+ def wrap(f):
+ def wrapped_f(*args):
+ return f(*args)
+ return wrapped_f
+ return wrap
+
+def write_user_exception(filename, stream_err_output, num_lines_offset_trace=0):
+ import sys
+ import traceback
+ import inspect
+ (t, v, tb) = sys.exc_info()
+ name = t.__name__
+ record_error = False
+
+ if name in ['SyntaxError', 'IndentationError']:
+ syntax_error_values = v.args
+ user_line_number = syntax_error_values[1][1] - num_lines_offset_trace
+ error_message = "%s: %s\n\tFile: %s, line %s column %s\n\t%s" % \
+ (name,
+ syntax_error_values[0],
+ syntax_error_values[1][0],
+ user_line_number,
+ syntax_error_values[1][2],
+ syntax_error_values[1][3])
+ else:
+ error_message = "%s: %s\n" % (name, v)
+ user_line_number = None
+ while 1:
+ e_file_name = tb.tb_frame.f_code.co_filename
+ if e_file_name.find(filename) > 0:
+ record_error = True
+ if not record_error:
+ if not tb.tb_next:
+ break
+ tb = tb.tb_next
+ continue
+
+ line_number = tb.tb_lineno
+ mod = inspect.getmodule(tb)
+ if mod:
+ lines, offset = inspect.getsourcelines(mod)
+ line = lines[line_number - offset - 1]
+ else:
+ #Useful to catch exceptions with an invalid module (like syntax
+ #errors)
+ lines, offset = inspect.getsourcelines(tb.tb_frame)
+ if (line_number - 1) >= len(lines):
+ line = "Unknown Line"
+ else:
+ line = lines[line_number - 1]
+
+ user_line_number = line_number - num_lines_offset_trace
+ func_name = tb.tb_frame.f_code.co_name
+ error_message += 'File %s, line %s, in %s\n\t%s\n' % \
+ (e_file_name, user_line_number, func_name, line)
+ if not tb.tb_next:
+ break
+ tb = tb.tb_next
+ if name in ['UnicodeEncodeError']:
+ error_message += "\nTo print a unicode string in your udf use encode('utf-8'). Example: \n\tprint 'Example'.encode('utf-8')"
+ if user_line_number:
+ stream_err_output.write("%s\n" % user_line_number)
+ stream_err_output.write("%s\n" % error_message)
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/java/org/apache/tajo/TajoTestingCluster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/TajoTestingCluster.java b/tajo-core/src/test/java/org/apache/tajo/TajoTestingCluster.java
index e680736..54f9e92 100644
--- a/tajo-core/src/test/java/org/apache/tajo/TajoTestingCluster.java
+++ b/tajo-core/src/test/java/org/apache/tajo/TajoTestingCluster.java
@@ -57,6 +57,7 @@ import java.io.File;
import java.io.IOException;
import java.io.Writer;
import java.net.InetSocketAddress;
+import java.net.URISyntaxException;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;
@@ -160,6 +161,8 @@ public class TajoTestingCluster {
// Memory cache termination
conf.setIntVar(ConfVars.WORKER_HISTORY_EXPIRE_PERIOD, 1);
+ conf.setStrings(ConfVars.PYTHON_CODE_DIR.varname, getClass().getResource("/python").toString());
+
/* Since Travi CI limits the size of standard output log up to 4MB */
if (!StringUtils.isEmpty(LOG_LEVEL)) {
Level defaultLevel = Logger.getRootLogger().getLevel();
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java b/tajo-core/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
index e2dac05..36ffd0c 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
@@ -36,7 +36,10 @@ import org.apache.tajo.engine.codegen.TajoClassLoader;
import org.apache.tajo.engine.function.FunctionLoader;
import org.apache.tajo.engine.json.CoreGsonHelper;
import org.apache.tajo.engine.parser.SQLAnalyzer;
+import org.apache.tajo.function.FunctionSignature;
+import org.apache.tajo.master.exec.QueryExecutor;
import org.apache.tajo.plan.*;
+import org.apache.tajo.plan.expr.EvalContext;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.serder.EvalNodeDeserializer;
import org.apache.tajo.plan.serder.EvalNodeSerializer;
@@ -58,6 +61,7 @@ import org.junit.BeforeClass;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
import java.util.TimeZone;
import static org.apache.tajo.TajoConstants.DEFAULT_DATABASE_NAME;
@@ -91,7 +95,9 @@ public class ExprTestBase {
cat = util.getMiniCatalogCluster().getCatalog();
cat.createTablespace(DEFAULT_TABLESPACE_NAME, "hdfs://localhost:1234/warehouse");
cat.createDatabase(DEFAULT_DATABASE_NAME, DEFAULT_TABLESPACE_NAME);
- for (FunctionDesc funcDesc : FunctionLoader.load()) {
+ Map<FunctionSignature, FunctionDesc> map = FunctionLoader.load();
+ map = FunctionLoader.loadUserDefinedFunctions(conf, map);
+ for (FunctionDesc funcDesc : map.values()) {
cat.createFunction(funcDesc);
}
@@ -126,8 +132,8 @@ public class ExprTestBase {
* @return
* @throws PlanningException
*/
- private static Target[] getRawTargets(QueryContext context, String query, boolean condition) throws PlanningException,
- InvalidStatementException {
+ private static Target[] getRawTargets(QueryContext context, String query, boolean condition)
+ throws PlanningException, InvalidStatementException {
List<ParsedResult> parsedResults = SimpleParser.parseScript(query);
if (parsedResults.size() > 1) {
@@ -265,6 +271,7 @@ public class ExprTestBase {
Target [] targets;
TajoClassLoader classLoader = new TajoClassLoader();
+ EvalContext evalContext = new EvalContext();
try {
targets = getRawTargets(queryContext, query, condition);
@@ -274,6 +281,7 @@ public class ExprTestBase {
codegen = new EvalCodeGenerator(classLoader);
}
+ QueryExecutor.startScriptExecutors(queryContext, evalContext, targets);
Tuple outTuple = new VTuple(targets.length);
for (int i = 0; i < targets.length; i++) {
EvalNode eval = targets[i].getEvalTree();
@@ -281,7 +289,7 @@ public class ExprTestBase {
if (queryContext.getBool(SessionVars.CODEGEN)) {
eval = codegen.compile(inputSchema, eval);
}
- eval.bind(inputSchema);
+ eval.bind(evalContext, inputSchema);
outTuple.put(i, eval.eval(vtuple));
}
@@ -318,11 +326,12 @@ public class ExprTestBase {
if (schema != null) {
cat.dropTable(qualifiedTableName);
}
+ QueryExecutor.stopScriptExecutors(evalContext);
}
}
public static void assertEvalTreeProtoSerDer(OverridableConf context, EvalNode evalNode) {
PlanProto.EvalNodeTree converted = EvalNodeSerializer.serialize(evalNode);
- assertEquals(evalNode, EvalNodeDeserializer.deserialize(context, converted));
+ assertEquals(evalNode, EvalNodeDeserializer.deserialize(context, null, converted));
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/java/org/apache/tajo/engine/eval/TestEvalTree.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/eval/TestEvalTree.java b/tajo-core/src/test/java/org/apache/tajo/engine/eval/TestEvalTree.java
index a2d0598..ea42783 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/eval/TestEvalTree.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/eval/TestEvalTree.java
@@ -48,7 +48,7 @@ public class TestEvalTree extends ExprTestBase {
schema1.addColumn("table1.score", INT4);
BinaryEval expr = new BinaryEval(EvalType.PLUS, e1, e2);
- expr.bind(schema1);
+ expr.bind(null, schema1);
assertCloneEqual(expr);
VTuple tuple = new VTuple(2);
@@ -161,19 +161,19 @@ public class TestEvalTree extends ExprTestBase {
MockFalseExpr falseExpr = new MockFalseExpr();
BinaryEval andExpr = new BinaryEval(EvalType.AND, trueExpr, trueExpr);
- andExpr.bind(null);
+ andExpr.bind(null, null);
assertTrue(andExpr.eval(null).asBool());
andExpr = new BinaryEval(EvalType.AND, falseExpr, trueExpr);
- andExpr.bind(null);
+ andExpr.bind(null, null);
assertFalse(andExpr.eval(null).asBool());
andExpr = new BinaryEval(EvalType.AND, trueExpr, falseExpr);
- andExpr.bind(null);
+ andExpr.bind(null, null);
assertFalse(andExpr.eval(null).asBool());
andExpr = new BinaryEval(EvalType.AND, falseExpr, falseExpr);
- andExpr.bind(null);
+ andExpr.bind(null, null);
assertFalse(andExpr.eval(null).asBool());
}
@@ -183,19 +183,19 @@ public class TestEvalTree extends ExprTestBase {
MockFalseExpr falseExpr = new MockFalseExpr();
BinaryEval orExpr = new BinaryEval(EvalType.OR, trueExpr, trueExpr);
- orExpr.bind(null);
+ orExpr.bind(null, null);
assertTrue(orExpr.eval(null).asBool());
orExpr = new BinaryEval(EvalType.OR, falseExpr, trueExpr);
- orExpr.bind(null);
+ orExpr.bind(null, null);
assertTrue(orExpr.eval(null).asBool());
orExpr = new BinaryEval(EvalType.OR, trueExpr, falseExpr);
- orExpr.bind(null);
+ orExpr.bind(null, null);
assertTrue(orExpr.eval(null).asBool());
orExpr = new BinaryEval(EvalType.OR, falseExpr, falseExpr);
- orExpr.bind(null);
+ orExpr.bind(null, null);
assertFalse(orExpr.eval(null).asBool());
}
@@ -209,41 +209,41 @@ public class TestEvalTree extends ExprTestBase {
e1 = new ConstEval(DatumFactory.createInt4(9));
e2 = new ConstEval(DatumFactory.createInt4(34));
expr = new BinaryEval(EvalType.LTH, e1, e2);
- assertTrue(expr.bind(null).eval(null).asBool());
+ assertTrue(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.LEQ, e1, e2);
- assertTrue(expr.bind(null).eval(null).asBool());
+ assertTrue(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.LTH, e2, e1);
- assertFalse(expr.bind(null).eval(null).asBool());
+ assertFalse(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.LEQ, e2, e1);
- assertFalse(expr.bind(null).eval(null).asBool());
+ assertFalse(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.GTH, e2, e1);
- assertTrue(expr.bind(null).eval(null).asBool());
+ assertTrue(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.GEQ, e2, e1);
- assertTrue(expr.bind(null).eval(null).asBool());
+ assertTrue(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.GTH, e1, e2);
- assertFalse(expr.bind(null).eval(null).asBool());
+ assertFalse(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.GEQ, e1, e2);
- assertFalse(expr.bind(null).eval(null).asBool());
+ assertFalse(expr.bind(null, null).eval(null).asBool());
BinaryEval plus = new BinaryEval(EvalType.PLUS, e1, e2);
expr = new BinaryEval(EvalType.LTH, e1, plus);
- assertTrue(expr.bind(null).eval(null).asBool());
+ assertTrue(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.LEQ, e1, plus);
- assertTrue(expr.bind(null).eval(null).asBool());
+ assertTrue(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.LTH, plus, e1);
- assertFalse(expr.bind(null).eval(null).asBool());
+ assertFalse(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.LEQ, plus, e1);
- assertFalse(expr.bind(null).eval(null).asBool());
+ assertFalse(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.GTH, plus, e1);
- assertTrue(expr.bind(null).eval(null).asBool());
+ assertTrue(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.GEQ, plus, e1);
- assertTrue(expr.bind(null).eval(null).asBool());
+ assertTrue(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.GTH, e1, plus);
- assertFalse(expr.bind(null).eval(null).asBool());
+ assertFalse(expr.bind(null, null).eval(null).asBool());
expr = new BinaryEval(EvalType.GEQ, e1, plus);
- assertFalse(expr.bind(null).eval(null).asBool());
+ assertFalse(expr.bind(null, null).eval(null).asBool());
}
@Test
@@ -256,28 +256,28 @@ public class TestEvalTree extends ExprTestBase {
e1 = new ConstEval(DatumFactory.createInt4(9));
e2 = new ConstEval(DatumFactory.createInt4(34));
BinaryEval expr = new BinaryEval(EvalType.PLUS, e1, e2);
- assertEquals(expr.bind(null).eval(null).asInt4(), 43);
+ assertEquals(expr.bind(null, null).eval(null).asInt4(), 43);
assertCloneEqual(expr);
// MINUS
e1 = new ConstEval(DatumFactory.createInt4(5));
e2 = new ConstEval(DatumFactory.createInt4(2));
expr = new BinaryEval(EvalType.MINUS, e1, e2);
- assertEquals(expr.bind(null).eval(null).asInt4(), 3);
+ assertEquals(expr.bind(null, null).eval(null).asInt4(), 3);
assertCloneEqual(expr);
// MULTIPLY
e1 = new ConstEval(DatumFactory.createInt4(5));
e2 = new ConstEval(DatumFactory.createInt4(2));
expr = new BinaryEval(EvalType.MULTIPLY, e1, e2);
- assertEquals(expr.bind(null).eval(null).asInt4(), 10);
+ assertEquals(expr.bind(null, null).eval(null).asInt4(), 10);
assertCloneEqual(expr);
// DIVIDE
e1 = new ConstEval(DatumFactory.createInt4(10));
e2 = new ConstEval(DatumFactory.createInt4(5));
expr = new BinaryEval(EvalType.DIVIDE, e1, e2);
- assertEquals(expr.bind(null).eval(null).asInt4(), 2);
+ assertEquals(expr.bind(null, null).eval(null).asInt4(), 2);
assertCloneEqual(expr);
}
@@ -293,7 +293,7 @@ public class TestEvalTree extends ExprTestBase {
assertEquals(CatalogUtil.newSimpleDataType(INT4), expr.getValueType());
expr = new BinaryEval(EvalType.LTH, e1, e2);
- assertTrue(expr.bind(null).eval(null).asBool());
+ assertTrue(expr.bind(null, null).eval(null).asBool());
assertEquals(CatalogUtil.newSimpleDataType(BOOLEAN), expr.getValueType());
e1 = new ConstEval(DatumFactory.createFloat8(9.3));
@@ -384,7 +384,7 @@ public class TestEvalTree extends ExprTestBase {
binEval.eval(null);
fail("EvalNode is not binded");
} catch (IllegalStateException e) {
- assertTrue(binEval.bind(null).eval(null).asBool());
+ assertTrue(binEval.bind(null, null).eval(null).asBool());
}
CaseWhenEval caseWhenEval = new CaseWhenEval();
@@ -393,7 +393,7 @@ public class TestEvalTree extends ExprTestBase {
caseWhenEval.eval(null);
fail("EvalNode is not binded");
} catch (IllegalStateException e) {
- assertEquals(caseWhenEval.bind(null).eval(null).asInt4(), 1);
+ assertEquals(caseWhenEval.bind(null, null).eval(null).asInt4(), 1);
}
Schema schema = new Schema(new Column[]{new Column("test", TajoDataTypes.Type.INT4)});
@@ -404,7 +404,7 @@ public class TestEvalTree extends ExprTestBase {
regexEval.eval(null);
fail("EvalNode is not binded");
} catch (IllegalStateException e) {
- assertEquals(regexEval.bind(schema).eval(tuple).asBool(), true);
+ assertEquals(regexEval.bind(null, schema).eval(tuple).asBool(), true);
}
RowConstantEval rowConstantEval = new RowConstantEval(new Datum[]{});
@@ -412,7 +412,7 @@ public class TestEvalTree extends ExprTestBase {
rowConstantEval.eval(null);
fail("EvalNode is not binded");
} catch (IllegalStateException e) {
- assertEquals(rowConstantEval.bind(null).eval(null).isNull(), true);
+ assertEquals(rowConstantEval.bind(null, null).eval(null).isNull(), true);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/java/org/apache/tajo/engine/eval/TestEvalTreeUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/eval/TestEvalTreeUtil.java b/tajo-core/src/test/java/org/apache/tajo/engine/eval/TestEvalTreeUtil.java
index e23a34b..0466a24 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/eval/TestEvalTreeUtil.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/eval/TestEvalTreeUtil.java
@@ -265,12 +265,12 @@ public class TestEvalTreeUtil {
FieldEval field = first.getLeftExpr();
assertEquals(col1, field.getColumnRef());
assertEquals(EvalType.LTH, first.getType());
- assertEquals(10, first.getRightExpr().bind(null).eval(null).asInt4());
+ assertEquals(10, first.getRightExpr().bind(null, null).eval(null).asInt4());
field = second.getRightExpr();
assertEquals(col1, field.getColumnRef());
assertEquals(EvalType.LTH, second.getType());
- assertEquals(4, second.getLeftExpr().bind(null).eval(null).asInt4());
+ assertEquals(4, second.getLeftExpr().bind(null, null).eval(null).asInt4());
}
@Test
@@ -304,10 +304,10 @@ public class TestEvalTreeUtil {
Target [] targets = getRawTargets(QUERIES[0]);
EvalNode node = AlgebraicUtil.eliminateConstantExprs(targets[0].getEvalTree());
assertEquals(EvalType.CONST, node.getType());
- assertEquals(7, node.bind(null).eval(null).asInt4());
+ assertEquals(7, node.bind(null, null).eval(null).asInt4());
node = AlgebraicUtil.eliminateConstantExprs(targets[1].getEvalTree());
assertEquals(EvalType.CONST, node.getType());
- assertTrue(7.0d == node.bind(null).eval(null).asFloat8());
+ assertTrue(7.0d == node.bind(null, null).eval(null).asFloat8());
Expr expr = analyzer.parse(QUERIES[1]);
LogicalPlan plan = planner.createPlan(defaultContext, expr, true);
@@ -335,7 +335,7 @@ public class TestEvalTreeUtil {
assertEquals(EvalType.GTH, transposed.getType());
FieldEval field = transposed.getLeftExpr();
assertEquals(col1, field.getColumnRef());
- assertEquals(1, transposed.getRightExpr().bind(null).eval(null).asInt4());
+ assertEquals(1, transposed.getRightExpr().bind(null, null).eval(null).asInt4());
node = getRootSelection(QUERIES[4]);
// we expect that score < 3
@@ -343,7 +343,7 @@ public class TestEvalTreeUtil {
assertEquals(EvalType.LTH, transposed.getType());
field = transposed.getLeftExpr();
assertEquals(col1, field.getColumnRef());
- assertEquals(2, transposed.getRightExpr().bind(null).eval(null).asInt4());
+ assertEquals(2, transposed.getRightExpr().bind(null, null).eval(null).asInt4());
}
@Test
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/java/org/apache/tajo/engine/function/TestMathFunctions.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/function/TestMathFunctions.java b/tajo-core/src/test/java/org/apache/tajo/engine/function/TestMathFunctions.java
index 49ef7e0..78509f7 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/function/TestMathFunctions.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/function/TestMathFunctions.java
@@ -394,9 +394,9 @@ public class TestMathFunctions extends ExprTestBase {
@Test
public void testPow() throws IOException {
- testSimpleEval("select pow(9,3) as col1 ", new String[]{String.valueOf(Math.pow(9,3))});
- testSimpleEval("select pow(1.0,3) as col1 ", new String[]{String.valueOf(Math.pow(1.0,3))});
- testSimpleEval("select pow(20.1,3.1) as col1 ", new String[]{String.valueOf(Math.pow(20.1,3.1))});
+ testSimpleEval("select pow(9,3) as col1 ", new String[]{String.valueOf(Math.pow(9, 3))});
+ testSimpleEval("select pow(1.0,3) as col1 ", new String[]{String.valueOf(Math.pow(1.0, 3))});
+ testSimpleEval("select pow(20.1,3.1) as col1 ", new String[]{String.valueOf(Math.pow(20.1, 3.1))});
testSimpleEval("select pow(null,3.1) as col1 ", new String[]{""});
testSimpleEval("select pow(20.1,null) as col1 ", new String[]{""});
@@ -408,8 +408,8 @@ public class TestMathFunctions extends ExprTestBase {
testEval(schema, "table1", "0.4,2.7,3,2", "select pow(col1, col2), pow(col3, col4) from table1",
new String[]{
- String.valueOf(Math.pow((float)0.4, 2.7)),
- String.valueOf(Math.pow(3,2))
+ String.valueOf(Math.pow((float) 0.4, 2.7)),
+ String.valueOf(Math.pow(3, 2))
});
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/java/org/apache/tajo/engine/function/TestPythonFunctions.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/function/TestPythonFunctions.java b/tajo-core/src/test/java/org/apache/tajo/engine/function/TestPythonFunctions.java
new file mode 100644
index 0000000..47a0ad2
--- /dev/null
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/function/TestPythonFunctions.java
@@ -0,0 +1,44 @@
+/*
+ * 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.tajo.engine.function;
+
+import org.apache.tajo.engine.eval.ExprTestBase;
+import org.junit.Test;
+
+import java.io.IOException;
+
+public class TestPythonFunctions extends ExprTestBase {
+
+ @Test
+ public void testFunctions() throws IOException {
+ testSimpleEval("select return_one()", new String[]{"1"});
+ testSimpleEval("select helloworld()", new String[]{"Hello, World"});
+ testSimpleEval("select concat_py('1')", new String[]{"11"});
+ testSimpleEval("select comma_format(12345)", new String[]{"12,345"});
+ testSimpleEval("select sum_py(1,2)", new String[]{"3"});
+ testSimpleEval("select percent(386, 1000)", new String[]{"38.6"});
+ testSimpleEval("select concat4('Tajo', 'is', 'awesome', '!')", new String[]{"Tajo is awesome !"});
+ }
+
+ @Test
+ public void testNestedFunctions() throws IOException {
+ testSimpleEval("select sum_py(3, return_one())", new String[]{"4"});
+ testSimpleEval("select concat_py(helloworld())", new String[]{"Hello, WorldHello, World"});
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
index d1756e1..15a1c9f 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
@@ -799,4 +799,18 @@ public class TestGroupByQuery extends QueryTestCaseBase {
assertResultSet(res);
cleanupQuery(res);
}
+
+ @Test
+ public final void testGroupbyWithPythonFunc() throws Exception {
+ ResultSet res = executeQuery();
+ assertResultSet(res);
+ cleanupQuery(res);
+ }
+
+ @Test
+ public final void testGroupbyWithPythonFunc2() throws Exception {
+ ResultSet res = executeQuery();
+ assertResultSet(res);
+ cleanupQuery(res);
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
index 513868b..002f05d 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
@@ -710,6 +710,27 @@ public class TestSelectQuery extends QueryTestCaseBase {
}
@Test
+ public void testSelectPythonFuncs() throws Exception {
+ ResultSet res = executeQuery();
+ assertResultSet(res);
+ cleanupQuery(res);
+ }
+
+ @Test
+ public void testSelectWithPredicateOnPythonFunc() throws Exception {
+ ResultSet res = executeQuery();
+ assertResultSet(res);
+ cleanupQuery(res);
+ }
+
+ @Test
+ public void testNestedPythonFunction() throws Exception {
+ ResultSet res = executeQuery();
+ assertResultSet(res);
+ cleanupQuery(res);
+ }
+
+ @Test
public void testSelectWithParentheses1() throws Exception {
ResultSet res = executeQuery();
assertResultSet(res);
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/python/__init__.py
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/python/__init__.py b/tajo-core/src/test/resources/python/__init__.py
new file mode 100644
index 0000000..8093a2f
--- /dev/null
+++ b/tajo-core/src/test/resources/python/__init__.py
@@ -0,0 +1,17 @@
+#!/usr/bin/python
+
+############################################################################
+# 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.
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/python/test_funcs.py
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/python/test_funcs.py b/tajo-core/src/test/resources/python/test_funcs.py
new file mode 100644
index 0000000..d6b7db5
--- /dev/null
+++ b/tajo-core/src/test/resources/python/test_funcs.py
@@ -0,0 +1,33 @@
+############################################################################
+# 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.
+
+from tajo_util import outputType
+
+@outputType('int4')
+def return_one():
+ return 1
+
+@outputType("text")
+def helloworld():
+ return 'Hello, World'
+
+# No decorator - blob
+def concat_py(str):
+ return str+str
+
+@outputType('int4')
+def sum_py(a,b):
+ return a+b
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/python/test_funcs.pyc
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/python/test_funcs.pyc b/tajo-core/src/test/resources/python/test_funcs.pyc
new file mode 100644
index 0000000..cc84dc1
Binary files /dev/null and b/tajo-core/src/test/resources/python/test_funcs.pyc differ
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/python/test_funcs2.py
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/python/test_funcs2.py b/tajo-core/src/test/resources/python/test_funcs2.py
new file mode 100644
index 0000000..e8db7b5
--- /dev/null
+++ b/tajo-core/src/test/resources/python/test_funcs2.py
@@ -0,0 +1,32 @@
+############################################################################
+# 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.
+
+from tajo_util import outputType
+
+#Percent- Percentage
+@outputType("float8")
+def percent(num, total):
+ return num * 100 / float(total)
+
+#commaFormat- format a number with commas, 12345-> 12,345
+@outputType("text")
+def comma_format(num):
+ return '{:,}'.format(num)
+
+#concatMultiple- concat multiple words
+@outputType("text")
+def concat4(word1, word2, word3, word4):
+ return word1 + " " + word2 + " " + word3 + " " + word4
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupbyWithPythonFunc.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupbyWithPythonFunc.sql b/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupbyWithPythonFunc.sql
new file mode 100644
index 0000000..888552a
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupbyWithPythonFunc.sql
@@ -0,0 +1 @@
+select count(*) from nation where sum_py(n_nationkey, 1) > 2 group by n_regionkey
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupbyWithPythonFunc2.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupbyWithPythonFunc2.sql b/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupbyWithPythonFunc2.sql
new file mode 100644
index 0000000..bcfce13
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestGroupByQuery/testGroupbyWithPythonFunc2.sql
@@ -0,0 +1 @@
+select n_regionkey, count(*) as cnt from nation group by n_regionkey having percent(cnt, 25) > 10
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/queries/TestSelectQuery/testNestedPythonFunction.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSelectQuery/testNestedPythonFunction.sql b/tajo-core/src/test/resources/queries/TestSelectQuery/testNestedPythonFunction.sql
new file mode 100644
index 0000000..75b33ae
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSelectQuery/testNestedPythonFunction.sql
@@ -0,0 +1 @@
+select * from nation where sum_py(n_regionkey, return_one()) < 2
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/queries/TestSelectQuery/testSelectPythonFuncs.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSelectQuery/testSelectPythonFuncs.sql b/tajo-core/src/test/resources/queries/TestSelectQuery/testSelectPythonFuncs.sql
new file mode 100644
index 0000000..bcb9806
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSelectQuery/testSelectPythonFuncs.sql
@@ -0,0 +1,2 @@
+select helloworld(), sum_py(n_nationkey, n_regionkey) as sum, concat_py(n_name) as concat
+from nation where n_nationkey < 5
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/queries/TestSelectQuery/testSelectWithPredicateOnPythonFunc.sql
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/queries/TestSelectQuery/testSelectWithPredicateOnPythonFunc.sql b/tajo-core/src/test/resources/queries/TestSelectQuery/testSelectWithPredicateOnPythonFunc.sql
new file mode 100644
index 0000000..d2c5082
--- /dev/null
+++ b/tajo-core/src/test/resources/queries/TestSelectQuery/testSelectWithPredicateOnPythonFunc.sql
@@ -0,0 +1 @@
+select * from nation where sum_py(n_regionkey,1) > 2
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupbyWithPythonFunc.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupbyWithPythonFunc.result b/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupbyWithPythonFunc.result
new file mode 100644
index 0000000..2a5fb8a
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupbyWithPythonFunc.result
@@ -0,0 +1,7 @@
+?count
+-------------------------------
+4
+5
+5
+5
+4
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupbyWithPythonFunc2.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupbyWithPythonFunc2.result b/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupbyWithPythonFunc2.result
new file mode 100644
index 0000000..08561fd
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestGroupByQuery/testGroupbyWithPythonFunc2.result
@@ -0,0 +1,7 @@
+n_regionkey,cnt
+-------------------------------
+0,5
+3,5
+4,5
+2,5
+1,5
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/results/TestSelectQuery/testNestedPythonFunction.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestSelectQuery/testNestedPythonFunction.result b/tajo-core/src/test/resources/results/TestSelectQuery/testNestedPythonFunction.result
new file mode 100644
index 0000000..899f034
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSelectQuery/testNestedPythonFunction.result
@@ -0,0 +1,7 @@
+n_nationkey,n_name,n_regionkey,n_comment
+-------------------------------
+0,ALGERIA,0, haggle. carefully final deposits detect slyly agai
+5,ETHIOPIA,0,ven packages wake quickly. regu
+14,KENYA,0, pending excuses haggle furiously deposits. pending, express pinto beans wake fluffily past t
+15,MOROCCO,0,rns. blithely bold courts among the closely regular packages use furiously bold platelets?
+16,MOZAMBIQUE,0,s. ironic, unusual asymptotes wake blithely r
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/results/TestSelectQuery/testSelectPythonFuncs.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestSelectQuery/testSelectPythonFuncs.result b/tajo-core/src/test/resources/results/TestSelectQuery/testSelectPythonFuncs.result
new file mode 100644
index 0000000..877e9fc
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSelectQuery/testSelectPythonFuncs.result
@@ -0,0 +1,7 @@
+?helloworld,sum,concat
+-------------------------------
+Hello, World,0,ALGERIAALGERIA
+Hello, World,2,ARGENTINAARGENTINA
+Hello, World,3,BRAZILBRAZIL
+Hello, World,4,CANADACANADA
+Hello, World,8,EGYPTEGYPT
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/test/resources/results/TestSelectQuery/testSelectWithPredicateOnPythonFunc.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestSelectQuery/testSelectWithPredicateOnPythonFunc.result b/tajo-core/src/test/resources/results/TestSelectQuery/testSelectWithPredicateOnPythonFunc.result
new file mode 100644
index 0000000..6d533b4
--- /dev/null
+++ b/tajo-core/src/test/resources/results/TestSelectQuery/testSelectWithPredicateOnPythonFunc.result
@@ -0,0 +1,17 @@
+n_nationkey,n_name,n_regionkey,n_comment
+-------------------------------
+4,EGYPT,4,y above the carefully unusual theodolites. final dugouts are quickly across the furiously regular d
+6,FRANCE,3,refully final requests. regular, ironi
+7,GERMANY,3,l platelets. regular accounts x-ray: unusual, regular acco
+8,INDIA,2,ss excuses cajole slyly across the packages. deposits print aroun
+9,INDONESIA,2, slyly express asymptotes. regular deposits haggle slyly. carefully ironic hockey players sleep blithely. carefull
+10,IRAN,4,efully alongside of the slyly final dependencies.
+11,IRAQ,4,nic deposits boost atop the quickly final requests? quickly regula
+12,JAPAN,2,ously. final, express gifts cajole a
+13,JORDAN,4,ic deposits are blithely about the carefully regular pa
+18,CHINA,2,c dependencies. furiously express notornis sleep slyly regular accounts. ideas sleep. depos
+19,ROMANIA,3,ular asymptotes are about the furious multipliers. express dependencies nag above the ironically ironic account
+20,SAUDI ARABIA,4,ts. silent requests haggle. closely express packages sleep across the blithely
+21,VIETNAM,2,hely enticingly express accounts. even, final
+22,RUSSIA,3, requests against the platelets use never according to the quickly regular pint
+23,UNITED KINGDOM,3,eans boost carefully special requests. accounts are. carefull
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-docs/src/main/sphinx/functions.rst
----------------------------------------------------------------------
diff --git a/tajo-docs/src/main/sphinx/functions.rst b/tajo-docs/src/main/sphinx/functions.rst
index 453edf4..8200bfd 100644
--- a/tajo-docs/src/main/sphinx/functions.rst
+++ b/tajo-docs/src/main/sphinx/functions.rst
@@ -2,6 +2,12 @@
Functions
******************
+Tajo provides extensive supports for functions. It includes a lot of built-in functions and user-defined functions which is implemented in Python.
+
+===================
+Built-in Functions
+===================
+
.. toctree::
:maxdepth: 1
@@ -9,4 +15,66 @@ Functions
functions/string_func_and_operators
functions/datetime_func_and_operators
functions/network_func_and_operators
- functions/json_func
\ No newline at end of file
+ functions/json_func
+
+==============================
+Python User-defined Functions
+==============================
+
+-----------------------
+Function registration
+-----------------------
+
+To register Python UDFs, you must install script files in all cluster nodes.
+After that, you can register your functions by specifying the paths to those script files in ``tajo-site.xml``. Here is an example of the configuration.
+
+.. code-block:: xml
+
+ <property>
+ <name>tajo.function.python.code-dir</name>
+ <value>/path/to/script1.py,/path/to/script2.py</value>
+ </property>
+
+Please note that you can specify multiple paths with ``','`` as a delimiter. Each file can contain multiple functions. Here is a typical example of a script file.
+
+.. code-block:: python
+
+ # /path/to/script1.py
+
+ @outputType('int4')
+ def return_one():
+ return 1
+
+ @outputType("text")
+ def helloworld():
+ return 'Hello, World'
+
+ # No decorator - blob
+ def concat_py(str):
+ return str+str
+
+ @outputType('int4')
+ def sum_py(a,b):
+ return a+b
+
+If the configuration is set properly, every function in the script files are registered when the Tajo cluster starts up.
+
+-----------------------
+Decorators and types
+-----------------------
+
+By default, every function has a return type of ``BLOB``.
+You can use Python decorators to define output types for the script functions. Tajo can figure out return types from the annotations of the Python script.
+
+* ``outputType``: Defines the return data type for a script UDF in a format that Tajo can understand. The defined type must be one of the types supported by Tajo. For supported types, please refer to :doc:`/sql_language/data_model`.
+
+-----------------------
+Query example
+-----------------------
+
+Once the Python UDFs are successfully registered, you can use them as other built-in functions.
+
+.. code-block:: sql
+
+ default> select concat_py(n_name)::text from nation where sum_py(n_regionkey,1) > 2;
+
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-docs/src/main/sphinx/functions/json_func.rst
----------------------------------------------------------------------
diff --git a/tajo-docs/src/main/sphinx/functions/json_func.rst b/tajo-docs/src/main/sphinx/functions/json_func.rst
index 5bf5814..d05cc1b 100644
--- a/tajo-docs/src/main/sphinx/functions/json_func.rst
+++ b/tajo-docs/src/main/sphinx/functions/json_func.rst
@@ -3,6 +3,7 @@ JSON Functions
*******************************
.. function:: json_extract_path_text (string json, string xpath)
+
Extracts JSON string from a JSON string based on json path specified and returns JSON string pointed to by xPath
:param string:
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/ExprAnnotator.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/ExprAnnotator.java b/tajo-plan/src/main/java/org/apache/tajo/plan/ExprAnnotator.java
index 42a7380..0c5a012 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/ExprAnnotator.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/ExprAnnotator.java
@@ -19,6 +19,7 @@
package org.apache.tajo.plan;
import com.google.common.collect.Sets;
+import org.apache.commons.collections.set.UnmodifiableSet;
import org.apache.tajo.OverridableConf;
import org.apache.tajo.SessionVars;
import org.apache.tajo.algebra.*;
@@ -34,7 +35,6 @@ import org.apache.tajo.exception.InvalidOperationException;
import org.apache.tajo.plan.algebra.BaseAlgebraVisitor;
import org.apache.tajo.plan.expr.*;
import org.apache.tajo.plan.function.AggFunction;
-import org.apache.tajo.plan.function.GeneralFunction;
import org.apache.tajo.plan.logical.NodeType;
import org.apache.tajo.plan.nameresolver.NameResolver;
import org.apache.tajo.plan.nameresolver.NameResolvingMode;
@@ -43,6 +43,7 @@ import org.apache.tajo.util.TUtil;
import org.apache.tajo.util.datetime.DateTimeUtil;
import org.apache.tajo.util.datetime.TimeMeta;
+import java.io.IOException;
import java.util.Set;
import java.util.Stack;
import java.util.TimeZone;
@@ -384,7 +385,7 @@ public class ExprAnnotator extends BaseAlgebraVisitor<ExprAnnotator.Context, Eva
if (!EvalTreeUtil.checkIfCanBeConstant(evalNodes[i])) {
throw new PlanningException("Non constant values cannot be included in IN PREDICATE.");
}
- values[i] = EvalTreeUtil.evaluateImmediately(evalNodes[i]);
+ values[i] = EvalTreeUtil.evaluateImmediately(null, evalNodes[i]);
}
return new RowConstantEval(values);
}
@@ -608,7 +609,7 @@ public class ExprAnnotator extends BaseAlgebraVisitor<ExprAnnotator.Context, Eva
FunctionType functionType = funcDesc.getFuncType();
if (functionType == FunctionType.GENERAL
|| functionType == FunctionType.UDF) {
- return new GeneralFunctionEval(ctx.queryContext, funcDesc, (GeneralFunction) funcDesc.newInstance(), givenArgs);
+ return new GeneralFunctionEval(ctx.queryContext, funcDesc, givenArgs);
} else if (functionType == FunctionType.AGGREGATION
|| functionType == FunctionType.UDA) {
if (!ctx.currentBlock.hasNode(NodeType.GROUP_BY)) {
@@ -623,6 +624,8 @@ public class ExprAnnotator extends BaseAlgebraVisitor<ExprAnnotator.Context, Eva
}
} catch (InternalException e) {
throw new PlanningException(e);
+ } catch (IOException e) {
+ throw new PlanningException(e);
}
}
@@ -679,7 +682,8 @@ public class ExprAnnotator extends BaseAlgebraVisitor<ExprAnnotator.Context, Eva
}
public static final Set<String> WINDOW_FUNCTIONS =
- Sets.newHashSet("row_number", "rank", "dense_rank", "percent_rank", "cume_dist", "first_value", "lag");
+ UnmodifiableSet.decorate(
+ Sets.newHashSet("row_number", "rank", "dense_rank", "percent_rank", "cume_dist", "first_value", "lag"));
public EvalNode visitWindowFunction(Context ctx, Stack<Expr> stack, WindowFunctionExpr windowFunc)
throws PlanningException {
@@ -904,9 +908,9 @@ public class ExprAnnotator extends BaseAlgebraVisitor<ExprAnnotator.Context, Eva
public static int [] dateToIntArray(String years, String months, String days)
throws PlanningException {
- int year = Integer.valueOf(years);
- int month = Integer.valueOf(months);
- int day = Integer.valueOf(days);
+ int year = Integer.parseInt(years);
+ int month = Integer.parseInt(months);
+ int day = Integer.parseInt(days);
if (!(1 <= year && year <= 9999)) {
throw new PlanningException(String.format("Years (%d) must be between 1 and 9999 integer value", year));
@@ -930,12 +934,12 @@ public class ExprAnnotator extends BaseAlgebraVisitor<ExprAnnotator.Context, Eva
public static int [] timeToIntArray(String hours, String minutes, String seconds, String fractionOfSecond)
throws PlanningException {
- int hour = Integer.valueOf(hours);
- int minute = Integer.valueOf(minutes);
- int second = Integer.valueOf(seconds);
+ int hour = Integer.parseInt(hours);
+ int minute = Integer.parseInt(minutes);
+ int second = Integer.parseInt(seconds);
int fraction = 0;
if (fractionOfSecond != null) {
- fraction = Integer.valueOf(fractionOfSecond);
+ fraction = Integer.parseInt(fractionOfSecond);
}
if (!(0 <= hour && hour <= 23)) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java
index d7c631b..d1c1a15 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java
@@ -115,6 +115,10 @@ public class LogicalPlanner extends BaseAlgebraVisitor<LogicalPlanner.PlanContex
return queryBlock;
}
+ public OverridableConf getQueryContext() {
+ return queryContext;
+ }
+
public String toString() {
return "block=" + queryBlock.getName() + ", relNum=" + queryBlock.getRelations().size() + ", "+
queryBlock.namedExprsMgr.toString();
@@ -132,7 +136,8 @@ public class LogicalPlanner extends BaseAlgebraVisitor<LogicalPlanner.PlanContex
}
@VisibleForTesting
- public LogicalPlan createPlan(OverridableConf queryContext, Expr expr, boolean debug) throws PlanningException {
+ public LogicalPlan createPlan(OverridableConf queryContext, Expr expr, boolean debug)
+ throws PlanningException {
LogicalPlan plan = new LogicalPlan(this);
@@ -810,7 +815,7 @@ public class LogicalPlanner extends BaseAlgebraVisitor<LogicalPlanner.PlanContex
limitNode.setInSchema(child.getOutSchema());
limitNode.setOutSchema(child.getOutSchema());
- firstFetNum.bind(null);
+ firstFetNum.bind(null, null);
limitNode.setFetchFirst(firstFetNum.eval(null).asInt8());
return limitNode;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/expr/AlgebraicUtil.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/AlgebraicUtil.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/AlgebraicUtil.java
index fb05f33..6cf7272 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/AlgebraicUtil.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/AlgebraicUtil.java
@@ -149,7 +149,7 @@ public class AlgebraicUtil {
}
if (lhs.getType() == EvalType.CONST && rhs.getType() == EvalType.CONST) {
- return new ConstEval(binaryEval.bind(null).eval(null));
+ return new ConstEval(binaryEval.bind(null, null).eval(null));
}
return binaryEval;
@@ -162,7 +162,7 @@ public class AlgebraicUtil {
stack.pop();
if (child.getType() == EvalType.CONST) {
- return new ConstEval(unaryEval.bind(null).eval(null));
+ return new ConstEval(unaryEval.bind(null, null).eval(null));
}
return unaryEval;
@@ -184,7 +184,7 @@ public class AlgebraicUtil {
}
if (constantOfAllDescendents && evalNode.getType() == EvalType.FUNCTION) {
- return new ConstEval(evalNode.bind(null).eval(null));
+ return new ConstEval(evalNode.bind(null, null).eval(null));
} else {
return evalNode;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/expr/BetweenPredicateEval.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/BetweenPredicateEval.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/BetweenPredicateEval.java
index 2db4368..eb3e36b 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/BetweenPredicateEval.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/BetweenPredicateEval.java
@@ -80,7 +80,7 @@ public class BetweenPredicateEval extends EvalNode implements Cloneable {
}
private static interface Checker {
- void bind(Schema schema);
+ void bind(EvalContext evalContext, Schema schema);
Datum eval(Tuple param);
}
@@ -103,8 +103,8 @@ public class BetweenPredicateEval extends EvalNode implements Cloneable {
}
@Override
- public void bind(Schema schema) {
- predicand.bind(schema);
+ public void bind(EvalContext evalContext, Schema schema) {
+ predicand.bind(evalContext, schema);
}
@Override
@@ -134,10 +134,10 @@ public class BetweenPredicateEval extends EvalNode implements Cloneable {
}
@Override
- public void bind(Schema schema) {
- predicand.bind(schema);
- begin.bind(schema);
- end.bind(schema);
+ public void bind(EvalContext evalContext, Schema schema) {
+ predicand.bind(evalContext, schema);
+ begin.bind(evalContext, schema);
+ end.bind(evalContext, schema);
}
@Override
@@ -170,10 +170,10 @@ public class BetweenPredicateEval extends EvalNode implements Cloneable {
}
@Override
- public void bind(Schema schema) {
- predicand.bind(schema);
- begin.bind(schema);
- end.bind(schema);
+ public void bind(EvalContext evalContext, Schema schema) {
+ predicand.bind(evalContext, schema);
+ begin.bind(evalContext, schema);
+ end.bind(evalContext, schema);
}
@Override
@@ -227,8 +227,8 @@ public class BetweenPredicateEval extends EvalNode implements Cloneable {
}
@Override
- public EvalNode bind(Schema schema) {
- super.bind(schema);
+ public EvalNode bind(EvalContext evalContext, Schema schema) {
+ super.bind(evalContext, schema);
if (begin.getType() == EvalType.CONST && end.getType() == EvalType.CONST) {
Datum beginValue = ((ConstEval)begin).getValue();
Datum endValue = ((ConstEval)end).getValue();
@@ -245,7 +245,7 @@ public class BetweenPredicateEval extends EvalNode implements Cloneable {
checker = new AsymmetricChecker(not, predicand, begin, end);
}
}
- checker.bind(schema);
+ checker.bind(evalContext, schema);
return this;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CastEval.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CastEval.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CastEval.java
index f174b79..b8b768b 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CastEval.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CastEval.java
@@ -31,7 +31,7 @@ import java.util.TimeZone;
import static org.apache.tajo.common.TajoDataTypes.DataType;
-public class CastEval extends UnaryEval {
+public class CastEval extends UnaryEval implements Cloneable {
@Expose private DataType target;
@Expose private TimeZone timezone;
@@ -97,11 +97,24 @@ public class CastEval extends UnaryEval {
boolean valid = obj != null && obj instanceof CastEval;
if (valid) {
CastEval another = (CastEval) obj;
- return child.equals(another.child) &&
- target.equals(another.target) &&
- TUtil.checkEquals(timezone, another.timezone);
+ boolean b1 = child.equals(another.child);
+ boolean b2 = target.equals(another.target);
+ boolean b3 = TUtil.checkEquals(timezone, another.timezone);
+ return b1 && b2 && b3;
} else {
return false;
}
}
+
+ @Override
+ public Object clone() throws CloneNotSupportedException {
+ CastEval clone = (CastEval) super.clone();
+ if (target != null) {
+ clone.target = target;
+ }
+ if (timezone != null) {
+ clone.timezone = timezone;
+ }
+ return clone;
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalContext.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalContext.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalContext.java
new file mode 100644
index 0000000..6a30e77
--- /dev/null
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalContext.java
@@ -0,0 +1,45 @@
+/**
+ * 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.tajo.plan.expr;
+
+import org.apache.tajo.plan.function.python.TajoScriptEngine;
+import org.apache.tajo.util.TUtil;
+
+import java.util.Collection;
+import java.util.Map;
+
+public class EvalContext {
+ private final Map<EvalNode, TajoScriptEngine> scriptEngineMap = TUtil.newHashMap();
+
+ public void addScriptEngine(EvalNode evalNode, TajoScriptEngine scriptExecutor) {
+ this.scriptEngineMap.put(evalNode, scriptExecutor);
+ }
+
+ public boolean hasScriptEngine(EvalNode evalNode) {
+ return this.scriptEngineMap.containsKey(evalNode);
+ }
+
+ public TajoScriptEngine getScriptEngine(EvalNode evalNode) {
+ return this.scriptEngineMap.get(evalNode);
+ }
+
+ public Collection<TajoScriptEngine> getAllScriptEngines() {
+ return this.scriptEngineMap.values();
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalNode.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalNode.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalNode.java
index 154b0fd..9abb0bc 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalNode.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalNode.java
@@ -19,6 +19,7 @@
package org.apache.tajo.plan.expr;
import com.google.gson.annotations.Expose;
+import org.apache.tajo.annotation.Nullable;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.common.ProtoObject;
import org.apache.tajo.common.TajoDataTypes.DataType;
@@ -61,9 +62,9 @@ public abstract class EvalNode implements Cloneable, GsonObject, ProtoObject<Pla
return PlanGsonHelper.toJson(this, EvalNode.class);
}
- public EvalNode bind(Schema schema) {
+ public EvalNode bind(@Nullable EvalContext evalContext, Schema schema) {
for (int i = 0; i < childNum(); i++) {
- getChild(i).bind(schema);
+ getChild(i).bind(evalContext, schema);
}
isBinded = true;
return this;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalTreeUtil.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalTreeUtil.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalTreeUtil.java
index 5e3843c..1fa2fe0 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalTreeUtil.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalTreeUtil.java
@@ -570,8 +570,8 @@ public class EvalTreeUtil {
return findUniqueColumns(evalNode).size() == 0 && findDistinctAggFunction(evalNode).size() == 0;
}
- public static Datum evaluateImmediately(EvalNode evalNode) {
- evalNode.bind(null);
+ public static Datum evaluateImmediately(EvalContext evalContext, EvalNode evalNode) {
+ evalNode.bind(evalContext, null);
return evalNode.eval(null);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/expr/FieldEval.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/FieldEval.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/FieldEval.java
index 83d00b9..870970b 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/FieldEval.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/FieldEval.java
@@ -40,8 +40,8 @@ public class FieldEval extends EvalNode implements Cloneable {
}
@Override
- public EvalNode bind(Schema schema) {
- super.bind(schema);
+ public EvalNode bind(EvalContext evalContext, Schema schema) {
+ super.bind(evalContext, schema);
// TODO - column namespace should be improved to simplify name handling and resolving.
if (column.hasQualifier()) {
fieldId = schema.getColumnId(column.getQualifiedName());
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/expr/FunctionEval.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/FunctionEval.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/FunctionEval.java
index 4ff7548..dd9121b 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/FunctionEval.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/FunctionEval.java
@@ -49,8 +49,8 @@ public abstract class FunctionEval extends EvalNode implements Cloneable {
}
@Override
- public EvalNode bind(Schema schema) {
- super.bind(schema);
+ public EvalNode bind(EvalContext evalContext, Schema schema) {
+ super.bind(evalContext, schema);
this.params = new VTuple(argEvals.length);
return this;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/expr/GeneralFunctionEval.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/GeneralFunctionEval.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/GeneralFunctionEval.java
index 2ce7850..30fbe91 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/GeneralFunctionEval.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/GeneralFunctionEval.java
@@ -18,54 +18,56 @@
package org.apache.tajo.plan.expr;
-import com.google.common.base.Objects;
import com.google.gson.annotations.Expose;
import org.apache.tajo.OverridableConf;
import org.apache.tajo.catalog.FunctionDesc;
+import org.apache.tajo.catalog.Schema;
import org.apache.tajo.datum.Datum;
-import org.apache.tajo.plan.function.GeneralFunction;
+import org.apache.tajo.plan.function.FunctionInvoke;
+import org.apache.tajo.plan.function.FunctionInvokeContext;
import org.apache.tajo.storage.Tuple;
-import org.apache.tajo.util.TUtil;
-import javax.annotation.Nullable;
+import java.io.IOException;
public class GeneralFunctionEval extends FunctionEval {
- @Expose protected GeneralFunction instance;
+ protected FunctionInvoke funcInvoke;
+ @Expose protected FunctionInvokeContext invokeContext;
- public GeneralFunctionEval(@Nullable OverridableConf queryContext, FunctionDesc desc, GeneralFunction instance,
- EvalNode[] givenArgs) {
+ public GeneralFunctionEval(OverridableConf queryContext, FunctionDesc desc, EvalNode[] givenArgs)
+ throws IOException {
super(EvalType.FUNCTION, desc, givenArgs);
- this.instance = instance;
- this.instance.init(queryContext, getParamType());
+ this.invokeContext = new FunctionInvokeContext(queryContext, getParamType());
+ }
+
+ @Override
+ public EvalNode bind(EvalContext evalContext, Schema schema) {
+ super.bind(evalContext, schema);
+ try {
+ this.funcInvoke = FunctionInvoke.newInstance(funcDesc);
+ if (evalContext != null && evalContext.hasScriptEngine(this)) {
+ this.invokeContext.setScriptEngine(evalContext.getScriptEngine(this));
+ }
+ this.funcInvoke.init(invokeContext);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return this;
}
@Override
@SuppressWarnings("unchecked")
public Datum eval(Tuple tuple) {
super.eval(tuple);
- return instance.eval(evalParams(tuple));
+ Datum res = funcInvoke.eval(evalParams(tuple));
+ return res;
}
@Override
- public boolean equals(Object obj) {
- if (obj instanceof GeneralFunctionEval) {
- GeneralFunctionEval other = (GeneralFunctionEval) obj;
- return super.equals(other) &&
- TUtil.checkEquals(instance, other.instance);
- }
-
- return false;
- }
-
- @Override
- public int hashCode() {
- return Objects.hashCode(funcDesc, instance);
- }
-
- @Override
public Object clone() throws CloneNotSupportedException {
GeneralFunctionEval eval = (GeneralFunctionEval) super.clone();
- eval.instance = (GeneralFunction) instance.clone();
+ if (funcInvoke != null) {
+ eval.funcInvoke = (FunctionInvoke) funcInvoke.clone();
+ }
return eval;
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/expr/PatternMatchPredicateEval.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/PatternMatchPredicateEval.java b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/PatternMatchPredicateEval.java
index 6faa667..cdd8dfb 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/PatternMatchPredicateEval.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/PatternMatchPredicateEval.java
@@ -74,8 +74,8 @@ public abstract class PatternMatchPredicateEval extends BinaryEval {
}
@Override
- public EvalNode bind(Schema schema) {
- super.bind(schema);
+ public EvalNode bind(EvalContext evalContext, Schema schema) {
+ super.bind(evalContext, schema);
compile(pattern);
return this;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/EvalTreeOptimizationRule.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/EvalTreeOptimizationRule.java b/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/EvalTreeOptimizationRule.java
index 7e03224..2440c52 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/EvalTreeOptimizationRule.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/EvalTreeOptimizationRule.java
@@ -20,9 +20,10 @@ package org.apache.tajo.plan.exprrewrite;
import org.apache.tajo.plan.LogicalPlanner;
import org.apache.tajo.plan.annotator.Prioritized;
+import org.apache.tajo.plan.expr.EvalContext;
import org.apache.tajo.plan.expr.EvalNode;
@Prioritized
public interface EvalTreeOptimizationRule {
- public EvalNode optimize(LogicalPlanner.PlanContext context, EvalNode tree);
+ EvalNode optimize(LogicalPlanner.PlanContext context, EvalNode tree);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/EvalTreeOptimizer.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/EvalTreeOptimizer.java b/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/EvalTreeOptimizer.java
index a8a3ff3..680600a 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/EvalTreeOptimizer.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/EvalTreeOptimizer.java
@@ -24,6 +24,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.tajo.plan.LogicalPlanner;
import org.apache.tajo.plan.annotator.Prioritized;
+import org.apache.tajo.plan.expr.EvalContext;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.util.ClassUtil;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/rules/ConstantFolding.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/rules/ConstantFolding.java b/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/rules/ConstantFolding.java
index d77f072..a8b0945 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/rules/ConstantFolding.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/exprrewrite/rules/ConstantFolding.java
@@ -18,17 +18,23 @@
package org.apache.tajo.plan.exprrewrite.rules;
+import org.apache.tajo.datum.Datum;
import org.apache.tajo.plan.expr.*;
import org.apache.tajo.plan.exprrewrite.EvalTreeOptimizationRule;
import org.apache.tajo.plan.annotator.Prioritized;
import org.apache.tajo.plan.LogicalPlanner;
+import org.apache.tajo.plan.function.python.PythonScriptEngine;
+import org.apache.tajo.plan.function.python.TajoScriptEngine;
+import java.io.IOException;
import java.util.Stack;
@Prioritized(priority = 10)
public class ConstantFolding extends SimpleEvalNodeVisitor<LogicalPlanner.PlanContext>
implements EvalTreeOptimizationRule {
+ private final static String SLEEP_FUNCTION_NAME = "sleep";
+
@Override
public EvalNode optimize(LogicalPlanner.PlanContext context, EvalNode evalNode) {
return visit(context, evalNode, new Stack<EvalNode>());
@@ -49,7 +55,7 @@ public class ConstantFolding extends SimpleEvalNodeVisitor<LogicalPlanner.PlanCo
}
if (lhs.getType() == EvalType.CONST && rhs.getType() == EvalType.CONST) {
- binaryEval.bind(null);
+ binaryEval.bind(null, null);
return new ConstEval(binaryEval.eval(null));
}
@@ -62,8 +68,9 @@ public class ConstantFolding extends SimpleEvalNodeVisitor<LogicalPlanner.PlanCo
EvalNode child = visit(context, unaryEval.getChild(), stack);
stack.pop();
+ unaryEval.setChild(child);
if (child.getType() == EvalType.CONST) {
- unaryEval.bind(null);
+ unaryEval.bind(null, null);
return new ConstEval(unaryEval.eval(null));
}
@@ -74,7 +81,7 @@ public class ConstantFolding extends SimpleEvalNodeVisitor<LogicalPlanner.PlanCo
public EvalNode visitFuncCall(LogicalPlanner.PlanContext context, FunctionEval evalNode, Stack<EvalNode> stack) {
boolean constantOfAllDescendents = true;
- if ("sleep".equals(evalNode.getFuncDesc().getFunctionName())) {
+ if (SLEEP_FUNCTION_NAME.equals(evalNode.getFuncDesc().getFunctionName())) {
constantOfAllDescendents = false;
} else {
for (EvalNode arg : evalNode.getArgs()) {
@@ -84,8 +91,24 @@ public class ConstantFolding extends SimpleEvalNodeVisitor<LogicalPlanner.PlanCo
}
if (constantOfAllDescendents && evalNode.getType() == EvalType.FUNCTION) {
- evalNode.bind(null);
- return new ConstEval(evalNode.eval(null));
+ if (evalNode.getFuncDesc().getInvocation().hasPython()) {
+ TajoScriptEngine executor = new PythonScriptEngine(evalNode.getFuncDesc());
+ try {
+ executor.start(context.getQueryContext().getConf());
+ EvalContext evalContext = new EvalContext();
+ evalContext.addScriptEngine(evalNode, executor);
+ evalNode.bind(evalContext, null);
+ Datum funcRes = evalNode.eval(null);
+ executor.shutdown();
+ return new ConstEval(funcRes);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ evalNode.bind(null, null);
+ return new ConstEval(evalNode.eval(null));
+ }
+
} else {
return evalNode;
}
[4/4] tajo git commit: TAJO-1344: Python UDF support. (jihoon)
Posted by ji...@apache.org.
TAJO-1344: Python UDF support. (jihoon)
Closes #526
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/a7453853
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/a7453853
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/a7453853
Branch: refs/heads/master
Commit: a7453853975dc8b5f4162ae16a1669aeb7ed995c
Parents: 14a1e53
Author: Jihoon Son <ji...@apache.org>
Authored: Sat Apr 18 11:56:47 2015 +0900
Committer: Jihoon Son <ji...@apache.org>
Committed: Sat Apr 18 11:57:24 2015 +0900
----------------------------------------------------------------------
CHANGES | 8 +-
.../org/apache/tajo/catalog/CatalogUtil.java | 3 +-
.../org/apache/tajo/catalog/FunctionDesc.java | 4 +-
.../tajo/function/FunctionInvocation.java | 24 +-
.../org/apache/tajo/function/FunctionUtil.java | 10 +
.../tajo/function/PythonInvocationDesc.java | 98 +++++
.../src/main/proto/CatalogProtos.proto | 6 +
.../apache/tajo/catalog/TestFunctionDesc.java | 4 +-
.../tajo/catalog/store/AbstractDBStore.java | 5 +-
.../org/apache/tajo/catalog/store/MemStore.java | 1 -
.../org/apache/tajo/catalog/TestCatalog.java | 4 +-
.../main/java/org/apache/tajo/QueryVars.java | 2 +-
.../java/org/apache/tajo/conf/TajoConf.java | 4 +
.../java/org/apache/tajo/datum/AnyDatum.java | 82 +++++
.../java/org/apache/tajo/datum/BlobDatum.java | 5 +-
.../org/apache/tajo/datum/DatumFactory.java | 21 +-
.../java/org/apache/tajo/json/DatumAdapter.java | 6 +-
.../java/org/apache/tajo/storage/Tuple.java | 54 +--
.../java/org/apache/tajo/util/FileUtil.java | 23 ++
.../java/org/apache/tajo/util/KeyValueSet.java | 2 +-
.../tajo/util/datetime/DateTimeFormat.java | 2 -
.../tajo/engine/codegen/EvalCodeGenContext.java | 6 +-
.../codegen/LegacyFunctionBindingEmitter.java | 4 +-
.../engine/codegen/VariablesPreBuilder.java | 2 +-
.../tajo/engine/function/FunctionLoader.java | 59 ++-
.../apache/tajo/engine/planner/Projector.java | 2 +-
.../rules/GlobalPlanEqualityTester.java | 2 +-
.../planner/physical/AggregationExec.java | 2 +-
.../planner/physical/BSTIndexScanExec.java | 2 +-
.../engine/planner/physical/CommonJoinExec.java | 2 +-
.../DistinctGroupbyFirstAggregationExec.java | 2 +-
.../DistinctGroupbyHashAggregationExec.java | 2 +-
.../DistinctGroupbySecondAggregationExec.java | 2 +-
.../DistinctGroupbyThirdAggregationExec.java | 2 +-
.../engine/planner/physical/EvalExprExec.java | 2 +-
.../planner/physical/HashLeftOuterJoinExec.java | 4 +-
.../engine/planner/physical/HavingExec.java | 2 +-
.../engine/planner/physical/SelectionExec.java | 2 +-
.../engine/planner/physical/SeqScanExec.java | 4 +-
.../engine/planner/physical/WindowAggExec.java | 2 +-
.../apache/tajo/engine/query/QueryContext.java | 2 +-
.../org/apache/tajo/master/GlobalEngine.java | 2 -
.../java/org/apache/tajo/master/TajoMaster.java | 16 +-
.../apache/tajo/master/exec/QueryExecutor.java | 81 ++--
.../java/org/apache/tajo/worker/TajoWorker.java | 11 +-
.../main/java/org/apache/tajo/worker/Task.java | 35 +-
.../apache/tajo/worker/TaskAttemptContext.java | 7 +
tajo-core/src/main/resources/python/__init__.py | 17 +
.../src/main/resources/python/controller.py | 330 +++++++++++++++++
.../src/main/resources/python/tajo_util.py | 103 ++++++
.../org/apache/tajo/TajoTestingCluster.java | 3 +
.../apache/tajo/engine/eval/ExprTestBase.java | 19 +-
.../apache/tajo/engine/eval/TestEvalTree.java | 68 ++--
.../tajo/engine/eval/TestEvalTreeUtil.java | 12 +-
.../tajo/engine/function/TestMathFunctions.java | 10 +-
.../engine/function/TestPythonFunctions.java | 44 +++
.../tajo/engine/query/TestGroupByQuery.java | 14 +
.../tajo/engine/query/TestSelectQuery.java | 21 ++
tajo-core/src/test/resources/python/__init__.py | 17 +
.../src/test/resources/python/test_funcs.py | 33 ++
.../src/test/resources/python/test_funcs.pyc | Bin 0 -> 1042 bytes
.../src/test/resources/python/test_funcs2.py | 32 ++
.../testGroupbyWithPythonFunc.sql | 1 +
.../testGroupbyWithPythonFunc2.sql | 1 +
.../testNestedPythonFunction.sql | 1 +
.../TestSelectQuery/testSelectPythonFuncs.sql | 2 +
.../testSelectWithPredicateOnPythonFunc.sql | 1 +
.../testGroupbyWithPythonFunc.result | 7 +
.../testGroupbyWithPythonFunc2.result | 7 +
.../testNestedPythonFunction.result | 7 +
.../testSelectPythonFuncs.result | 7 +
.../testSelectWithPredicateOnPythonFunc.result | 17 +
tajo-docs/src/main/sphinx/functions.rst | 70 +++-
.../src/main/sphinx/functions/json_func.rst | 1 +
.../org/apache/tajo/plan/ExprAnnotator.java | 26 +-
.../org/apache/tajo/plan/LogicalPlanner.java | 9 +-
.../apache/tajo/plan/expr/AlgebraicUtil.java | 6 +-
.../tajo/plan/expr/BetweenPredicateEval.java | 28 +-
.../org/apache/tajo/plan/expr/CastEval.java | 21 +-
.../org/apache/tajo/plan/expr/EvalContext.java | 45 +++
.../org/apache/tajo/plan/expr/EvalNode.java | 5 +-
.../org/apache/tajo/plan/expr/EvalTreeUtil.java | 4 +-
.../org/apache/tajo/plan/expr/FieldEval.java | 4 +-
.../org/apache/tajo/plan/expr/FunctionEval.java | 4 +-
.../tajo/plan/expr/GeneralFunctionEval.java | 56 +--
.../plan/expr/PatternMatchPredicateEval.java | 4 +-
.../exprrewrite/EvalTreeOptimizationRule.java | 3 +-
.../plan/exprrewrite/EvalTreeOptimizer.java | 1 +
.../plan/exprrewrite/rules/ConstantFolding.java | 33 +-
.../tajo/plan/function/FunctionInvoke.java | 90 +++++
.../plan/function/FunctionInvokeContext.java | 74 ++++
.../function/LegacyScalarFunctionInvoke.java | 81 ++++
.../plan/function/PythonFunctionInvoke.java | 59 +++
.../function/python/PythonScriptEngine.java | 368 +++++++++++++++++++
.../plan/function/python/TajoScriptEngine.java | 83 +++++
.../tajo/plan/function/stream/BufferPool.java | 74 ++++
.../function/stream/ByteBufInputChannel.java | 71 ++++
.../plan/function/stream/ByteBufLineReader.java | 176 +++++++++
.../function/stream/CSVLineDeserializer.java | 99 +++++
.../tajo/plan/function/stream/CSVLineSerDe.java | 42 +++
.../plan/function/stream/CSVLineSerializer.java | 118 ++++++
.../stream/FieldSerializerDeserializer.java | 36 ++
.../function/stream/FieldSplitProcessor.java | 34 ++
.../tajo/plan/function/stream/InputHandler.java | 78 ++++
.../function/stream/LineSplitProcessor.java | 45 +++
.../plan/function/stream/OutputHandler.java | 156 ++++++++
.../plan/function/stream/StreamingUtil.java | 91 +++++
.../stream/TextFieldSerializerDeserializer.java | 257 +++++++++++++
.../function/stream/TextLineDeserializer.java | 60 +++
.../function/stream/TextLineParsingError.java | 31 ++
.../plan/function/stream/TextLineSerDe.java | 65 ++++
.../function/stream/TextLineSerializer.java | 45 +++
.../rules/LogicalPlanEqualityTester.java | 2 +-
.../rewrite/rules/PartitionedTableRewriter.java | 2 +-
.../tajo/plan/serder/EvalNodeDeserializer.java | 16 +-
.../tajo/plan/serder/EvalNodeSerializer.java | 10 +-
.../plan/serder/LogicalNodeDeserializer.java | 137 +++----
tajo-plan/src/main/proto/Plan.proto | 28 +-
.../org/apache/tajo/storage/TestLazyTuple.java | 2 +-
.../testErrorTolerance1.json | 12 +-
.../dataset/TestJsonSerDe/testVariousType.json | 2 +-
121 files changed, 3816 insertions(+), 349 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 4899ed2..2084737 100644
--- a/CHANGES
+++ b/CHANGES
@@ -4,14 +4,16 @@ Release 0.11.0 - unreleased
NEW FEATURES
+ TAJO-1344: Python UDF support. (jihoon)
+
+ TAJO-923: Add VAR_SAMP and VAR_POP window functions.
+ (Contributed by Dongjoon Hyun, Committed by jihoon)
+
TAJO-1494: Add SeekableScanner support to DelimitedTextFileScanner.
(jinho)
TAJO-921: Add STDDEV_SAMP and STDDEV_POP window functions. (Keuntae Park)
- TAJO-923: Add VAR_SAMP and VAR_POP window functions.
- (Contributed by Dongjoon Hyun, Committed by jihoon)
-
TAJO-1135: Implement queryable virtual table for cluster information.
(jihun)
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
index 628d710..dcfad8d 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
@@ -24,7 +24,6 @@ import org.apache.tajo.DataTypeUtil;
import org.apache.tajo.TajoConstants;
import org.apache.tajo.catalog.partition.PartitionMethodDesc;
import org.apache.tajo.catalog.proto.CatalogProtos;
-import org.apache.tajo.catalog.proto.CatalogProtos.ColumnProto;
import org.apache.tajo.catalog.proto.CatalogProtos.SchemaProto;
import org.apache.tajo.catalog.proto.CatalogProtos.TableDescProto;
import org.apache.tajo.common.TajoDataTypes;
@@ -418,7 +417,7 @@ public class CatalogUtil {
// (a) ()
// (a,b) (a)
- int definedSize = definedTypes == null ? 0 : definedTypes.size();
+ int definedSize = definedTypes.size();
int givenParamSize = givenTypes == null ? 0 : givenTypes.size();
int paramDiff = givenParamSize - definedSize;
if (paramDiff < 0) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/FunctionDesc.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/FunctionDesc.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/FunctionDesc.java
index 23d39f2..6ea6ac6 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/FunctionDesc.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/FunctionDesc.java
@@ -92,7 +92,7 @@ public class FunctionDesc implements ProtoObject<FunctionDescProto>, Cloneable,
*/
public Function newInstance() throws InternalException {
try {
- Constructor<? extends Function> cons = getFuncClass().getConstructor();
+ Constructor<? extends Function> cons = getLegacyFuncClass().getConstructor();
return cons.newInstance();
} catch (Exception ioe) {
throw new InternalException("Cannot initiate function " + signature);
@@ -124,7 +124,7 @@ public class FunctionDesc implements ProtoObject<FunctionDescProto>, Cloneable,
////////////////////////////////////////
@SuppressWarnings("unchecked")
- public Class<? extends Function> getFuncClass() {
+ public Class<? extends Function> getLegacyFuncClass() {
return invocation.getLegacy().getFunctionClass();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/FunctionInvocation.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/FunctionInvocation.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/FunctionInvocation.java
index 653bdb6..911d5dd 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/FunctionInvocation.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/FunctionInvocation.java
@@ -35,6 +35,8 @@ public class FunctionInvocation implements ProtoObject<FunctionInvocationProto>
StaticMethodInvocationDesc scalarJIT;
@Expose
ClassBaseInvocationDesc<?> aggregationJIT;
+ @Expose
+ PythonInvocationDesc python;
public FunctionInvocation() {
}
@@ -55,6 +57,9 @@ public class FunctionInvocation implements ProtoObject<FunctionInvocationProto>
if (proto.hasAggregationJIT()) {
this.aggregationJIT = new ClassBaseInvocationDesc(proto.getAggregation());
}
+ if (proto.hasPython()) {
+ this.python = new PythonInvocationDesc(proto.getPython());
+ }
}
public boolean isAvailable() {
@@ -121,6 +126,18 @@ public class FunctionInvocation implements ProtoObject<FunctionInvocationProto>
return aggregationJIT;
}
+ public boolean hasPython() {
+ return python != null;
+ }
+
+ public void setPython(PythonInvocationDesc python) {
+ this.python = python;
+ }
+
+ public PythonInvocationDesc getPython() {
+ return python;
+ }
+
@Override
public FunctionInvocationProto getProto() {
FunctionInvocationProto.Builder builder = FunctionInvocationProto.newBuilder();
@@ -139,16 +156,19 @@ public class FunctionInvocation implements ProtoObject<FunctionInvocationProto>
if (hasAggregationJIT()) {
builder.setAggregationJIT(aggregationJIT.getProto());
}
+ if (hasPython()) {
+ builder.setPython(python.getProto());
+ }
return builder.build();
}
@Override
public int hashCode() {
- return Objects.hashCode(legacy, scalar, scalarJIT);
+ return Objects.hashCode(legacy, scalar, scalarJIT, python);
}
public String toString() {
return "legacy=" + hasLegacy() + ",scalar=" + hasScalar() + ",agg=" + hasAggregation() +
- ",scalarJIT=" + hasScalarJIT() + ",aggJIT=" + hasAggregationJIT();
+ ",scalarJIT=" + hasScalarJIT() + ",aggJIT=" + hasAggregationJIT() + ",python=" + hasPython();
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/FunctionUtil.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/FunctionUtil.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/FunctionUtil.java
index dee5d1c..ef70428 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/FunctionUtil.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/FunctionUtil.java
@@ -18,6 +18,8 @@
package org.apache.tajo.function;
+import org.apache.tajo.catalog.FunctionDesc;
+
import java.util.Collection;
import static org.apache.tajo.common.TajoDataTypes.DataType;
@@ -53,4 +55,12 @@ public class FunctionUtil {
public static boolean isNullableParam(Class<?> clazz) {
return !clazz.isPrimitive();
}
+
+ public static boolean isLegacyFunction(FunctionDesc desc) {
+ return desc.getInvocation().hasLegacy();
+ }
+
+ public static boolean isScriptFunction(FunctionDesc desc) {
+ return desc.getInvocation().hasPython();
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/PythonInvocationDesc.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/PythonInvocationDesc.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/PythonInvocationDesc.java
new file mode 100644
index 0000000..160b169
--- /dev/null
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/function/PythonInvocationDesc.java
@@ -0,0 +1,98 @@
+/*
+ * 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.tajo.function;
+
+import com.google.common.base.Objects;
+import com.google.gson.annotations.Expose;
+import org.apache.tajo.catalog.proto.CatalogProtos.PythonInvocationDescProto;
+import org.apache.tajo.common.ProtoObject;
+import org.apache.tajo.util.TUtil;
+
+/**
+ * <code>PythonInvocationDesc</code> describes a function name
+ * and a file path to the script where the function is defined.
+ */
+public class PythonInvocationDesc implements ProtoObject<PythonInvocationDescProto>, Cloneable {
+ @Expose private String funcName;
+ @Expose private String filePath;
+
+ public PythonInvocationDesc() {
+
+ }
+
+ public PythonInvocationDesc(String funcName, String filePath) {
+ this.funcName = funcName;
+ this.filePath = filePath;
+ }
+
+ public void setFuncName(String funcName) {
+ this.funcName = funcName;
+ }
+
+ public void setFilePath(String filePath) {
+ this.filePath = filePath;
+ }
+
+ public PythonInvocationDesc(PythonInvocationDescProto proto) {
+ this(proto.getFuncName(), proto.getFilePath());
+ }
+
+ public String getName() {
+ return funcName;
+ }
+
+ public String getPath() {
+ return filePath;
+ }
+
+ @Override
+ public PythonInvocationDescProto getProto() {
+ PythonInvocationDescProto.Builder builder = PythonInvocationDescProto.newBuilder();
+ builder.setFuncName(funcName).setFilePath(filePath);
+ return builder.build();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof PythonInvocationDesc) {
+ PythonInvocationDesc other = (PythonInvocationDesc) o;
+ return TUtil.checkEquals(funcName, other.funcName) &&
+ TUtil.checkEquals(filePath, other.filePath);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(funcName, filePath);
+ }
+
+ @Override
+ public String toString() {
+ return funcName + " at " + filePath;
+ }
+
+ @Override
+ public Object clone() throws CloneNotSupportedException {
+ PythonInvocationDesc clone = (PythonInvocationDesc) super.clone();
+ clone.funcName = funcName == null ? null : funcName;
+ clone.filePath = filePath == null ? null : filePath;
+ return clone;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
index 1402bbc..fd2cb19 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
+++ b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
@@ -411,6 +411,7 @@ message FunctionInvocationProto {
optional ClassBaseInvocationDescProto aggregation = 3;
optional StaticMethodInvocationDescProto scalarJIT = 4;
optional ClassBaseInvocationDescProto aggregationJIT = 5;
+ optional PythonInvocationDescProto python = 6;
}
message ClassBaseInvocationDescProto {
@@ -423,3 +424,8 @@ message StaticMethodInvocationDescProto {
required string returnClass = 3;
repeated string paramClasses = 4;
}
+
+message PythonInvocationDescProto {
+ required string funcName = 1;
+ required string filePath = 2;
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestFunctionDesc.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestFunctionDesc.java b/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestFunctionDesc.java
index 92d2aa4..4a67ce6 100644
--- a/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestFunctionDesc.java
+++ b/tajo-catalog/tajo-catalog-common/src/test/java/org/apache/tajo/catalog/TestFunctionDesc.java
@@ -68,7 +68,7 @@ public class TestFunctionDesc {
desc.setDetail("detail");
assertEquals("sum", desc.getFunctionName());
- assertEquals(TestSum.class, desc.getFuncClass());
+ assertEquals(TestSum.class, desc.getLegacyFuncClass());
assertEquals(FunctionType.GENERAL, desc.getFuncType());
assertEquals(Type.INT4, desc.getReturnType().getType());
assertArrayEquals(CatalogUtil.newSimpleDataTypeArray(Type.INT4, Type.INT8),
@@ -84,7 +84,7 @@ public class TestFunctionDesc {
FunctionDesc newDesc = new FunctionDesc(proto);
assertEquals("sum", newDesc.getFunctionName());
- assertEquals(TestSum.class, newDesc.getFuncClass());
+ assertEquals(TestSum.class, newDesc.getLegacyFuncClass());
assertEquals(FunctionType.GENERAL, newDesc.getFuncType());
assertEquals(Type.INT4, newDesc.getReturnType().getType());
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
index 1773033..b10e5a5 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
@@ -62,8 +62,6 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
private Connection conn;
- protected Map<String, Boolean> baseTableMaps = new HashMap<String, Boolean>();
-
protected XMLCatalogSchemaManager catalogSchemaManager;
protected abstract String getCatalogDriverName();
@@ -1266,11 +1264,11 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
conn = getConnection();
pstmt = conn.prepareStatement(ADD_PARTITION_SQL);
-
pstmt.setInt(1, tableId);
pstmt.setString(2, partition.getPartitionName());
pstmt.setString(3, partition.getPath());
pstmt.executeUpdate();
+ pstmt.close();
if (partition.getPartitionKeysCount() > 0) {
pstmt = conn.prepareStatement(ADD_PARTITION_KEYS_SQL);
@@ -1349,6 +1347,7 @@ public abstract class AbstractDBStore extends CatalogConstants implements Catalo
pstmt = conn.prepareStatement(sqlDeletePartitionKeys);
pstmt.setInt(1, partitionId);
pstmt.executeUpdate();
+ pstmt.close();
pstmt = conn.prepareStatement(sqlDeletePartition);
pstmt.setInt(1, partitionId);
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
index b058504..821b00c 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/MemStore.java
@@ -30,7 +30,6 @@ import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.FunctionDesc;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.exception.*;
-import org.apache.tajo.catalog.partition.PartitionDesc;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.proto.CatalogProtos.ColumnProto;
import org.apache.tajo.catalog.proto.CatalogProtos.DatabaseProto;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java b/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
index 0a2a8cc..306f581 100644
--- a/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
+++ b/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
@@ -666,7 +666,7 @@ public class TestCatalog {
FunctionDesc retrived = catalog.getFunction("test10", CatalogUtil.newSimpleDataTypeArray(Type.INT4, Type.BLOB));
assertEquals(retrived.getFunctionName(), "test10");
- assertEquals(retrived.getFuncClass(), TestFunc2.class);
+ assertEquals(retrived.getLegacyFuncClass(), TestFunc2.class);
assertEquals(retrived.getFuncType(), FunctionType.GENERAL);
assertFalse(catalog.containFunction("test10", CatalogUtil.newSimpleDataTypeArray(Type.BLOB, Type.INT4)));
@@ -685,7 +685,7 @@ public class TestCatalog {
FunctionDesc retrived = catalog.getFunction("test2", CatalogUtil.newSimpleDataTypeArray(Type.INT4));
assertEquals(retrived.getFunctionName(),"test2");
- assertEquals(retrived.getFuncClass(),TestFunc1.class);
+ assertEquals(retrived.getLegacyFuncClass(),TestFunc1.class);
assertEquals(retrived.getFuncType(),FunctionType.UDF);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-common/src/main/java/org/apache/tajo/QueryVars.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/QueryVars.java b/tajo-common/src/main/java/org/apache/tajo/QueryVars.java
index ba76d63..55ca700 100644
--- a/tajo-common/src/main/java/org/apache/tajo/QueryVars.java
+++ b/tajo-common/src/main/java/org/apache/tajo/QueryVars.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
index 8e2b3d2..bfba290 100644
--- a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
@@ -293,6 +293,10 @@ public class TajoConf extends Configuration {
// Geo IP
GEOIP_DATA("tajo.function.geoip-database-location", ""),
+ // Python UDF
+ PYTHON_CODE_DIR("tajo.function.python.code-dir", ""),
+ PYTHON_CONTROLLER_LOG_DIR("tajo.function.python.controller.log-dir", ""),
+
/////////////////////////////////////////////////////////////////////////////////
// User Session Configuration
//
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-common/src/main/java/org/apache/tajo/datum/AnyDatum.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/datum/AnyDatum.java b/tajo-common/src/main/java/org/apache/tajo/datum/AnyDatum.java
new file mode 100644
index 0000000..0771a6e
--- /dev/null
+++ b/tajo-common/src/main/java/org/apache/tajo/datum/AnyDatum.java
@@ -0,0 +1,82 @@
+/*
+ * 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.tajo.datum;
+
+import com.google.gson.annotations.Expose;
+
+import static org.apache.tajo.common.TajoDataTypes.Type.ANY;
+
+/**
+ * <code>AnyDatum</code> can contain any types of datum.
+ */
+public class AnyDatum extends Datum {
+ @Expose Datum val;
+
+ public AnyDatum(Datum val) {
+ super(ANY);
+ this.val = val;
+ }
+
+ public Datum getActual() {
+ return val;
+ }
+
+ @Override
+ public int size() {
+ return this.val.size();
+ }
+
+ @Override
+ public int hashCode() {
+ return val.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof AnyDatum) {
+ AnyDatum other = (AnyDatum) obj;
+ return val.equals(other.val);
+ }
+ return false;
+ }
+
+ @Override
+ public Datum equalsTo(Datum datum) {
+ if (datum.type() == ANY) {
+ AnyDatum other = (AnyDatum) datum;
+ return val.equalsTo(other.val);
+ }
+ return DatumFactory.createBool(false);
+ }
+
+ @Override
+ public int compareTo(Datum datum) {
+ if (datum.type() == ANY) {
+ AnyDatum other = (AnyDatum) datum;
+ return val.compareTo(other.val);
+ }
+ // Any datums will be lastly appeared.
+ return 1;
+ }
+
+ @Override
+ public String toString() {
+ return val.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-common/src/main/java/org/apache/tajo/datum/BlobDatum.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/datum/BlobDatum.java b/tajo-common/src/main/java/org/apache/tajo/datum/BlobDatum.java
index 4f296a1..cf190e2 100644
--- a/tajo-common/src/main/java/org/apache/tajo/datum/BlobDatum.java
+++ b/tajo-common/src/main/java/org/apache/tajo/datum/BlobDatum.java
@@ -23,6 +23,7 @@ package org.apache.tajo.datum;
import com.google.gson.annotations.Expose;
import org.apache.tajo.exception.InvalidOperationException;
+import org.apache.tajo.util.TUtil;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
@@ -130,8 +131,8 @@ public class BlobDatum extends Datum {
BlobDatum other = (BlobDatum) obj;
initFromBytes();
other.initFromBytes();
- return bb.equals(other.bb);
- }
+ return Arrays.equals(this.val, other.val);
+ }
return false;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java b/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
index 9f48cad..bd1b88f 100644
--- a/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
+++ b/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
@@ -19,6 +19,7 @@
package org.apache.tajo.datum;
import com.google.protobuf.Message;
+import org.apache.commons.codec.binary.Base64;
import org.apache.tajo.common.TajoDataTypes.DataType;
import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.exception.InvalidCastException;
@@ -66,7 +67,7 @@ public class DatumFactory {
case INET4:
return Inet4Datum.class;
case ANY:
- return NullDatum.class;
+ return AnyDatum.class;
case NULL_TYPE:
return NullDatum.class;
default:
@@ -377,16 +378,16 @@ public class DatumFactory {
return new TimestampDatum(DateTimeUtil.toJulianTimestampWithTZ(str, tz));
}
- public static BlobDatum createBlob(byte[] val) {
- return new BlobDatum(val);
+ public static BlobDatum createBlob(byte[] encoded) {
+ return new BlobDatum(encoded);
}
- public static BlobDatum createBlob(byte[] val, int offset, int length) {
- return new BlobDatum(val, offset, length);
+ public static BlobDatum createBlob(byte[] encoded, int offset, int length) {
+ return new BlobDatum(encoded, offset, length);
}
- public static BlobDatum createBlob(String val) {
- return new BlobDatum(val.getBytes());
+ public static BlobDatum createBlob(String plainString) {
+ return new BlobDatum(Base64.encodeBase64(plainString.getBytes()));
}
public static Inet4Datum createInet4(int encoded) {
@@ -405,6 +406,10 @@ public class DatumFactory {
return new Inet4Datum(val);
}
+ public static AnyDatum createAny(Datum val) {
+ return new AnyDatum(val);
+ }
+
public static Datum cast(Datum operandDatum, DataType target, @Nullable TimeZone tz) {
switch (target.getType()) {
case BOOLEAN:
@@ -454,6 +459,8 @@ public class DatumFactory {
return DatumFactory.createBlob(operandDatum.asByteArray());
case INET4:
return DatumFactory.createInet4(operandDatum.asByteArray());
+ case ANY:
+ return DatumFactory.createAny(operandDatum);
default:
throw new InvalidCastException(operandDatum.type(), target.getType());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-common/src/main/java/org/apache/tajo/json/DatumAdapter.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/json/DatumAdapter.java b/tajo-common/src/main/java/org/apache/tajo/json/DatumAdapter.java
index d65559d..b9d8aef 100644
--- a/tajo-common/src/main/java/org/apache/tajo/json/DatumAdapter.java
+++ b/tajo-common/src/main/java/org/apache/tajo/json/DatumAdapter.java
@@ -41,8 +41,9 @@ public class DatumAdapter implements GsonSerDerAdapter<Datum> {
return new TimestampDatum(CommonGsonHelper.getOrDie(jsonObject, "value").getAsLong());
case INTERVAL:
String[] values = CommonGsonHelper.getOrDie(jsonObject, "value").getAsString().split(",");
-
return new IntervalDatum(Integer.parseInt(values[0]), Long.parseLong(values[1]));
+ case ANY:
+ return new AnyDatum(deserialize(CommonGsonHelper.getOrDie(jsonObject, "actual"), typeOfT, context));
default:
return context.deserialize(CommonGsonHelper.getOrDie(jsonObject, "body"),
DatumFactory.getDatumClass(TajoDataTypes.Type.valueOf(typeName)));
@@ -67,6 +68,9 @@ public class DatumAdapter implements GsonSerDerAdapter<Datum> {
IntervalDatum interval = (IntervalDatum)src;
jsonObj.addProperty("value", interval.getMonths() + "," + interval.getMilliSeconds());
break;
+ case ANY:
+ jsonObj.add("actual", serialize(((AnyDatum) src).getActual(), typeOfSrc, context));
+ break;
default:
jsonObj.add("body", context.serialize(src));
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-common/src/main/java/org/apache/tajo/storage/Tuple.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/storage/Tuple.java b/tajo-common/src/main/java/org/apache/tajo/storage/Tuple.java
index 1ba1926..aec784f 100644
--- a/tajo-common/src/main/java/org/apache/tajo/storage/Tuple.java
+++ b/tajo-common/src/main/java/org/apache/tajo/storage/Tuple.java
@@ -22,58 +22,58 @@ import org.apache.tajo.datum.Datum;
public interface Tuple extends Cloneable {
- public int size();
+ int size();
- public boolean contains(int fieldid);
+ boolean contains(int fieldid);
- public boolean isNull(int fieldid);
+ boolean isNull(int fieldid);
@SuppressWarnings("unused")
- public boolean isNotNull(int fieldid);
+ boolean isNotNull(int fieldid);
- public void clear();
+ void clear();
- public void put(int fieldId, Datum value);
+ void put(int fieldId, Datum value);
- public void put(int fieldId, Datum[] values);
+ void put(int fieldId, Datum[] values);
- public void put(int fieldId, Tuple tuple);
+ void put(int fieldId, Tuple tuple);
- public void put(Datum[] values);
+ void put(Datum[] values);
- public Datum get(int fieldId);
+ Datum get(int fieldId);
- public void setOffset(long offset);
+ void setOffset(long offset);
- public long getOffset();
+ long getOffset();
- public boolean getBool(int fieldId);
+ boolean getBool(int fieldId);
- public byte getByte(int fieldId);
+ byte getByte(int fieldId);
- public char getChar(int fieldId);
+ char getChar(int fieldId);
- public byte [] getBytes(int fieldId);
+ byte [] getBytes(int fieldId);
- public short getInt2(int fieldId);
+ short getInt2(int fieldId);
- public int getInt4(int fieldId);
+ int getInt4(int fieldId);
- public long getInt8(int fieldId);
+ long getInt8(int fieldId);
- public float getFloat4(int fieldId);
+ float getFloat4(int fieldId);
- public double getFloat8(int fieldId);
+ double getFloat8(int fieldId);
- public String getText(int fieldId);
+ String getText(int fieldId);
- public Datum getProtobufDatum(int fieldId);
+ Datum getProtobufDatum(int fieldId);
- public Datum getInterval(int fieldId);
+ Datum getInterval(int fieldId);
- public char [] getUnicodeChars(int fieldId);
+ char [] getUnicodeChars(int fieldId);
- public Tuple clone() throws CloneNotSupportedException;
+ Tuple clone() throws CloneNotSupportedException;
- public Datum[] getValues();
+ Datum[] getValues();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-common/src/main/java/org/apache/tajo/util/FileUtil.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/FileUtil.java b/tajo-common/src/main/java/org/apache/tajo/util/FileUtil.java
index 9403a2f..3e3d3a2 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/FileUtil.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/FileUtil.java
@@ -19,6 +19,7 @@
package org.apache.tajo.util;
import com.google.protobuf.Message;
+import org.apache.commons.logging.Log;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
@@ -143,4 +144,26 @@ public class FileUtil {
public static boolean isLocalPath(Path path) {
return path.toUri().getScheme().equals("file");
}
+
+
+ /**
+ * Close the Closeable objects and <b>ignore</b> any {@link IOException} or
+ * null pointers. Must only be used for cleanup in exception handlers.
+ *
+ * @param log the log to record problems to at debug level. Can be null.
+ * @param closeables the objects to close
+ */
+ public static void cleanup(Log log, java.io.Closeable... closeables) {
+ for (java.io.Closeable c : closeables) {
+ if (c != null) {
+ try {
+ c.close();
+ } catch(IOException e) {
+ if (log != null && log.isDebugEnabled()) {
+ log.debug("Exception in closing " + c, e);
+ }
+ }
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-common/src/main/java/org/apache/tajo/util/KeyValueSet.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/KeyValueSet.java b/tajo-common/src/main/java/org/apache/tajo/util/KeyValueSet.java
index 6af0c9e..5dba9e2 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/KeyValueSet.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/KeyValueSet.java
@@ -104,7 +104,7 @@ public class KeyValueSet implements ProtoObject<KeyValueSetProto>, Cloneable, Gs
} else if (defaultVal != null) {
return defaultVal;
} else {
- throw new IllegalArgumentException("No such a config key: " + key);
+ throw new IllegalArgumentException("No such config key: " + key);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeFormat.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeFormat.java b/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeFormat.java
index c3aa71e..798b9c5 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeFormat.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeFormat.java
@@ -477,7 +477,6 @@ public class DateTimeFormat {
KeyWord keyword = new KeyWord();
keyword.name = (String)eachKeywordValue[0];
keyword.len = ((Integer)eachKeywordValue[1]).intValue();
- keyword.id = ((DCH_poz)eachKeywordValue[2]).getValue();
keyword.idType = ((DCH_poz)eachKeywordValue[2]);
keyword.is_digit = ((Boolean)eachKeywordValue[3]).booleanValue();
keyword.date_mode = (FromCharDateMode)eachKeywordValue[4];
@@ -513,7 +512,6 @@ public class DateTimeFormat {
static class KeyWord {
String name;
int len;
- int id;
DCH_poz idType;
boolean is_digit;
FromCharDateMode date_mode;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/codegen/EvalCodeGenContext.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/EvalCodeGenContext.java b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/EvalCodeGenContext.java
index c8197b7..32fb562 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/EvalCodeGenContext.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/EvalCodeGenContext.java
@@ -174,7 +174,7 @@ public class EvalCodeGenContext extends TajoGeneratorAdapter {
} else if (entry.getKey().getType() == EvalType.FUNCTION) {
GeneralFunctionEval function = (GeneralFunctionEval) entry.getKey();
- final String internalName = TajoGeneratorAdapter.getInternalName(function.getFuncDesc().getFuncClass());
+ final String internalName = TajoGeneratorAdapter.getInternalName(function.getFuncDesc().getLegacyFuncClass());
// new and initialization of function
initMethod.visitTypeInsn(Opcodes.NEW, internalName);
@@ -198,12 +198,12 @@ public class EvalCodeGenContext extends TajoGeneratorAdapter {
initMethod.visitVarInsn(Opcodes.ALOAD, FUNCTION);
consAdapter.aload(PARAM_TYPE_ARRAY);
- consAdapter.invokeVirtual(function.getFuncDesc().getFuncClass(), "init", void.class, new Class[] {FunctionEval.ParamType[].class});
+ consAdapter.invokeVirtual(function.getFuncDesc().getLegacyFuncClass(), "init", void.class, new Class[] {FunctionEval.ParamType[].class});
initMethod.visitVarInsn(Opcodes.ALOAD, 0);
initMethod.visitVarInsn(Opcodes.ALOAD, FUNCTION);
initMethod.visitFieldInsn(Opcodes.PUTFIELD, this.owner, entry.getValue(),
- "L" + TajoGeneratorAdapter.getInternalName(function.getFuncDesc().getFuncClass()) + ";");
+ "L" + TajoGeneratorAdapter.getInternalName(function.getFuncDesc().getLegacyFuncClass()) + ";");
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/codegen/LegacyFunctionBindingEmitter.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/LegacyFunctionBindingEmitter.java b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/LegacyFunctionBindingEmitter.java
index 36dfe4f..6b47149 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/LegacyFunctionBindingEmitter.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/LegacyFunctionBindingEmitter.java
@@ -63,12 +63,12 @@ public class LegacyFunctionBindingEmitter {
FunctionDesc desc = func.getFuncDesc();
String fieldName = context.symbols.get(func);
- String funcDescName = "L" + TajoGeneratorAdapter.getInternalName(desc.getFuncClass()) + ";";
+ String funcDescName = "L" + TajoGeneratorAdapter.getInternalName(desc.getLegacyFuncClass()) + ";";
context.aload(0);
context.methodvisitor.visitFieldInsn(Opcodes.GETFIELD, context.owner, fieldName, funcDescName);
context.aload(TUPLE);
- context.invokeVirtual(desc.getFuncClass(), "eval", Datum.class, new Class[] {Tuple.class});
+ context.invokeVirtual(desc.getLegacyFuncClass(), "eval", Datum.class, new Class[] {Tuple.class});
context.convertToPrimitive(func.getValueType());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/codegen/VariablesPreBuilder.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/VariablesPreBuilder.java b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/VariablesPreBuilder.java
index a055b04..95ec371 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/codegen/VariablesPreBuilder.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/codegen/VariablesPreBuilder.java
@@ -75,7 +75,7 @@ class VariablesPreBuilder extends SimpleEvalNodeVisitor<EvalCodeGenContext> {
String fieldName = function.getFuncDesc().getFunctionName() + "_" + context.seqId++;
context.symbols.put(function, fieldName);
context.classWriter.visitField(Opcodes.ACC_PRIVATE, fieldName,
- "L" + TajoGeneratorAdapter.getInternalName(function.getFuncDesc().getFuncClass()) + ";", null, null);
+ "L" + TajoGeneratorAdapter.getInternalName(function.getFuncDesc().getLegacyFuncClass()) + ";", null, null);
}
return function;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/function/FunctionLoader.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/FunctionLoader.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/FunctionLoader.java
index 3b3e7c7..6061d1b 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/FunctionLoader.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/FunctionLoader.java
@@ -25,16 +25,24 @@ import com.google.common.collect.Sets;
import org.apache.commons.collections.Predicate;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
import org.apache.tajo.annotation.Nullable;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.FunctionDesc;
import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.function.annotation.Description;
import org.apache.tajo.engine.function.annotation.ParamOptionTypes;
import org.apache.tajo.engine.function.annotation.ParamTypes;
import org.apache.tajo.function.*;
+import org.apache.tajo.plan.function.python.PythonScriptEngine;
import org.apache.tajo.util.ClassUtil;
+import org.apache.tajo.util.TUtil;
+import java.io.IOException;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.*;
@@ -44,8 +52,14 @@ import static org.apache.tajo.catalog.proto.CatalogProtos.FunctionType.GENERAL;
public class FunctionLoader {
private static Log LOG = LogFactory.getLog(FunctionLoader.class);
+ public static final String PYTHON_FUNCTION_NAMESPACE = "python";
- public static Collection<FunctionDesc> load() {
+ /**
+ * Load built-in functions
+ *
+ * @return
+ */
+ public static Map<FunctionSignature, FunctionDesc> load() {
Map<FunctionSignature, FunctionDesc> map = Maps.newHashMap();
List<FunctionDesc> dd = Lists.newArrayList();
@@ -66,7 +80,48 @@ public class FunctionLoader {
}
}
- return map.values();
+ return map;
+ }
+
+ /**
+ * Load functions that are defined by users.
+ *
+ * @param conf
+ * @param functionMap
+ * @return
+ * @throws IOException
+ */
+ public static Map<FunctionSignature, FunctionDesc> loadUserDefinedFunctions(TajoConf conf,
+ Map<FunctionSignature, FunctionDesc> functionMap)
+ throws IOException {
+
+ String[] codePaths = conf.getStrings(TajoConf.ConfVars.PYTHON_CODE_DIR.varname);
+ if (codePaths != null) {
+ FileSystem localFS = FileSystem.getLocal(conf);
+ for (String codePathStr : codePaths) {
+ Path codePath = new Path(codePathStr);
+ List<Path> filePaths = TUtil.newList();
+ if (localFS.isDirectory(codePath)) {
+ for (FileStatus file : localFS.listStatus(codePath, new PathFilter() {
+ @Override
+ public boolean accept(Path path) {
+ return path.getName().endsWith(PythonScriptEngine.FILE_EXTENSION);
+ }
+ })) {
+ filePaths.add(file.getPath());
+ }
+ } else {
+ filePaths.add(codePath);
+ }
+ for (Path filePath : filePaths) {
+ for (FunctionDesc f : PythonScriptEngine.registerFunctions(filePath.toUri(),
+ FunctionLoader.PYTHON_FUNCTION_NAMESPACE)) {
+ functionMap.put(f.getSignature(), f);
+ }
+ }
+ }
+ }
+ return functionMap;
}
public static Set<FunctionDesc> findScalarFunctions() {
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/Projector.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/Projector.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/Projector.java
index cec1862..a73478f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/Projector.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/Projector.java
@@ -63,7 +63,7 @@ public class Projector {
public void init() {
for (EvalNode eval : evals) {
- eval.bind(inSchema);
+ eval.bind(context.getEvalContext(), inSchema);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/rewriter/rules/GlobalPlanEqualityTester.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/rewriter/rules/GlobalPlanEqualityTester.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/rewriter/rules/GlobalPlanEqualityTester.java
index e2fd47f..e55a258 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/rewriter/rules/GlobalPlanEqualityTester.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/rewriter/rules/GlobalPlanEqualityTester.java
@@ -51,7 +51,7 @@ public class GlobalPlanEqualityTester implements GlobalPlanRewriteRule {
LogicalNode node = eb.getPlan();
if (node != null) {
PlanProto.LogicalNodeTree tree = LogicalNodeSerializer.serialize(node);
- LogicalNode deserialize = LogicalNodeDeserializer.deserialize(plan.getContext(), tree);
+ LogicalNode deserialize = LogicalNodeDeserializer.deserialize(plan.getContext(), null, tree);
assert node.deepEquals(deserialize);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/AggregationExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/AggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/AggregationExec.java
index 6d9e38a..4b53b39 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/AggregationExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/AggregationExec.java
@@ -63,7 +63,7 @@ public abstract class AggregationExec extends UnaryPhysicalExec {
public void init() throws IOException {
super.init();
for (EvalNode aggFunction : aggFunctions) {
- aggFunction.bind(inSchema);
+ aggFunction.bind(context.getEvalContext(), inSchema);
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java
index be6c046..806d34c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/BSTIndexScanExec.java
@@ -72,7 +72,7 @@ public class BSTIndexScanExec extends PhysicalExec {
super.init();
progress = 0.0f;
if (qual != null) {
- qual.bind(inSchema);
+ qual.bind(context.getEvalContext(), inSchema);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/CommonJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/CommonJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/CommonJoinExec.java
index 0781041..2535edf 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/CommonJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/CommonJoinExec.java
@@ -54,7 +54,7 @@ public abstract class CommonJoinExec extends BinaryPhysicalExec {
public void init() throws IOException {
super.init();
if (hasJoinQual) {
- joinQual.bind(inSchema);
+ joinQual.bind(context.getEvalContext(), inSchema);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java
index 37bc5a7..94429a0 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyFirstAggregationExec.java
@@ -250,7 +250,7 @@ public class DistinctGroupbyFirstAggregationExec extends UnaryPhysicalExec {
}
for (AggregationFunctionCallEval eachFunction: aggFunctions) {
- eachFunction.bind(inSchema);
+ eachFunction.bind(context.getEvalContext(), inSchema);
eachFunction.setFirstPhase();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java
index e96e750..0f25d6c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyHashAggregationExec.java
@@ -375,7 +375,7 @@ public class DistinctGroupbyHashAggregationExec extends UnaryPhysicalExec {
}
for (EvalNode aggFunction : aggFunctions) {
- aggFunction.bind(schema);
+ aggFunction.bind(context.getEvalContext(), schema);
}
tupleSize = groupingKeyIds.length + aggFunctionsNum;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySecondAggregationExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySecondAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySecondAggregationExec.java
index 7b01a9b..b394390 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySecondAggregationExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbySecondAggregationExec.java
@@ -119,7 +119,7 @@ public class DistinctGroupbySecondAggregationExec extends UnaryPhysicalExec {
nonDistinctAggrFunctions = eachGroupby.getAggFunctions();
if (nonDistinctAggrFunctions != null) {
for (AggregationFunctionCallEval eachFunction: nonDistinctAggrFunctions) {
- eachFunction.bind(inSchema);
+ eachFunction.bind(context.getEvalContext(), inSchema);
eachFunction.setIntermediatePhase();
}
nonDistinctAggrContexts = new FunctionContext[nonDistinctAggrFunctions.length];
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java
index 7bd71e2..e71976c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/DistinctGroupbyThirdAggregationExec.java
@@ -252,7 +252,7 @@ public class DistinctGroupbyThirdAggregationExec extends UnaryPhysicalExec {
aggrFunctions = groupbyNode.getAggFunctions();
if (aggrFunctions != null) {
for (AggregationFunctionCallEval eachFunction: aggrFunctions) {
- eachFunction.bind(inSchema);
+ eachFunction.bind(context.getEvalContext(), inSchema);
eachFunction.setFinalPhase();
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/EvalExprExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/EvalExprExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/EvalExprExec.java
index 32ec772..4581b4a 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/EvalExprExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/EvalExprExec.java
@@ -41,7 +41,7 @@ public class EvalExprExec extends PhysicalExec {
super.init();
progress = 0.0f;
for (Target target : plan.getTargets()) {
- target.getEvalTree().bind(inSchema);
+ target.getEvalTree().bind(context.getEvalContext(), inSchema);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
index fa9ba94..6f573d0 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
@@ -121,9 +121,9 @@ public class HashLeftOuterJoinExec extends BinaryPhysicalExec {
rightNumCols = rightChild.getSchema().size();
- joinQual.bind(inSchema);
+ joinQual.bind(context.getEvalContext(), inSchema);
if (joinFilter != null) {
- joinFilter.bind(inSchema);
+ joinFilter.bind(context.getEvalContext(), inSchema);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HavingExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HavingExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HavingExec.java
index b71c770..6897e92 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HavingExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HavingExec.java
@@ -39,7 +39,7 @@ public class HavingExec extends UnaryPhysicalExec {
@Override
public void init() throws IOException {
super.init();
- qual.bind(inSchema);
+ qual.bind(context.getEvalContext(), inSchema);
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SelectionExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SelectionExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SelectionExec.java
index c090fa7..7f5bbe9 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SelectionExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SelectionExec.java
@@ -39,7 +39,7 @@ public class SelectionExec extends UnaryPhysicalExec {
@Override
public void init() throws IOException {
super.init();
- qual.bind(inSchema);
+ qual.bind(context.getEvalContext(), inSchema);
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
index 671555c..ff9477f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
@@ -109,7 +109,7 @@ public class SeqScanExec extends ScanExec {
FieldEval targetExpr = new FieldEval(column);
Datum datum = NullDatum.get();
if (partitionRow != null) {
- targetExpr.bind(columnPartitionSchema);
+ targetExpr.bind(context.getEvalContext(), columnPartitionSchema);
datum = targetExpr.eval(partitionRow);
}
ConstEval constExpr = new ConstEval(datum);
@@ -163,7 +163,7 @@ public class SeqScanExec extends ScanExec {
super.init();
if (plan.hasQual()) {
- qual.bind(inSchema);
+ qual.bind(context.getEvalContext(), inSchema);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
index 2f1fc46..05b0418 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
@@ -183,7 +183,7 @@ public class WindowAggExec extends UnaryPhysicalExec {
public void init() throws IOException {
super.init();
for (EvalNode functionEval : functions) {
- functionEval.bind(inSchema);
+ functionEval.bind(context.getEvalContext(), inSchema);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryContext.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryContext.java b/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryContext.java
index 7b3c00d..ee50221 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryContext.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/query/QueryContext.java
@@ -25,8 +25,8 @@ import org.apache.tajo.QueryVars;
import org.apache.tajo.SessionVars;
import org.apache.tajo.catalog.partition.PartitionMethodDesc;
import org.apache.tajo.conf.TajoConf;
-import org.apache.tajo.session.Session;
import org.apache.tajo.plan.logical.NodeType;
+import org.apache.tajo.session.Session;
import static org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.KeyValueSetProto;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java b/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
index 08403ff..074f34e 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/GlobalEngine.java
@@ -44,7 +44,6 @@ import org.apache.tajo.ipc.ClientProtos;
import org.apache.tajo.master.TajoMaster.MasterContext;
import org.apache.tajo.master.exec.DDLExecutor;
import org.apache.tajo.master.exec.QueryExecutor;
-import org.apache.tajo.master.exec.prehook.DistributedQueryHookManager;
import org.apache.tajo.session.Session;
import org.apache.tajo.plan.*;
import org.apache.tajo.plan.logical.InsertNode;
@@ -77,7 +76,6 @@ public class GlobalEngine extends AbstractService {
private LogicalPlanner planner;
private LogicalOptimizer optimizer;
private LogicalPlanVerifier annotatedPlanVerifier;
- private DistributedQueryHookManager hookManager;
private QueryExecutor queryExecutor;
private DDLExecutor ddlExecutor;
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
index 51f82f8..0a5de58 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
@@ -21,10 +21,7 @@ package org.apache.tajo.master;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.service.CompositeService;
@@ -36,10 +33,12 @@ import org.apache.hadoop.yarn.util.RackResolver;
import org.apache.hadoop.yarn.util.SystemClock;
import org.apache.tajo.catalog.CatalogServer;
import org.apache.tajo.catalog.CatalogService;
+import org.apache.tajo.catalog.FunctionDesc;
import org.apache.tajo.catalog.LocalCatalogWrapper;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.conf.TajoConf.ConfVars;
import org.apache.tajo.engine.function.FunctionLoader;
+import org.apache.tajo.function.FunctionSignature;
import org.apache.tajo.master.rm.TajoWorkerResourceManager;
import org.apache.tajo.master.rm.WorkerResourceManager;
import org.apache.tajo.metrics.CatalogMetricsGaugeSet;
@@ -68,7 +67,9 @@ import java.lang.management.ThreadMXBean;
import java.lang.reflect.Constructor;
import java.net.InetSocketAddress;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.List;
+import java.util.Map;
import static org.apache.tajo.TajoConstants.DEFAULT_DATABASE_NAME;
import static org.apache.tajo.TajoConstants.DEFAULT_TABLESPACE_NAME;
@@ -177,7 +178,7 @@ public class TajoMaster extends CompositeService {
diagnoseTajoMaster();
this.storeManager = StorageManager.getFileStorageManager(systemConf);
- catalogServer = new CatalogServer(FunctionLoader.load());
+ catalogServer = new CatalogServer(loadFunctions());
addIfService(catalogServer);
catalog = new LocalCatalogWrapper(catalogServer, systemConf);
@@ -207,6 +208,11 @@ public class TajoMaster extends CompositeService {
LOG.info("Tajo Master is initialized.");
}
+ private Collection<FunctionDesc> loadFunctions() throws IOException {
+ Map<FunctionSignature, FunctionDesc> functionMap = FunctionLoader.load();
+ return FunctionLoader.loadUserDefinedFunctions(systemConf, functionMap).values();
+ }
+
private void initSystemMetrics() {
systemMetrics = new TajoSystemMetrics(systemConf, METRICS_GROUP_NAME, getMasterName());
systemMetrics.start();
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/master/exec/QueryExecutor.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/exec/QueryExecutor.java b/tajo-core/src/main/java/org/apache/tajo/master/exec/QueryExecutor.java
index 2eb3c5f..ad1a8e3 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/exec/QueryExecutor.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/exec/QueryExecutor.java
@@ -47,6 +47,10 @@ import org.apache.tajo.master.*;
import org.apache.tajo.master.exec.prehook.CreateTableHook;
import org.apache.tajo.master.exec.prehook.DistributedQueryHookManager;
import org.apache.tajo.master.exec.prehook.InsertIntoHook;
+import org.apache.tajo.plan.expr.EvalContext;
+import org.apache.tajo.plan.expr.GeneralFunctionEval;
+import org.apache.tajo.plan.function.python.PythonScriptEngine;
+import org.apache.tajo.plan.function.python.TajoScriptEngine;
import org.apache.tajo.plan.rewrite.LogicalPlanRewriteRule;
import org.apache.tajo.querymaster.*;
import org.apache.tajo.session.Session;
@@ -113,11 +117,9 @@ public class QueryExecutor {
} else if (PlannerUtil.checkIfSimpleQuery(plan)) {
execSimpleQuery(queryContext, session, sql, plan, response);
-
// NonFromQuery indicates a form of 'select a, x+y;'
} else if (PlannerUtil.checkIfNonFromQuery(plan)) {
- execNonFromQuery(queryContext, session, sql, plan, response);
-
+ execNonFromQuery(queryContext, plan, response);
} else { // it requires distributed execution. So, the query is forwarded to a query master.
executeDistributedQuery(queryContext, session, plan, sql, jsonExpr, response);
@@ -263,37 +265,66 @@ public class QueryExecutor {
response.setResultCode(ClientProtos.ResultCode.OK);
}
- public void execNonFromQuery(QueryContext queryContext, Session session, String query,
- LogicalPlan plan, SubmitQueryResponse.Builder responseBuilder) throws Exception {
+ public void execNonFromQuery(QueryContext queryContext, LogicalPlan plan, SubmitQueryResponse.Builder responseBuilder)
+ throws Exception {
LogicalRootNode rootNode = plan.getRootBlock().getRoot();
+ EvalContext evalContext = new EvalContext();
Target[] targets = plan.getRootBlock().getRawTargets();
if (targets == null) {
throw new PlanningException("No targets");
}
- final Tuple outTuple = new VTuple(targets.length);
+ try {
+ // start script executor
+ startScriptExecutors(queryContext, evalContext, targets);
+ final Tuple outTuple = new VTuple(targets.length);
+ for (int i = 0; i < targets.length; i++) {
+ EvalNode eval = targets[i].getEvalTree();
+ eval.bind(evalContext, null);
+ outTuple.put(i, eval.eval(null));
+ }
+ boolean isInsert = rootNode.getChild() != null && rootNode.getChild().getType() == NodeType.INSERT;
+ if (isInsert) {
+ InsertNode insertNode = rootNode.getChild();
+ insertNonFromQuery(queryContext, insertNode, responseBuilder);
+ } else {
+ Schema schema = PlannerUtil.targetToSchema(targets);
+ RowStoreUtil.RowStoreEncoder encoder = RowStoreUtil.createEncoder(schema);
+ byte[] serializedBytes = encoder.toBytes(outTuple);
+ ClientProtos.SerializedResultSet.Builder serializedResBuilder = ClientProtos.SerializedResultSet.newBuilder();
+ serializedResBuilder.addSerializedTuples(ByteString.copyFrom(serializedBytes));
+ serializedResBuilder.setSchema(schema.getProto());
+ serializedResBuilder.setBytesNum(serializedBytes.length);
+
+ responseBuilder.setResultSet(serializedResBuilder);
+ responseBuilder.setMaxRowNum(1);
+ responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto());
+ responseBuilder.setResultCode(ClientProtos.ResultCode.OK);
+ }
+ } finally {
+ // stop script executor
+ stopScriptExecutors(evalContext);
+ }
+ }
+
+ public static void startScriptExecutors(QueryContext queryContext, EvalContext evalContext, Target[] targets)
+ throws IOException {
for (int i = 0; i < targets.length; i++) {
EvalNode eval = targets[i].getEvalTree();
- eval.bind(null);
- outTuple.put(i, eval.eval(null));
+ if (eval instanceof GeneralFunctionEval) {
+ GeneralFunctionEval functionEval = (GeneralFunctionEval) eval;
+ if (functionEval.getFuncDesc().getInvocation().hasPython()) {
+ TajoScriptEngine scriptExecutor = new PythonScriptEngine(functionEval.getFuncDesc());
+ evalContext.addScriptEngine(eval, scriptExecutor);
+ scriptExecutor.start(queryContext.getConf());
+ }
+ }
}
- boolean isInsert = rootNode.getChild() != null && rootNode.getChild().getType() == NodeType.INSERT;
- if (isInsert) {
- InsertNode insertNode = rootNode.getChild();
- insertNonFromQuery(queryContext, insertNode, responseBuilder);
- } else {
- Schema schema = PlannerUtil.targetToSchema(targets);
- RowStoreUtil.RowStoreEncoder encoder = RowStoreUtil.createEncoder(schema);
- byte[] serializedBytes = encoder.toBytes(outTuple);
- ClientProtos.SerializedResultSet.Builder serializedResBuilder = ClientProtos.SerializedResultSet.newBuilder();
- serializedResBuilder.addSerializedTuples(ByteString.copyFrom(serializedBytes));
- serializedResBuilder.setSchema(schema.getProto());
- serializedResBuilder.setBytesNum(serializedBytes.length);
-
- responseBuilder.setResultSet(serializedResBuilder);
- responseBuilder.setMaxRowNum(1);
- responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto());
- responseBuilder.setResultCode(ClientProtos.ResultCode.OK);
+ }
+
+ public static void stopScriptExecutors(EvalContext evalContext) {
+ for (TajoScriptEngine executor : evalContext.getAllScriptEngines()) {
+ executor.shutdown();
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
index 1d0293b..17af71a 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
@@ -33,7 +33,10 @@ import org.apache.hadoop.yarn.util.RackResolver;
import org.apache.tajo.TajoConstants;
import org.apache.tajo.catalog.CatalogClient;
import org.apache.tajo.catalog.CatalogService;
+import org.apache.tajo.catalog.FunctionDesc;
import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.engine.function.FunctionLoader;
+import org.apache.tajo.function.FunctionSignature;
import org.apache.tajo.service.ServiceTracker;
import org.apache.tajo.service.ServiceTrackerFactory;
import org.apache.tajo.service.TajoMasterInfo;
@@ -51,10 +54,7 @@ import org.apache.tajo.rule.SelfDiagnosisRuleEngine;
import org.apache.tajo.rule.SelfDiagnosisRuleSession;
import org.apache.tajo.storage.HashShuffleAppenderManager;
import org.apache.tajo.storage.StorageManager;
-import org.apache.tajo.util.CommonTestingUtil;
-import org.apache.tajo.util.JvmPauseMonitor;
-import org.apache.tajo.util.NetUtils;
-import org.apache.tajo.util.StringUtils;
+import org.apache.tajo.util.*;
import org.apache.tajo.util.history.HistoryReader;
import org.apache.tajo.util.history.HistoryWriter;
import org.apache.tajo.util.metrics.TajoSystemMetrics;
@@ -65,6 +65,7 @@ import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -235,6 +236,8 @@ public class TajoWorker extends CompositeService {
taskHistoryWriter.init(conf);
historyReader = new HistoryReader(workerContext.getWorkerName(), this.systemConf);
+
+ FunctionLoader.loadUserDefinedFunctions(systemConf, new HashMap<FunctionSignature, FunctionDesc>());
diagnoseTajoWorker();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
index 61a05dc..a983f78 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
@@ -21,7 +21,7 @@ package org.apache.tajo.worker;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
-
+import io.netty.handler.codec.http.QueryStringDecoder;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -30,25 +30,26 @@ import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.tajo.TaskAttemptId;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.TajoProtos.TaskAttemptState;
+import org.apache.tajo.TaskAttemptId;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.conf.TajoConf;
-import org.apache.tajo.master.cluster.WorkerConnectionInfo;
-import org.apache.tajo.plan.serder.LogicalNodeDeserializer;
-import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.engine.planner.physical.PhysicalExec;
import org.apache.tajo.engine.query.QueryContext;
import org.apache.tajo.engine.query.TaskRequest;
import org.apache.tajo.ipc.QueryMasterProtocol;
import org.apache.tajo.ipc.TajoWorkerProtocol.*;
import org.apache.tajo.ipc.TajoWorkerProtocol.EnforceProperty.EnforceType;
+import org.apache.tajo.master.cluster.WorkerConnectionInfo;
+import org.apache.tajo.plan.function.python.TajoScriptEngine;
import org.apache.tajo.plan.logical.*;
+import org.apache.tajo.plan.serder.LogicalNodeDeserializer;
+import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.pullserver.TajoPullServerService;
import org.apache.tajo.pullserver.retriever.FileChunk;
import org.apache.tajo.rpc.NettyClientBase;
@@ -57,8 +58,6 @@ import org.apache.tajo.storage.*;
import org.apache.tajo.storage.fragment.FileFragment;
import org.apache.tajo.util.NetUtils;
-import io.netty.handler.codec.http.QueryStringDecoder;
-
import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
@@ -135,7 +134,7 @@ public class Task {
}
public void initPlan() throws IOException {
- plan = LogicalNodeDeserializer.deserialize(queryContext, request.getPlan());
+ plan = LogicalNodeDeserializer.deserialize(queryContext, context.getEvalContext(), request.getPlan());
LogicalNode [] scanNode = PlannerUtil.findAllNodes(plan, NodeType.SCAN);
if (scanNode != null) {
for (LogicalNode node : scanNode) {
@@ -173,7 +172,7 @@ public class Task {
LOG.info("==================================");
LOG.info("* Stage " + request.getId() + " is initialized");
LOG.info("* InterQuery: " + interQuery
- + (interQuery ? ", Use " + this.shuffleType + " shuffle":"") +
+ + (interQuery ? ", Use " + this.shuffleType + " shuffle" : "") +
", Fragments (num: " + request.getFragments().size() + ")" +
", Fetches (total:" + request.getFetches().size() + ") :");
@@ -190,8 +189,21 @@ public class Task {
LOG.info("==================================");
}
+ private void startScriptExecutors() throws IOException {
+ for (TajoScriptEngine executor : context.getEvalContext().getAllScriptEngines()) {
+ executor.start(systemConf);
+ }
+ }
+
+ private void stopScriptExecutors() {
+ for (TajoScriptEngine executor : context.getEvalContext().getAllScriptEngines()) {
+ executor.shutdown();
+ }
+ }
+
public void init() throws IOException {
initPlan();
+ startScriptExecutors();
if (context.getState() == TaskAttemptState.TA_PENDING) {
// initialize a task temporal dir
@@ -257,11 +269,13 @@ public class Task {
}
public void kill() {
+ stopScriptExecutors();
context.setState(TaskAttemptState.TA_KILLED);
context.stop();
}
public void abort() {
+ stopScriptExecutors();
context.stop();
}
@@ -410,6 +424,7 @@ public class Task {
} catch (Throwable e) {
error = e ;
LOG.error(e.getMessage(), e);
+ stopScriptExecutors();
context.stop();
} finally {
if (executor != null) {
@@ -487,6 +502,7 @@ public class Task {
}
executionBlockContext.getWorkerContext().getTaskHistoryWriter().appendHistory(taskHistory);
+ stopScriptExecutors();
}
public TaskHistory createTaskHistory() {
@@ -630,6 +646,7 @@ public class Task {
if (retryNum == maxRetryNum) {
LOG.error("ERROR: the maximum retry (" + retryNum + ") on the fetch exceeded (" + fetcher.getURI() + ")");
}
+ stopScriptExecutors();
context.stop(); // retry task
ctx.getFetchLatch().countDown();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java b/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
index 706e9b8..58028ac 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
@@ -32,6 +32,7 @@ import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.planner.enforce.Enforcer;
import org.apache.tajo.engine.planner.global.DataChannel;
import org.apache.tajo.engine.query.QueryContext;
+import org.apache.tajo.plan.expr.EvalContext;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.storage.HashShuffleAppenderManager;
import org.apache.tajo.storage.fragment.FileFragment;
@@ -83,6 +84,8 @@ public class TaskAttemptContext {
private Map<Integer, Long> partitionOutputVolume;
private HashShuffleAppenderManager hashShuffleAppenderManager;
+ private EvalContext evalContext = new EvalContext();
+
public TaskAttemptContext(QueryContext queryContext, final ExecutionBlockContext executionBlockContext,
final TaskAttemptId queryId,
final FragmentProto[] fragments,
@@ -403,4 +406,8 @@ public class TaskAttemptContext {
public HashShuffleAppenderManager getHashShuffleAppenderManager() {
return hashShuffleAppenderManager;
}
+
+ public EvalContext getEvalContext() {
+ return evalContext;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/a7453853/tajo-core/src/main/resources/python/__init__.py
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/resources/python/__init__.py b/tajo-core/src/main/resources/python/__init__.py
new file mode 100644
index 0000000..8093a2f
--- /dev/null
+++ b/tajo-core/src/main/resources/python/__init__.py
@@ -0,0 +1,17 @@
+#!/usr/bin/python
+
+############################################################################
+# 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.
\ No newline at end of file