refactor PTF ifc;

- clearer separation of duties between Resolver and Evaluator.
- document ifc.
- provide a PItr to execute method.
- linking to LeadLag done in Executor.
WindowTblFunc:
- hold onto WindowFnDefs
- QDeserializer, no special handling for WindowTblFunc

Author: Harish Butani

windowing/src/main/java/com/sap/hadoop/windowing/functions2/TableFunctionEvaluator.java

@@ -10,127 +10,136 @@
 import com.sap.hadoop.windowing.query2.definition.QueryDef;
 import com.sap.hadoop.windowing.query2.definition.TableFuncDef;
 import com.sap.hadoop.windowing.runtime2.Partition;
+import com.sap.hadoop.windowing.runtime2.PartitionIterator;
+import com.sap.hadoop.windowing.runtime2.RuntimeUtils;
 
 /**
- * Based on Hive {@link GenericUDAFEvaluator}. Break up the responsibility of the old AsbtractTableFunction class into a Resolver and Evaluator.
- * 
+ * Based on Hive {@link GenericUDAFEvaluator}. Break up the responsibility of the old AsbtractTableFunction 
+ * class into a Resolver and Evaluator.
+ * <p>
+ * The Evaluator also holds onto the {@link TableFunctionDef}. This provides information
+ * about the arguments to the function, the shape of the Input partition and the Partitioning details. 
+ * The Evaluator is responsible for providing the 2 execute methods:
+ * <ol>
+ * <li><b>execute:</b> which is invoked after the input is partitioned; the contract
+ * is, it is given an input Partition and must return an output Partition. The shape of the output
+ * Partition is obtained from the getOutputOI call. 
+ * <li><b>transformRawInput:</b> In the case where this function indicates that it will transform the raw input
+ * before it is fed through the partitioning mechanics, this function is called. Again the contract is
+ * t is given an input Partition and must return an Partition. The shape of the output Partition is
+ * obtained from getRawInputOI() call.
+ * </ol>
+
+ * </ol> 
  *
  */
 public abstract class TableFunctionEvaluator
 {
-	protected TableFunctionResolver resolver;
 	transient protected StructObjectInspector OI;
-	transient protected StructObjectInspector mapOI;
+	transient protected StructObjectInspector rawInputOI;
 	protected TableFuncDef tDef;
 	protected QueryDef qDef;
 	String partitionClass;
 	int partitionMemSize;
-	
+	boolean transformsRawInput;
+
 	static{
 		SerializationUtils.makeTransient(TableFunctionEvaluator.class, "OI");
-		SerializationUtils.makeTransient(TableFunctionEvaluator.class, "mapOI");
+		SerializationUtils.makeTransient(TableFunctionEvaluator.class, "rawInputOI");
 	}
 
-	
-	public TableFunctionResolver getResolver()
-	{
-		return resolver;
-	}
 
-	public void setResolver(TableFunctionResolver resolver)
-	{
-		this.resolver = resolver;
-	}
-	
 	public StructObjectInspector getOutputOI()
 	{
 		return OI;
 	}
 
-	public void setTableDef(TableFuncDef tDef)
+	protected void setOutputOI(StructObjectInspector outputOI)
 	{
-		this.tDef = tDef;
-	}
-
-	public void setQueryDef(QueryDef qDef)
-	{
-		this.qDef = qDef;
+		OI = outputOI;
 	}
 
-	public void setPartitionClass(String partitionClass)
+	public TableFuncDef getTableDef()
 	{
-		this.partitionClass = partitionClass;
+		return tDef;
 	}
 
-	public void setPartitionMemSize(int partitionMemSize)
+	public void setTableDef(TableFuncDef tDef)
 	{
-		this.partitionMemSize = partitionMemSize;
+		this.tDef = tDef;
 	}
 
-	public StructObjectInspector getMapOutputOI()
-	{
-		return mapOI;
-	}
-	
-	public TableFuncDef getTableDef()
-	{
-		return tDef;
-	}
-	
-	public QueryDef getQueryDef()
+	protected QueryDef getQueryDef()
 	{
 		return qDef;
 	}
-	
-	public abstract void setupOI() throws WindowingException; 
-	
-	public void setupMapOI() throws WindowingException
+
+	protected void setQueryDef(QueryDef qDef)
 	{
-		if (!resolver.hasMapPhase())
-		{
-			return;
-		}
-		mapOI = OI;
+		this.qDef = qDef;
 	}
-	
+
 	public String getPartitionClass()
 	{
 		return partitionClass;
 	}
 
+	public void setPartitionClass(String partitionClass)
+	{
+		this.partitionClass = partitionClass;
+	}
+
 	public int getPartitionMemSize()
 	{
 		return partitionMemSize;
 	}
-	
-	public boolean hasMapPhase() throws WindowingException
+
+	public void setPartitionMemSize(int partitionMemSize)
 	{
-		resolver.setHasMapPhase();
-		return resolver.hasMapPhase();
+		this.partitionMemSize = partitionMemSize;
 	}
-	
-	public Partition execute(Partition iPart) throws WindowingException
+
+	public StructObjectInspector getRawInputOI()
 	{
-		Partition outP = new Partition(getPartitionClass(), getPartitionMemSize(), tDef.getSerde(), OI);
-		execute(iPart, outP);
-		return outP;
+		return rawInputOI;
 	}
 
-	protected void execute(Partition iPart, Partition oPart) throws WindowingException
+	protected void setRawInputOI(StructObjectInspector rawInputOI)
 	{
-		
+		this.rawInputOI = rawInputOI;
+	}
+
+	public boolean isTransformsRawInput() {
+    return transformsRawInput;
+  }
+
+  public void setTransformsRawInput(boolean transformsRawInput) {
+    this.transformsRawInput = transformsRawInput;
+  }
+
+	public Partition execute(Partition iPart)
+			throws WindowingException
+	{
+		PartitionIterator<Object> pItr = iPart.iterator();
+		RuntimeUtils.connectLeadLagFunctionsToPartition(qDef, pItr);
+		Partition outP = new Partition(getPartitionClass(),
+				getPartitionMemSize(), tDef.getSerde(), OI);
+		execute(pItr, outP);
+		return outP;
 	}
 
-	public Partition mapExecute(Partition iPart) throws WindowingException
+	protected abstract void execute(PartitionIterator<Object> pItr, Partition oPart) throws WindowingException;
+
+	public Partition transformRawInput(Partition iPart) throws WindowingException
 	{
-		if ( !resolver.hasMapPhase())
+		if ( !isTransformsRawInput())
 		{
 			throw new WindowingException(sprintf("Internal Error: mapExecute called on function (%s)that has no Map Phase", tDef.getName()));
 		}
-		return _mapExecute(iPart);
+		return _transformRawInput(iPart);
 	}
-	
-	protected Partition _mapExecute(Partition iPart) throws WindowingException
+
+	protected Partition _transformRawInput(Partition iPart) throws WindowingException
 	{
 		return null;
 	}

windowing/src/main/java/com/sap/hadoop/windowing/functions2/TableFunctionResolver.java

@@ -2,6 +2,7 @@
 
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFResolver;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
 
 import com.sap.hadoop.ds.list.ByteBasedList;
 import com.sap.hadoop.windowing.Constants;
@@ -11,41 +12,129 @@
 
 /**
  * Based on Hive {@link GenericUDAFResolver}. Break up the responsibility of the
- * old AsbtractTableFunction class into a Resolver and Evaluator.
- * 
- * 
+ * old AbstractTableFunction class into a Resolver and Evaluator.
+ * The Resolver is responsible for:
+ * <ol>
+ * <li> setting up the {@link tableFunctionEvaluator}
+ * <li> Setting up the The raw and output ObjectInspectors of the Evaluator.
+ * <li> The Evaluator also holds onto the {@link TableFunctionDef}. This provides information
+ * about the arguments to the function, the shape of the Input partition and the Partitioning details.
+ * </ol>
+ * The Resolver for a function is obtained from the {@link FunctionRegistry}. The Resolver is initialized
+ * by the following 4 step process:
+ * <ol>
+ * <li> The initialize method is called; which is passed the {@link QueryDef} and the {@link TableFunctionDef}.
+ * <li> The resolver is then asked to setup the Raw ObjectInspector. This is only required if the Function reshapes
+ * the raw input.
+ * <li> Once the Resolver has had a chance to compute the shape of the Raw Input that is fed to the partitioning 
+ * machinery; the translator sets up the partitioning details on the tableFuncDef.
+ * <li> finally the resolver is asked to setup the output ObjectInspector.
+ * </ol> 
  */
 @SuppressWarnings("deprecation")
-public abstract class TableFunctionResolver {
-	protected boolean hasMapPhase;
+public abstract class TableFunctionResolver
+{
+	TableFunctionEvaluator evaluator;
+	QueryDef qDef;
 
-	public TableFunctionEvaluator initialize(QueryDef qDef, TableFuncDef tDef)
-			throws WindowingException {
+	/*
+	 * - called during translation.
+	 * - invokes createEvaluator which must be implemented by a subclass
+	 * - sets up the evaluator with references to the TableDef, PartitionClass, PartitonMemsize and
+	 *   the transformsRawInput boolean.
+	 */
+	public void initialize(QueryDef qDef, TableFuncDef tDef)
+			throws WindowingException
+	{
+		this.qDef = qDef;
 		HiveConf cfg = qDef.getTranslationInfo().getHiveCfg();
 		String partitionClass = cfg.get(Constants.WINDOW_PARTITION_CLASS,
 				Constants.DEFAULT_WINDOW_PARTITION_CLASS);
 		int partitionMemSize = cfg.getInt(Constants.WINDOW_PARTITION_MEM_SIZE,
 				ByteBasedList.MEDIUM_SIZE);
 
-		TableFunctionEvaluator tfEval = createEvaluator(qDef, tDef);
-		tfEval.setResolver(this);
-		tfEval.setQueryDef(qDef);
-		tfEval.setTableDef(tDef);
-		tfEval.setPartitionClass(partitionClass);
-		tfEval.setPartitionMemSize(partitionMemSize);
+		evaluator = createEvaluator(qDef, tDef);
+		evaluator.setTransformsRawInput(transformsRawInput());
+		evaluator.setTableDef(tDef);
+		evaluator.setQueryDef(qDef);
+		evaluator.setPartitionClass(partitionClass);
+		evaluator.setPartitionMemSize(partitionMemSize);
 
-		return tfEval;
+	}
+	
+	/*
+	 * called during deserialization of a QueryDef during runtime.
+	 */
+	public void initialize(QueryDef qDef, TableFuncDef tDef, TableFunctionEvaluator evaluator)
+			throws WindowingException 
+	{
+		this.evaluator = evaluator;
+		this.qDef = qDef;
+		evaluator.setTableDef(tDef);
+		evaluator.setQueryDef(qDef);
+	}
+	
+	public TableFunctionEvaluator getEvaluator()
+	{
+		return evaluator;
 	}
 
-	public boolean hasMapPhase() {
-		return hasMapPhase;
+	/*
+	 * - a subclass must provide this method.
+	 * - this method is invoked during translation and also when the Operator is initialized during runtime.
+	 * - a subclass must use this call to setup the shape of its output.
+	 * - subsequent to this call, a call to getOutputOI call on the {@link TableFunctionEvaluator} must return the OI
+	 * of the output of this function.
+	 */
+	public abstract void setupOutputOI() throws WindowingException;
+
+	/*
+	 * - Called on functions that transform the raw input.
+	 * - this method is invoked during translation and also when the Operator is initialized during runtime.
+	 * - a subclass must use this call to setup the shape of the raw input, that is fed to the partitioning mechanics.
+	 * - subsequent to this call, a call to getRawInputOI call on the {@link TableFunctionEvaluator} must return the OI
+	 *   of the output of this function.
+	 */
+	public void setupRawInputOI() throws WindowingException
+	{
+		if (!transformsRawInput())
+		{
+			return;
+		}
+		throw new WindowingException(
+				"Function has map phase, must extend setupMapOI");
+	}
+	
+	/*
+	 * callback method used by subclasses to set the RawInputOI on the Evaluator.
+	 */
+	protected void setRawInputOI(StructObjectInspector rawInputOI)
+	{
+		evaluator.setRawInputOI(rawInputOI);
+	}
+	
+	/*
+	 * callback method used by subclasses to set the OutputOI on the Evaluator.
+	 */
+	protected void setOutputOI(StructObjectInspector outputOI)
+	{
+		evaluator.setOutputOI(outputOI);
 	}
 
-	protected void setHasMapPhase() throws WindowingException {
-		hasMapPhase = false;
+	public QueryDef getQueryDef()
+	{
+		return qDef;
 	}
 
+	/*
+	 * a subclass must indicate whether it will transform the raw input before it is fed through the
+	 * partitioning mechanics.
+	 */
+	public abstract boolean transformsRawInput();
+
+	/*
+	 * a subclass must provide the {@link TableFunctionEvaluator} instance.
+	 */
 	protected abstract TableFunctionEvaluator createEvaluator(QueryDef qDef,
 			TableFuncDef tDef);
-
 }