| /* |
| * 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.pig.pen; |
| |
| import java.math.BigDecimal; |
| import java.math.BigInteger; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import org.joda.time.DateTime; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.pig.backend.executionengine.ExecException; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLimit; |
| import org.apache.pig.data.BagFactory; |
| import org.apache.pig.data.DataBag; |
| import org.apache.pig.data.DataByteArray; |
| import org.apache.pig.data.DataType; |
| import org.apache.pig.data.Tuple; |
| import org.apache.pig.data.TupleFactory; |
| import org.apache.pig.impl.io.FileSpec; |
| import org.apache.pig.impl.logicalLayer.FrontendException; |
| import org.apache.pig.impl.util.MultiMap; |
| import org.apache.pig.newplan.Operator; |
| import org.apache.pig.newplan.OperatorPlan; |
| import org.apache.pig.newplan.logical.expression.AddExpression; |
| import org.apache.pig.newplan.logical.expression.AndExpression; |
| import org.apache.pig.newplan.logical.expression.BinaryExpression; |
| import org.apache.pig.newplan.logical.expression.CastExpression; |
| import org.apache.pig.newplan.logical.expression.ConstantExpression; |
| import org.apache.pig.newplan.logical.expression.DivideExpression; |
| import org.apache.pig.newplan.logical.expression.EqualExpression; |
| import org.apache.pig.newplan.logical.expression.GreaterThanEqualExpression; |
| import org.apache.pig.newplan.logical.expression.GreaterThanExpression; |
| import org.apache.pig.newplan.logical.expression.IsNullExpression; |
| import org.apache.pig.newplan.logical.expression.LessThanEqualExpression; |
| import org.apache.pig.newplan.logical.expression.LessThanExpression; |
| import org.apache.pig.newplan.logical.expression.LogicalExpression; |
| import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan; |
| import org.apache.pig.newplan.logical.expression.ModExpression; |
| import org.apache.pig.newplan.logical.expression.MultiplyExpression; |
| import org.apache.pig.newplan.logical.expression.NotEqualExpression; |
| import org.apache.pig.newplan.logical.expression.NotExpression; |
| import org.apache.pig.newplan.logical.expression.OrExpression; |
| import org.apache.pig.newplan.logical.expression.ProjectExpression; |
| import org.apache.pig.newplan.logical.expression.RegexExpression; |
| import org.apache.pig.newplan.logical.expression.SubtractExpression; |
| import org.apache.pig.newplan.logical.expression.UserFuncExpression; |
| import org.apache.pig.newplan.logical.relational.LOCogroup; |
| import org.apache.pig.newplan.logical.relational.LOCross; |
| import org.apache.pig.newplan.logical.relational.LODistinct; |
| import org.apache.pig.newplan.logical.relational.LOFilter; |
| import org.apache.pig.newplan.logical.relational.LOForEach; |
| import org.apache.pig.newplan.logical.relational.LOJoin; |
| import org.apache.pig.newplan.logical.relational.LOLimit; |
| import org.apache.pig.newplan.logical.relational.LOLoad; |
| import org.apache.pig.newplan.logical.relational.LOSort; |
| import org.apache.pig.newplan.logical.relational.LOSplit; |
| import org.apache.pig.newplan.logical.relational.LOStore; |
| import org.apache.pig.newplan.logical.relational.LOUnion; |
| import org.apache.pig.newplan.logical.relational.LogicalPlan; |
| import org.apache.pig.newplan.logical.relational.LogicalRelationalNodesVisitor; |
| import org.apache.pig.newplan.logical.relational.LogicalRelationalOperator; |
| import org.apache.pig.newplan.logical.relational.LogicalSchema; |
| import org.apache.pig.pen.util.ExampleTuple; |
| import org.apache.pig.pen.util.PreOrderDepthFirstWalker; |
| |
| //This is used to generate synthetic data |
| //Synthetic data generation is done by making constraint tuples for each operator as we traverse the plan |
| //and try to replace the constraints with values as far as possible. We only deal with simple conditions right now |
| |
| public class AugmentBaseDataVisitor extends LogicalRelationalNodesVisitor { |
| |
| Map<LOLoad, DataBag> baseData = null; |
| Map<LOLoad, DataBag> newBaseData = new HashMap<LOLoad, DataBag>(); |
| Map<Operator, DataBag> derivedData = null; |
| private boolean limit = false; |
| private final Map<Operator, PhysicalOperator> logToPhysMap; |
| private Map<LOLimit, Long> oriLimitMap; |
| |
| Map<Operator, DataBag> outputConstraintsMap = new HashMap<Operator, DataBag>(); |
| |
| Log log = LogFactory.getLog(getClass()); |
| |
| // Augmentation moves from the leaves to root and hence needs a |
| // depthfirstwalker |
| public AugmentBaseDataVisitor(OperatorPlan plan, |
| Map<Operator, PhysicalOperator> logToPhysMap, |
| Map<LOLoad, DataBag> baseData, |
| Map<Operator, DataBag> derivedData) throws FrontendException { |
| super(plan, new PreOrderDepthFirstWalker( |
| plan)); |
| this.baseData = baseData; |
| this.derivedData = derivedData; |
| this.logToPhysMap = logToPhysMap; |
| } |
| |
| public void setLimit() { |
| limit = true; |
| } |
| |
| public Map<LOLoad, DataBag> getNewBaseData() throws ExecException { |
| // consolidate base data from different LOADs on the same inputs |
| MultiMap<FileSpec, DataBag> inputDataMap = new MultiMap<FileSpec, DataBag>(); |
| for (Map.Entry<LOLoad, DataBag> e : newBaseData.entrySet()) { |
| inputDataMap.put(e.getKey().getFileSpec(), e.getValue()); |
| } |
| |
| int index = 0; |
| for (FileSpec fs : inputDataMap.keySet()) { |
| int maxSchemaSize = 0; |
| Tuple tupleOfMaxSchemaSize = null; |
| for (DataBag bag : inputDataMap.get(fs)) { |
| if (bag.size() > 0) { |
| int size = 0; |
| Tuple t = null; |
| t = bag.iterator().next(); |
| size = t.size(); |
| if (size > maxSchemaSize) { |
| maxSchemaSize = size; |
| tupleOfMaxSchemaSize = t; |
| } |
| } |
| } |
| for (DataBag bag : inputDataMap.get(fs)) { |
| if (bag.size() > 0) { |
| for (Iterator<Tuple> it = bag.iterator(); it.hasNext();) { |
| Tuple t = it.next(); |
| for (int i = t.size(); i < maxSchemaSize; ++i) { |
| t.append(tupleOfMaxSchemaSize.get(i)); |
| } |
| } |
| } |
| } |
| index++; |
| } |
| |
| |
| for (Map.Entry<LOLoad, DataBag> e : baseData.entrySet()) { |
| DataBag bag = newBaseData.get(e.getKey()); |
| if (bag == null) { |
| bag = BagFactory.getInstance().newDefaultBag(); |
| newBaseData.put(e.getKey(), bag); |
| } |
| bag.addAll(e.getValue()); |
| } |
| return newBaseData; |
| } |
| |
| public Map<LOLimit, Long> getOriLimitMap() { |
| return oriLimitMap; |
| } |
| |
| @Override |
| public void visit(LOCogroup cg) throws FrontendException { |
| if (limit && !((PreOrderDepthFirstWalker) currentWalker).getBranchFlag()) |
| return; |
| // we first get the outputconstraints for the current cogroup |
| DataBag outputConstraints = outputConstraintsMap.get(cg); |
| outputConstraintsMap.remove(cg); |
| boolean ableToHandle = true; |
| // we then check if we can handle this cogroup and try to collect some |
| // information about grouping |
| List<List<Integer>> groupSpecs = new LinkedList<List<Integer>>(); |
| int numCols = -1; |
| |
| for (int index = 0; index < cg.getInputs((LogicalPlan)plan).size(); ++index) { |
| Collection<LogicalExpressionPlan> groupByPlans = |
| cg.getExpressionPlans().get(index); |
| List<Integer> groupCols = new ArrayList<Integer>(); |
| for (LogicalExpressionPlan plan : groupByPlans) { |
| Operator leaf = plan.getSinks().get(0); |
| if (leaf instanceof ProjectExpression) { |
| groupCols.add(Integer.valueOf(((ProjectExpression) leaf).getColNum())); |
| } else { |
| ableToHandle = false; |
| break; |
| } |
| } |
| if (numCols == -1) { |
| numCols = groupCols.size(); |
| } |
| if (groupCols.size() != groupByPlans.size() |
| || groupCols.size() != numCols) { |
| // we came across an unworkable cogroup plan |
| break; |
| } else { |
| groupSpecs.add(groupCols); |
| } |
| } |
| |
| // we should now have some workable data at this point to synthesize |
| // tuples |
| try { |
| if (ableToHandle) { |
| // we need to go through the output constraints first |
| int numInputs = cg.getInputs((LogicalPlan) plan).size(); |
| if (outputConstraints != null) { |
| for (Iterator<Tuple> it = outputConstraints.iterator(); it |
| .hasNext();) { |
| Tuple outputConstraint = it.next(); |
| Object groupLabel = outputConstraint.get(0); |
| |
| for (int input = 0; input < numInputs; input++) { |
| |
| int numInputFields = ((LogicalRelationalOperator) cg.getInputs((LogicalPlan) plan).get(input)) |
| .getSchema().size(); |
| List<Integer> groupCols = groupSpecs.get(input); |
| |
| DataBag output = outputConstraintsMap.get(cg |
| .getInputs((LogicalPlan) plan).get(input)); |
| if (output == null) { |
| output = BagFactory.getInstance() |
| .newDefaultBag(); |
| outputConstraintsMap.put(cg.getInputs((LogicalPlan) plan).get( |
| input), output); |
| } |
| for (int i = 0; i < 2; i++) { |
| Tuple inputConstraint = GetGroupByInput( |
| groupLabel, groupCols, numInputFields); |
| if (inputConstraint != null) |
| output.add(inputConstraint); |
| } |
| } |
| } |
| } |
| // then, go through all organic data groups and add input |
| // constraints to make each group big enough |
| DataBag outputData = derivedData.get(cg); |
| |
| for (Iterator<Tuple> it = outputData.iterator(); it.hasNext();) { |
| Tuple groupTup = it.next(); |
| Object groupLabel = groupTup.get(0); |
| |
| for (int input = 0; input < numInputs; input++) { |
| int numInputFields = ((LogicalRelationalOperator)cg.getInputs((LogicalPlan) plan).get(input)) |
| .getSchema().size(); |
| List<Integer> groupCols = groupSpecs.get(input); |
| |
| DataBag output = outputConstraintsMap.get(cg |
| .getInputs((LogicalPlan) plan).get(input)); |
| if (output == null) { |
| output = BagFactory.getInstance().newDefaultBag(); |
| outputConstraintsMap.put(cg.getInputs((LogicalPlan) plan).get(input), |
| output); |
| } |
| int numTupsToAdd = 2 |
| - (int) ((DataBag) groupTup.get(input + 1)) |
| .size(); |
| for (int i = 0; i < numTupsToAdd; i++) { |
| Tuple inputConstraint = GetGroupByInput(groupLabel, |
| groupCols, numInputFields); |
| if (inputConstraint != null) |
| output.add(inputConstraint); |
| } |
| } |
| } |
| } |
| } catch (Exception e) { |
| log |
| .error("Error visiting Cogroup during Augmentation phase of Example Generator! " |
| + e.getMessage()); |
| throw new FrontendException( |
| "Error visiting Cogroup during Augmentation phase of Example Generator! " |
| + e.getMessage()); |
| } |
| } |
| |
| @Override |
| public void visit(LOJoin join) throws FrontendException { |
| if (limit && !((PreOrderDepthFirstWalker) currentWalker).getBranchFlag()) |
| return; |
| // we first get the outputconstraints for the current cogroup |
| DataBag outputConstraints = outputConstraintsMap.get(join); |
| outputConstraintsMap.remove(join); |
| boolean ableToHandle = true; |
| // we then check if we can handle this cogroup and try to collect some |
| // information about grouping |
| List<List<Integer>> groupSpecs = new LinkedList<List<Integer>>(); |
| int numCols = -1; |
| |
| for (int index = 0; index < join.getInputs((LogicalPlan)plan).size(); ++index) { |
| Collection<LogicalExpressionPlan> groupByPlans = |
| join.getExpressionPlans().get(index); |
| List<Integer> groupCols = new ArrayList<Integer>(); |
| for (LogicalExpressionPlan plan : groupByPlans) { |
| Operator leaf = plan.getSinks().get(0); |
| if (leaf instanceof ProjectExpression) { |
| groupCols.add(Integer.valueOf(((ProjectExpression) leaf).getColNum())); |
| } else { |
| ableToHandle = false; |
| break; |
| } |
| } |
| if (numCols == -1) { |
| numCols = groupCols.size(); |
| } |
| if (groupCols.size() != groupByPlans.size() |
| || groupCols.size() != numCols) { |
| // we came across an unworkable cogroup plan |
| break; |
| } else { |
| groupSpecs.add(groupCols); |
| } |
| } |
| |
| // we should now have some workable data at this point to synthesize |
| // tuples |
| try { |
| if (ableToHandle) { |
| // we need to go through the output constraints first |
| int numInputs = join.getInputs((LogicalPlan) plan).size(); |
| if (outputConstraints != null) { |
| for (Iterator<Tuple> it = outputConstraints.iterator(); it |
| .hasNext();) { |
| Tuple outputConstraint = it.next(); |
| |
| for (int input = 0; input < numInputs; input++) { |
| |
| int numInputFields = ((LogicalRelationalOperator) join.getInputs((LogicalPlan) plan).get(input)) |
| .getSchema().size(); |
| List<Integer> groupCols = groupSpecs.get(input); |
| |
| DataBag output = outputConstraintsMap.get(join |
| .getInputs((LogicalPlan) plan).get(input)); |
| if (output == null) { |
| output = BagFactory.getInstance() |
| .newDefaultBag(); |
| outputConstraintsMap.put(join.getInputs((LogicalPlan) plan).get( |
| input), output); |
| } |
| |
| Tuple inputConstraint = GetJoinInput( |
| outputConstraint, groupCols, numInputFields); |
| if (inputConstraint != null) |
| output.add(inputConstraint); |
| } |
| } |
| } |
| // then, go through all organic data groups and add input |
| // constraints to make each group big enough |
| DataBag outputData = derivedData.get(join); |
| |
| if (outputData.size() == 0) { |
| DataBag output0 = outputConstraintsMap.get(join.getInputs((LogicalPlan) plan).get(0)); |
| if (output0 == null || output0.size() == 0) { |
| output0 = derivedData.get(join.getInputs((LogicalPlan) plan).get(0)); |
| } |
| Tuple inputConstraint0 = output0.iterator().next(); |
| for (int input = 1; input < numInputs; input++) { |
| DataBag output = outputConstraintsMap.get(join.getInputs((LogicalPlan) plan).get(input)); |
| if (output == null) |
| { |
| output = BagFactory.getInstance().newDefaultBag(); |
| outputConstraintsMap.put(join.getInputs((LogicalPlan) plan).get(input), |
| output); |
| } |
| int numInputFields = ((LogicalRelationalOperator)join.getInputs((LogicalPlan) plan).get(input)).getSchema().size(); |
| Tuple inputConstraint = GetJoinInput(inputConstraint0, groupSpecs.get(0), groupSpecs.get(input), numInputFields); |
| if (inputConstraint != null) |
| output.add(inputConstraint); |
| } |
| } |
| } |
| } catch (Exception e) { |
| log |
| .error("Error visiting Cogroup during Augmentation phase of Example Generator! " |
| + e.getMessage()); |
| throw new FrontendException( |
| "Error visiting Cogroup during Augmentation phase of Example Generator! " |
| + e.getMessage()); |
| } |
| } |
| |
| @Override |
| public void visit(LOCross cs) throws FrontendException { |
| |
| } |
| |
| @Override |
| public void visit(LODistinct dt) throws FrontendException { |
| if (limit && !((PreOrderDepthFirstWalker) currentWalker).getBranchFlag()) |
| return; |
| |
| DataBag outputConstraints = outputConstraintsMap.get(dt); |
| outputConstraintsMap.remove(dt); |
| |
| DataBag inputConstraints = outputConstraintsMap.get(dt.getInput((LogicalPlan) plan)); |
| if (inputConstraints == null) { |
| inputConstraints = BagFactory.getInstance().newDefaultBag(); |
| outputConstraintsMap.put(dt.getInput((LogicalPlan) plan), inputConstraints); |
| } |
| |
| if (outputConstraints != null && outputConstraints.size() > 0) { |
| for (Iterator<Tuple> it = outputConstraints.iterator(); it.hasNext();) |
| { |
| inputConstraints.add(it.next()); |
| } |
| } |
| |
| boolean emptyInputConstraints = inputConstraints.size() == 0; |
| if (emptyInputConstraints) { |
| DataBag inputData = derivedData.get(dt.getInput((LogicalPlan) plan)); |
| for (Iterator<Tuple> it = inputData.iterator(); it.hasNext();) |
| { |
| inputConstraints.add(it.next()); |
| } |
| } |
| Set<Tuple> distinctSet = new HashSet<Tuple>(); |
| Iterator<Tuple> it; |
| for (it = inputConstraints.iterator(); it.hasNext();) { |
| if (!distinctSet.add(it.next())) |
| break; |
| } |
| if (!it.hasNext()) |
| { |
| // no duplicates found: generate one |
| if (inputConstraints.size()> 0) { |
| Tuple src = ((ExampleTuple)inputConstraints.iterator().next()).toTuple(), |
| tgt = TupleFactory.getInstance().newTuple(src.getAll()); |
| ExampleTuple inputConstraint = new ExampleTuple(tgt); |
| inputConstraint.synthetic = true; |
| inputConstraints.add(inputConstraint); |
| } else if (emptyInputConstraints) |
| inputConstraints.clear(); |
| } |
| } |
| |
| @Override |
| public void visit(LOFilter filter) throws FrontendException { |
| if (limit && !((PreOrderDepthFirstWalker) currentWalker).getBranchFlag()) |
| return; |
| |
| DataBag outputConstraints = outputConstraintsMap.get(filter); |
| outputConstraintsMap.remove(filter); |
| |
| LogicalExpressionPlan filterCond = filter.getFilterPlan(); |
| DataBag inputConstraints = outputConstraintsMap.get(filter.getInput((LogicalPlan) plan)); |
| if (inputConstraints == null) { |
| inputConstraints = BagFactory.getInstance().newDefaultBag(); |
| outputConstraintsMap.put(filter.getInput((LogicalPlan) plan), inputConstraints); |
| } |
| |
| DataBag outputData = derivedData.get(filter); |
| DataBag inputData = derivedData.get(filter.getInput((LogicalPlan) plan)); |
| try { |
| if (outputConstraints != null && outputConstraints.size() > 0) { // there |
| // 's |
| // one |
| // or |
| // more |
| // output |
| // constraints |
| // ; |
| // generate |
| // corresponding |
| // input |
| // constraints |
| for (Iterator<Tuple> it = outputConstraints.iterator(); it |
| .hasNext();) { |
| Tuple outputConstraint = it.next(); |
| ExampleTuple inputConstraint = GenerateMatchingTuple( |
| outputConstraint, filterCond, false); |
| if (inputConstraint != null) |
| inputConstraints.add(inputConstraint); |
| } |
| } else if (outputData.size() == 0) { // no output constraints, but |
| // output is empty; generate |
| // one input that will pass the |
| // filter |
| ExampleTuple inputConstraint = GenerateMatchingTuple(filter |
| .getSchema(), filterCond, false); |
| |
| if (inputConstraint != null) |
| inputConstraints.add(inputConstraint); |
| } |
| |
| // if necessary, insert a negative example (i.e. a tuple that does |
| // not pass the filter) |
| if (outputData.size() == inputData.size()) { // all tuples pass the |
| // filter; generate one |
| // input that will not |
| // pass the filter |
| |
| ExampleTuple inputConstraint = GenerateMatchingTuple(filter |
| .getSchema(), filterCond, true); |
| if (inputConstraint != null) |
| inputConstraints.add(inputConstraint); |
| |
| } |
| } catch (Exception e) { |
| log |
| .error("Error visiting Load during Augmentation phase of Example Generator! " |
| + e.getMessage(), e); |
| throw new FrontendException( |
| "Error visiting Load during Augmentation phase of Example Generator! " |
| + e.getMessage(), e); |
| } |
| } |
| |
| @Override |
| public void visit(LOForEach forEach) throws FrontendException { |
| if (limit && !((PreOrderDepthFirstWalker) currentWalker).getBranchFlag()) |
| return; |
| DataBag outputConstraints = outputConstraintsMap.get(forEach); |
| outputConstraintsMap.remove(forEach); |
| LogicalPlan plan = forEach.getInnerPlan(); |
| boolean ableToHandle = true; |
| List<Integer> cols = new ArrayList<Integer>(); |
| boolean cast = false; |
| |
| if (outputConstraints == null || outputConstraints.size() == 0) |
| // we dont have to do anything in this case |
| return; |
| |
| |
| Operator op = plan.getSinks().get(0); |
| if (op instanceof CastExpression) { |
| cast = true; |
| op = ((CastExpression) op).getExpression(); |
| } |
| |
| if (!(op instanceof ProjectExpression)) { |
| ableToHandle = false; |
| } else { |
| cols.add(Integer.valueOf(((ProjectExpression) op).getColNum())); |
| } |
| |
| if (ableToHandle) { |
| // we can only handle simple projections |
| DataBag output = BagFactory.getInstance().newDefaultBag(); |
| for (Iterator<Tuple> it = outputConstraints.iterator(); it |
| .hasNext();) { |
| Tuple outputConstraint = it.next(); |
| try { |
| Tuple inputConstraint = BackPropConstraint( |
| outputConstraint, cols, ((LogicalRelationalOperator)plan |
| .getPredecessors(forEach).get(0)) |
| .getSchema(), cast); |
| output.add(inputConstraint); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| throw new FrontendException( |
| "Operator error during Augmenting Phase in Example Generator " |
| + e.getMessage()); |
| } |
| } |
| outputConstraintsMap.put(plan.getPredecessors(forEach) |
| .get(0), output); |
| } |
| |
| } |
| |
| @Override |
| public void visit(LOLoad load) throws FrontendException { |
| DataBag inputData = baseData.get(load); |
| // check if the inputData exists |
| if (inputData == null || inputData.size() == 0) { |
| log.error("No (valid) input data found!"); |
| throw new RuntimeException("No (valid) input data found!"); |
| } |
| |
| DataBag newInputData = newBaseData.get(load); |
| if (newInputData == null) { |
| newInputData = BagFactory.getInstance().newDefaultBag(); |
| newBaseData.put(load, newInputData); |
| } |
| |
| LogicalSchema schema; |
| try { |
| schema = load.getSchema(); |
| if (schema == null) |
| throw new RuntimeException( |
| "Example Generator requires a schema. Please provide a schema while loading data"); |
| } catch (FrontendException e) { |
| |
| log |
| .error("Error visiting Load during Augmentation phase of Example Generator! " |
| + e.getMessage()); |
| throw new FrontendException( |
| "Error visiting Load during Augmentation phase of Example Generator! " |
| + e.getMessage()); |
| } |
| |
| Tuple exampleTuple = inputData.iterator().next(); |
| |
| DataBag outputConstraints = outputConstraintsMap.get(load); |
| outputConstraintsMap.remove(load); |
| |
| // first of all, we are required to guarantee that there is at least one |
| // output tuple |
| if (outputConstraints == null || outputConstraints.size() == 0) { |
| outputConstraints = BagFactory.getInstance().newDefaultBag(); |
| outputConstraints.add(TupleFactory.getInstance().newTuple( |
| schema.getFields().size())); |
| } |
| |
| // create example tuple to steal values from when we encounter |
| // "don't care" fields (i.e. null fields) |
| System.out.println(exampleTuple.toString()); |
| |
| // run through output constraints; for each one synthesize a tuple and |
| // add it to the base data |
| // (while synthesizing individual fields, try to match fields that exist |
| // in the real data) |
| boolean newInput = false; |
| for (Iterator<Tuple> it = outputConstraints.iterator(); it.hasNext();) { |
| Tuple outputConstraint = it.next(); |
| |
| // sanity check: |
| if (outputConstraint.size() != schema.getFields().size()) |
| throw new RuntimeException( |
| "Internal error: incorrect number of fields in constraint tuple."); |
| |
| Tuple inputT = TupleFactory.getInstance().newTuple( |
| outputConstraint.size()); |
| ExampleTuple inputTuple = new ExampleTuple(inputT); |
| |
| try { |
| for (int i = 0; i < inputTuple.size(); i++) { |
| Object d = outputConstraint.get(i); |
| if (d == null && i < exampleTuple.size()) |
| d = exampleTuple.get(i); |
| inputTuple.set(i, d); |
| } |
| if (outputConstraint instanceof ExampleTuple) |
| inputTuple.synthetic = ((ExampleTuple) outputConstraint).synthetic; |
| else |
| // raw tuple should have been synthesized |
| inputTuple.synthetic = true; |
| } catch (ExecException e) { |
| log |
| .error("Error visiting Load during Augmentation phase of Example Generator! " |
| + e.getMessage()); |
| throw new FrontendException( |
| "Error visiting Load during Augmentation phase of Example Generator! " |
| + e.getMessage()); |
| |
| } |
| try { |
| if (inputTuple.synthetic || !inInput(inputTuple, inputData, schema)) |
| { |
| inputTuple.synthetic = true; |
| |
| newInputData.add(inputTuple); |
| |
| if (!newInput) |
| newInput = true; |
| } |
| } catch (ExecException e) { |
| throw new FrontendException( |
| "Error visiting Load during Augmentation phase of Example Generator! " |
| + e.getMessage()); |
| } |
| } |
| } |
| |
| private boolean inInput(Tuple newTuple, DataBag input, LogicalSchema schema) throws ExecException { |
| boolean result; |
| for (Iterator<Tuple> iter = input.iterator(); iter.hasNext();) { |
| result = true; |
| Tuple tmp = iter.next(); |
| for (int i = 0; i < schema.size(); ++i) |
| if (!newTuple.get(i).equals(tmp.get(i))) |
| { |
| result = false; |
| break; |
| } |
| if (result) |
| return true; |
| } |
| return false; |
| } |
| |
| @Override |
| public void visit(LOSort s) throws FrontendException { |
| if (limit && !((PreOrderDepthFirstWalker) currentWalker).getBranchFlag()) |
| return; |
| DataBag outputConstraints = outputConstraintsMap.get(s); |
| outputConstraintsMap.remove(s); |
| |
| if (outputConstraints == null) |
| outputConstraintsMap.put(s.getInput((LogicalPlan) plan), BagFactory.getInstance() |
| .newDefaultBag()); |
| else |
| outputConstraintsMap.put(s.getInput((LogicalPlan) plan), outputConstraints); |
| } |
| |
| @Override |
| public void visit(LOSplit split) throws FrontendException { |
| if (limit && !((PreOrderDepthFirstWalker) currentWalker).getBranchFlag()) |
| return; |
| } |
| |
| @Override |
| public void visit(LOStore store) throws FrontendException { |
| if (limit && !((PreOrderDepthFirstWalker) currentWalker).getBranchFlag()) |
| return; |
| DataBag outputConstraints = outputConstraintsMap.get(store); |
| if (outputConstraints == null) { |
| outputConstraintsMap.put(plan.getPredecessors(store) |
| .get(0), BagFactory.getInstance().newDefaultBag()); |
| } else { |
| outputConstraintsMap.remove(store); |
| outputConstraintsMap.put(plan.getPredecessors(store) |
| .get(0), outputConstraints); |
| } |
| } |
| |
| @Override |
| public void visit(LOUnion u) throws FrontendException { |
| if (limit && !((PreOrderDepthFirstWalker) currentWalker).getBranchFlag()) |
| return; |
| DataBag outputConstraints = outputConstraintsMap.get(u); |
| outputConstraintsMap.remove(u); |
| if (outputConstraints == null || outputConstraints.size() == 0) { |
| // we dont need to do anything |
| // we just find the inputs, create empty bags as their |
| // outputConstraints and return |
| for (Operator op : u.getInputs((LogicalPlan) plan)) { |
| DataBag constraints = BagFactory.getInstance().newDefaultBag(); |
| outputConstraintsMap.put(op, constraints); |
| } |
| return; |
| } |
| |
| // since we have some outputConstraints, we apply them to the inputs |
| // round-robin |
| int count = 0; |
| List<Operator> inputs = u.getInputs(((LogicalPlan) plan)); |
| int noInputs = inputs.size(); |
| |
| for (Operator op : inputs) { |
| DataBag constraint = BagFactory.getInstance().newDefaultBag(); |
| outputConstraintsMap.put(op, constraint); |
| } |
| for (Iterator<Tuple> it = outputConstraints.iterator(); it.hasNext();) { |
| DataBag constraint = outputConstraintsMap.get(inputs.get(count)); |
| constraint.add(it.next()); |
| count = (count + 1) % noInputs; |
| } |
| |
| } |
| |
| @Override |
| public void visit(LOLimit lm) throws FrontendException { |
| if (!limit) // not augment for LIMIT in this traversal |
| return; |
| |
| if (oriLimitMap == null) |
| oriLimitMap = new HashMap<LOLimit, Long>(); |
| |
| DataBag outputConstraints = outputConstraintsMap.get(lm); |
| outputConstraintsMap.remove(lm); |
| |
| DataBag inputConstraints = outputConstraintsMap.get(lm.getInput((LogicalPlan) plan)); |
| if (inputConstraints == null) { |
| inputConstraints = BagFactory.getInstance().newDefaultBag(); |
| outputConstraintsMap.put(lm.getInput((LogicalPlan) plan), inputConstraints); |
| } |
| |
| DataBag inputData = derivedData.get(lm.getInput((LogicalPlan) plan)); |
| |
| if (outputConstraints != null && outputConstraints.size() > 0) { // there |
| // 's |
| // one |
| // or |
| // more |
| // output |
| // constraints |
| // ; |
| // generate |
| // corresponding |
| // input |
| // constraints |
| for (Iterator<Tuple> it = outputConstraints.iterator(); it |
| .hasNext();) { |
| inputConstraints.add(it.next()); |
| // ... plus one more if only one |
| if (inputConstraints.size() == 1) { |
| inputConstraints.add(inputData.iterator().next()); |
| ((PreOrderDepthFirstWalker) currentWalker).setBranchFlag(); |
| } |
| } |
| } else if (inputConstraints.size() == 0){ |
| // add all input to input constraints ... |
| inputConstraints.addAll(inputData); |
| // ... plus one more if only one |
| if (inputConstraints.size() == 1) { |
| inputConstraints.add(inputData.iterator().next()); |
| ((PreOrderDepthFirstWalker) currentWalker).setBranchFlag(); |
| } |
| } |
| POLimit poLimit = (POLimit) logToPhysMap.get(lm); |
| oriLimitMap.put(lm, Long.valueOf(poLimit.getLimit())); |
| poLimit.setLimit(inputConstraints.size()-1); |
| lm.setLimit(poLimit.getLimit()); |
| } |
| |
| Tuple GetGroupByInput(Object groupLabel, List<Integer> groupCols, |
| int numFields) throws ExecException { |
| Tuple t = TupleFactory.getInstance().newTuple(numFields); |
| |
| if (groupCols.size() == 1) { |
| // GroupLabel would be a data atom |
| t.set(groupCols.get(0), groupLabel); |
| } else { |
| if (!(groupLabel instanceof Tuple)) |
| throw new RuntimeException("Unrecognized group label!"); |
| Tuple group = (Tuple) groupLabel; |
| for (int i = 0; i < groupCols.size(); i++) { |
| t.set(groupCols.get(i), group.get(i)); |
| } |
| } |
| |
| return t; |
| } |
| |
| Tuple GetJoinInput(Tuple group, List<Integer> groupCols0, List<Integer> groupCols, |
| int numFields) throws ExecException { |
| Tuple t = TupleFactory.getInstance().newTuple(numFields); |
| |
| if (groupCols.size() == 1) { |
| // GroupLabel would be a data atom |
| t.set(groupCols.get(0), group.get(groupCols0.get(0))); |
| } else { |
| if (!(group instanceof Tuple)) |
| throw new RuntimeException("Unrecognized group label!"); |
| for (int i = 0; i < groupCols.size(); i++) { |
| t.set(groupCols.get(i), group.get(groupCols0.get(i))); |
| } |
| } |
| |
| return t; |
| } |
| |
| Tuple GetJoinInput(Tuple group, List<Integer> groupCols, |
| int numFields) throws ExecException { |
| Tuple t = TupleFactory.getInstance().newTuple(numFields); |
| |
| if (groupCols.size() == 1) { |
| // GroupLabel would be a data atom |
| t.set(groupCols.get(0), group); |
| } else { |
| if (!(group instanceof Tuple)) |
| throw new RuntimeException("Unrecognized group label!"); |
| for (int i = 0; i < groupCols.size(); i++) { |
| t.set(groupCols.get(i), group.get(i)); |
| } |
| } |
| |
| return t; |
| } |
| |
| Tuple BackPropConstraint(Tuple outputConstraint, List<Integer> cols, |
| LogicalSchema inputSchema, boolean cast) throws ExecException { |
| Tuple inputConst = TupleFactory.getInstance().newTuple( |
| inputSchema.getFields().size()); |
| |
| Tuple inputConstraint = new ExampleTuple(inputConst); |
| |
| for (int outCol = 0; outCol < outputConstraint.size(); outCol++) { |
| int inCol = cols.get(outCol); |
| Object outVal = outputConstraint.get(outCol); |
| Object inVal = inputConstraint.get(inCol); |
| |
| if (inVal == null && outVal != null) { |
| // inputConstraint.set(inCol, outVal); |
| inputConstraint.set(inCol, (cast) ? new DataByteArray(outVal |
| .toString().getBytes()) : outVal); |
| |
| } else { |
| if (outVal != null) { |
| // unable to back-propagate, due to conflicting column |
| // constraints, so give up |
| return null; |
| } |
| } |
| } |
| |
| return inputConstraint; |
| } |
| |
| // generate a constraint tuple that conforms to the schema and passes the |
| // predicate |
| // (or null if unable to find such a tuple) |
| |
| ExampleTuple GenerateMatchingTuple(LogicalSchema schema, LogicalExpressionPlan plan, |
| boolean invert) throws FrontendException, ExecException { |
| return GenerateMatchingTuple(TupleFactory.getInstance().newTuple( |
| schema.getFields().size()), plan, invert); |
| } |
| |
| // generate a constraint tuple that conforms to the constraint and passes |
| // the predicate |
| // (or null if unable to find such a tuple) |
| // |
| // for now, constraint tuples are tuples whose fields are a blend of actual |
| // data values and nulls, |
| // where a null stands for "don't care" |
| // |
| // in the future, may want to replace "don't care" with a more rich |
| // constraint language; this would |
| // help, e.g. in the case of two filters in a row (you want the downstream |
| // filter to tell the upstream filter |
| // what predicate it wants satisfied in a given field) |
| // |
| |
| ExampleTuple GenerateMatchingTuple(Tuple constraint, LogicalExpressionPlan predicate, |
| boolean invert) throws ExecException, FrontendException { |
| Tuple t = TupleFactory.getInstance().newTuple(constraint.size()); |
| ExampleTuple tOut = new ExampleTuple(t); |
| for (int i = 0; i < t.size(); i++) |
| tOut.set(i, constraint.get(i)); |
| |
| GenerateMatchingTupleHelper(tOut, predicate |
| .getSources().get(0), invert); |
| tOut.synthetic = true; |
| return tOut; |
| |
| } |
| |
| void GenerateMatchingTupleHelper(Tuple t, Operator pred, |
| boolean invert) throws FrontendException, ExecException { |
| if (pred instanceof BinaryExpression) |
| GenerateMatchingTupleHelper(t, (BinaryExpression) pred, |
| invert); |
| else if (pred instanceof NotExpression) |
| GenerateMatchingTupleHelper(t, (NotExpression) pred, invert); |
| else if (pred instanceof IsNullExpression) |
| GenerateMatchingTupleHelper(t, (IsNullExpression) pred, invert); |
| else if (pred instanceof UserFuncExpression) |
| // Don't know how to generate input tuple for UDF, return null |
| // to suppress the generation |
| t = null; |
| else |
| throw new FrontendException("Unknown operator in filter predicate"); |
| } |
| |
| void GenerateMatchingTupleHelper(Tuple t, BinaryExpression pred, |
| boolean invert) throws FrontendException, ExecException { |
| |
| if (pred instanceof AndExpression) { |
| GenerateMatchingTupleHelper(t, (AndExpression) pred, invert); |
| return; |
| } else if (pred instanceof OrExpression) { |
| GenerateMatchingTupleHelper(t, (OrExpression) pred, invert); |
| return; |
| } |
| |
| // now we are sure that the expression operators are the roots of the |
| // plan |
| |
| boolean leftIsConst = false, rightIsConst = false; |
| Object leftConst = null, rightConst = null; |
| byte leftDataType = 0, rightDataType = 0; |
| |
| int leftCol = -1, rightCol = -1; |
| |
| if (pred instanceof AddExpression || pred instanceof SubtractExpression |
| || pred instanceof MultiplyExpression || pred instanceof DivideExpression |
| || pred instanceof ModExpression || pred instanceof RegexExpression) |
| return; // We don't try to work around these operators right now |
| |
| if (pred.getLhs() instanceof ConstantExpression) { |
| leftIsConst = true; |
| leftConst = ((ConstantExpression) (pred.getLhs())).getValue(); |
| } else { |
| LogicalExpression lhs = pred.getLhs(); |
| if (lhs instanceof CastExpression) |
| lhs = ((CastExpression) lhs).getExpression(); |
| // if (!(pred.getLhsOperand() instanceof ProjectExpression && ((ProjectExpression) |
| // pred |
| // .getLhsOperand()).getProjection().size() == 1)) |
| // return; // too hard |
| if (!(lhs instanceof ProjectExpression)) |
| return; |
| leftCol = ((ProjectExpression) lhs).getColNum(); |
| leftDataType = ((ProjectExpression) lhs).getType(); |
| |
| Object d = t.get(leftCol); |
| if (d != null) { |
| leftIsConst = true; |
| leftConst = d; |
| } |
| } |
| |
| if (pred.getRhs() instanceof ConstantExpression) { |
| rightIsConst = true; |
| rightConst = ((ConstantExpression) (pred.getRhs())).getValue(); |
| } else { |
| Operator rhs = pred.getRhs(); |
| if (rhs instanceof CastExpression) |
| rhs = ((CastExpression) rhs).getExpression(); |
| // if (!(pred.getRhsOperand() instanceof ProjectExpression && ((ProjectExpression) |
| // pred |
| // .getRhsOperand()).getProjection().size() == 1)) |
| // return; // too hard |
| if (!(rhs instanceof ProjectExpression)) |
| return; |
| rightCol = ((ProjectExpression) rhs).getColNum(); |
| rightDataType = ((ProjectExpression) rhs).getType(); |
| |
| Object d = t.get(rightCol); |
| if (d != null) { |
| rightIsConst = true; |
| rightConst = d; |
| } |
| } |
| |
| if (leftIsConst && rightIsConst) |
| return; // can't really change the result if both are constants |
| |
| // now we try to change some nulls to constants |
| |
| // convert some nulls to constants |
| if (!invert) { |
| if (pred instanceof EqualExpression) { |
| if (leftIsConst) { |
| t.set(rightCol, generateData(rightDataType, leftConst |
| .toString())); |
| } else if (rightIsConst) { |
| t.set(leftCol, generateData(leftDataType, rightConst |
| .toString())); |
| } else { |
| t.set(leftCol, generateData(leftDataType, "0")); |
| t.set(rightCol, generateData(rightDataType, "0")); |
| } |
| } else if (pred instanceof NotEqualExpression) { |
| if (leftIsConst) { |
| t.set(rightCol, generateData(rightDataType, |
| GetUnequalValue(leftConst).toString())); |
| } else if (rightIsConst) { |
| t.set(leftCol, generateData(leftDataType, GetUnequalValue( |
| rightConst).toString())); |
| } else { |
| t.set(leftCol, generateData(leftDataType, "0")); |
| t.set(rightCol, generateData(rightDataType, "1")); |
| } |
| } else if (pred instanceof GreaterThanExpression |
| || pred instanceof GreaterThanEqualExpression) { |
| if (leftIsConst) { |
| t.set(rightCol, generateData(rightDataType, |
| GetSmallerValue(leftConst).toString())); |
| } else if (rightIsConst) { |
| t.set(leftCol, generateData(leftDataType, GetLargerValue( |
| rightConst).toString())); |
| } else { |
| t.set(leftCol, generateData(leftDataType, "1")); |
| t.set(rightCol, generateData(rightDataType, "0")); |
| } |
| } else if (pred instanceof LessThanExpression |
| || pred instanceof LessThanEqualExpression) { |
| if (leftIsConst) { |
| t.set(rightCol, generateData(rightDataType, GetLargerValue( |
| leftConst).toString())); |
| } else if (rightIsConst) { |
| t.set(leftCol, generateData(leftDataType, GetSmallerValue( |
| rightConst).toString())); |
| } else { |
| t.set(leftCol, generateData(leftDataType, "0")); |
| t.set(rightCol, generateData(rightDataType, "1")); |
| } |
| } |
| } else { |
| if (pred instanceof EqualExpression) { |
| if (leftIsConst) { |
| t.set(rightCol, generateData(rightDataType, |
| GetUnequalValue(leftConst).toString())); |
| } else if (rightIsConst) { |
| t.set(leftCol, generateData(leftDataType, GetUnequalValue( |
| rightConst).toString())); |
| } else { |
| t.set(leftCol, generateData(leftDataType, "0")); |
| t.set(rightCol, generateData(rightDataType, "1")); |
| } |
| } else if (pred instanceof NotEqualExpression) { |
| if (leftIsConst) { |
| t.set(rightCol, generateData(rightDataType, leftConst |
| .toString())); |
| } else if (rightIsConst) { |
| t.set(leftCol, generateData(leftDataType, rightConst |
| .toString())); |
| } else { |
| t.set(leftCol, generateData(leftDataType, "0")); |
| t.set(rightCol, generateData(rightDataType, "0")); |
| } |
| } else if (pred instanceof GreaterThanExpression |
| || pred instanceof GreaterThanEqualExpression) { |
| if (leftIsConst) { |
| t.set(rightCol, generateData(rightDataType, GetLargerValue( |
| leftConst).toString())); |
| } else if (rightIsConst) { |
| t.set(leftCol, generateData(leftDataType, GetSmallerValue( |
| rightConst).toString())); |
| } else { |
| t.set(leftCol, generateData(leftDataType, "0")); |
| t.set(rightCol, generateData(rightDataType, "1")); |
| } |
| } else if (pred instanceof LessThanExpression |
| || pred instanceof LessThanEqualExpression) { |
| if (leftIsConst) { |
| t.set(rightCol, generateData(rightDataType, |
| GetSmallerValue(leftConst).toString())); |
| } else if (rightIsConst) { |
| t.set(leftCol, generateData(leftDataType, GetLargerValue( |
| rightConst).toString())); |
| } else { |
| t.set(leftCol, generateData(leftDataType, "1")); |
| t.set(rightCol, generateData(rightDataType, "0")); |
| } |
| } |
| } |
| |
| } |
| |
| void GenerateMatchingTupleHelper(Tuple t, AndExpression op, boolean invert) |
| throws FrontendException, ExecException { |
| Operator input = op.getLhs(); |
| GenerateMatchingTupleHelper(t, input, invert); |
| input = op.getRhs(); |
| GenerateMatchingTupleHelper(t, input, invert); |
| |
| } |
| |
| void GenerateMatchingTupleHelper(Tuple t, OrExpression op, boolean invert) |
| throws FrontendException, ExecException { |
| Operator input = op.getLhs(); |
| GenerateMatchingTupleHelper(t, input, invert); |
| input = op.getRhs(); |
| GenerateMatchingTupleHelper(t, input, invert); |
| |
| } |
| |
| void GenerateMatchingTupleHelper(Tuple t, NotExpression op, boolean invert) |
| throws FrontendException, ExecException { |
| LogicalExpression input = op.getExpression(); |
| GenerateMatchingTupleHelper(t, input, !invert); |
| |
| } |
| |
| void GenerateMatchingTupleHelper(Tuple t, IsNullExpression op, boolean invert) |
| throws FrontendException, ExecException { |
| byte type = op.getExpression().getType(); |
| if (!invert) |
| t.set(0, null); |
| else |
| t.set(0, generateData(type, "0")); |
| } |
| |
| Object GetUnequalValue(Object v) { |
| byte type = DataType.findType(v); |
| |
| if (type == DataType.BAG || type == DataType.TUPLE |
| || type == DataType.MAP) |
| return null; |
| |
| Object zero = generateData(type, "0"); |
| |
| if (v.equals(zero)) |
| return generateData(type, "1"); |
| |
| return zero; |
| } |
| |
| Object GetSmallerValue(Object v) { |
| byte type = DataType.findType(v); |
| |
| if (type == DataType.BAG || type == DataType.TUPLE |
| || type == DataType.MAP) |
| return null; |
| |
| switch (type) { |
| case DataType.CHARARRAY: |
| String str = (String) v; |
| if (str.length() > 0) |
| return str.substring(0, str.length() - 1); |
| else |
| return null; |
| case DataType.BYTEARRAY: |
| DataByteArray data = (DataByteArray) v; |
| if (data.size() > 0) |
| return new DataByteArray(data.get(), 0, data.size() - 1); |
| else |
| return null; |
| case DataType.INTEGER: |
| return Integer.valueOf((Integer) v - 1); |
| case DataType.LONG: |
| return Long.valueOf((Long) v - 1); |
| case DataType.FLOAT: |
| return Float.valueOf((Float) v - 1); |
| case DataType.DOUBLE: |
| return Double.valueOf((Double) v - 1); |
| case DataType.BIGINTEGER: |
| return ((BigInteger)v).subtract(BigInteger.ONE); |
| case DataType.BIGDECIMAL: |
| return ((BigDecimal)v).subtract(BigDecimal.ONE); |
| case DataType.DATETIME: |
| DateTime dt = (DateTime) v; |
| if (dt.getMillisOfSecond() != 0) { |
| return dt.minusMillis(1); |
| } else if (dt.getSecondOfMinute() != 0) { |
| return dt.minusSeconds(1); |
| } else if (dt.getMinuteOfHour() != 0) { |
| return dt.minusMinutes(1); |
| } else if (dt.getHourOfDay() != 0) { |
| return dt.minusHours(1); |
| } else { |
| return dt.minusDays(1); |
| } |
| default: |
| return null; |
| } |
| |
| } |
| |
| Object GetLargerValue(Object v) { |
| byte type = DataType.findType(v); |
| |
| if (type == DataType.BAG || type == DataType.TUPLE |
| || type == DataType.MAP) |
| return null; |
| |
| switch (type) { |
| case DataType.CHARARRAY: |
| return (String) v + "0"; |
| case DataType.BYTEARRAY: |
| String str = ((DataByteArray) v).toString(); |
| str = str + "0"; |
| return new DataByteArray(str); |
| case DataType.INTEGER: |
| return Integer.valueOf((Integer) v + 1); |
| case DataType.LONG: |
| return Long.valueOf((Long) v + 1); |
| case DataType.FLOAT: |
| return Float.valueOf((Float) v + 1); |
| case DataType.DOUBLE: |
| return Double.valueOf((Double) v + 1); |
| case DataType.BIGINTEGER: |
| return ((BigInteger)v).add(BigInteger.ONE); |
| case DataType.BIGDECIMAL: |
| return ((BigDecimal)v).add(BigDecimal.ONE); |
| case DataType.DATETIME: |
| DateTime dt = (DateTime) v; |
| if (dt.getMillisOfSecond() != 0) { |
| return dt.plusMillis(1); |
| } else if (dt.getSecondOfMinute() != 0) { |
| return dt.plusSeconds(1); |
| } else if (dt.getMinuteOfHour() != 0) { |
| return dt.plusMinutes(1); |
| } else if (dt.getHourOfDay() != 0) { |
| return dt.plusHours(1); |
| } else { |
| return dt.plusDays(1); |
| } |
| default: |
| return null; |
| } |
| } |
| |
| Object generateData(byte type, String data) { |
| switch (type) { |
| case DataType.BOOLEAN: |
| if (data.equalsIgnoreCase("true")) { |
| return Boolean.TRUE; |
| } else if (data.equalsIgnoreCase("false")) { |
| return Boolean.FALSE; |
| } else { |
| return null; |
| } |
| case DataType.BYTEARRAY: |
| return new DataByteArray(data.getBytes()); |
| case DataType.DOUBLE: |
| return Double.valueOf(data); |
| case DataType.FLOAT: |
| return Float.valueOf(data); |
| case DataType.INTEGER: |
| return Integer.valueOf(data); |
| case DataType.LONG: |
| return Long.valueOf(data); |
| case DataType.BIGINTEGER: |
| return new BigInteger(data); |
| case DataType.BIGDECIMAL: |
| return new BigDecimal(data); |
| case DataType.DATETIME: |
| return new DateTime(data); |
| case DataType.CHARARRAY: |
| return data; |
| default: |
| return null; |
| } |
| } |
| |
| } |
