[HIVE-21191] I want to extends lag/lead functions to Implementing some special functions, And I met some problems

• Type: Wish
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Affects Version/s: 1.1.0
• Component/s: Hive, UDF, Windows
• Labels: LAG() UDAF UDF window_function
• Assignee: Unassigned
• Reporter: one
• Created: 31/Jan/19 12:32
• Updated: 08/Jul/19 14:41

Description

i want a distinctLag functions ,The function is like lag, but the difference is to select different values in front of it.
Example:
select * from active

session	sq	channel
1	1	A
1	2	B
1	3	B
1	4	C
1	5	B
2	1	C
2	2	B
2	3	B
2	4	A
2	5	B

select session,sq,lag(channel)over(partition by session order by sq) from active

session	sq	channel
1	1	null
1	2	A
1	3	B
1	4	B
1	5	C
2	1	null
2	2	C
2	3	B
2	4	B
2	5	A

The function I want is：
select session,sq,distinctLag(channel)over(partition by session order by sq) from active

session	sq	channel
1	1	null
1	2	A
1	3	A
1	4	B
1	5	C
2	1	null
2	2	C
2	3	C
2	4	B
2	5	A

 

i try to extend GenericUDFLeadLag and Override:

import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.UDFType;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFLeadLag;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils.ObjectInspectorCopyOption;
@Description(
	    name = "distinctLag",
	    value = "distinctLag  (scalar_expression [,offset] [,default]) OVER ([query_partition_clause] order_by_clause); "
	        + "The distinctLag function is used to access data from a distinct previous row.",
	    extended = "Example:\n "
	    + "select p1.p_mfgr, p1.p_name, p1.p_size,\n"
	    + " p1.p_size - distinctLag(p1.p_size,1,p1.p_size) over( distribute by p1.p_mfgr sort by p1.p_name) as deltaSz\n"
	    + " from part p1 join part p2 on p1.p_partkey = p2.p_partkey")

@UDFType(impliesOrder = true)
public class GenericUDFDistinctLag extends GenericUDFLeadLag {

	@Override
	public Object evaluate(DeferredObject[] arguments) throws HiveException {
		Object defaultVal = null;
		if (arguments.length == 3) {
			defaultVal = ObjectInspectorUtils.copyToStandardObject(getDefaultValueConverter().convert(arguments[2].get()), getDefaultArgOI());
		}

		int idx = getpItr().getIndex() - 1;
		int start = 0;
		int end = getpItr().getPartition().size();
		try {
			Object currValue = ObjectInspectorUtils.copyToStandardObject(getExprEvaluator().evaluate(getpItr().resetToIndex(idx)), getFirstArgOI(), ObjectInspectorCopyOption.WRITABLE);
			Object ret = null;
			int newIdx = idx;
			do {
				--newIdx;
				if (newIdx >= end || newIdx < start) {
					ret = defaultVal;
					return ret;
				}else{
					ret = ObjectInspectorUtils.copyToStandardObject(getExprEvaluator().evaluate(getpItr().lag(1)), getFirstArgOI(), ObjectInspectorCopyOption.WRITABLE);
					if(ret.equals(currValue)){
						setAmt(getAmt() - 1);
					}
				}
			} while (getAmt() > 0);
			return ret;
		} finally {
			Object currRow = getpItr().resetToIndex(idx);
			// reevaluate expression on current Row, to trigger the Lazy object
			// caches to be reset to the current row.
			getExprEvaluator().evaluate(currRow);
		}

	}

	@Override
	protected  String _getFnName(){
		 return "distinctLag";
	}

	@Override
	protected Object getRow(int amt) throws HiveException {
		throw new HiveException("distinctLag error: cannot call getRow");
	}

	@Override
	protected int getIndex(int amt) {
		// TODO Auto-generated method stub
		return 0;
	}
}

and package as a jar,add into hive,create a  temporary function.

then,i run:

select session,sq,distinctLag(channel)over(partition by session order by sq) from active;

It reported an error:

FAILED: SemanticException Failed to breakup Windowing invocations into Groups. At least 1 group must only depend on input columns. Also check for circular dependencies.
Underlying error: Invalid function distinctLag

I don't know exactly what the problem is. I hope someone can give me a hint. Thank you.

then，I noticed that there have a UDAF function GenericUDAFLag.I tried to imitate it.

import java.util.ArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.WindowFunctionDescription;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.plan.ptf.WindowFrameDef;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.ql.udf.generic.ISupportStreamingModeForWindowing;

@WindowFunctionDescription(description = @Description(name = "lag", value = "_FUNC_(expr, amt, default)"), supportsWindow = false, pivotResult = true, impliesOrder = true)
public class GenericUDAFDistinctLag extends GenericUDAFDistinctLeadLag {

	static final Log LOG = LogFactory.getLog(GenericUDAFDistinctLag.class.getName());

	@Override
	protected String functionName() {
		return "Lag";
	}

	@Override
	protected GenericUDAFDistinctLeadLagEvaluator createLLEvaluator() {
		return new GenericUDAFDistinctLagEvaluator();
	}

	public static class GenericUDAFDistinctLagEvaluator extends GenericUDAFDistinctLeadLagEvaluator {

		public GenericUDAFDistinctLagEvaluator() {
		}

		/*
		 * used to initialize Streaming Evaluator.
		 */
		protected GenericUDAFDistinctLagEvaluator(GenericUDAFDistinctLeadLagEvaluator src) {
			super(src);
		}

		@Override
		protected DistinctLeadLagBuffer getNewLLBuffer() throws HiveException {
			return new DistinctLagBuffer();
		}

		@Override
		public GenericUDAFEvaluator getWindowingEvaluator(WindowFrameDef wFrmDef) {

			return new GenericUDAFDistinctLagEvaluatorStreaming(this);
		}

	}

	static class DistinctLagBuffer implements DistinctLeadLagBuffer {
		ArrayList<Object> values;
		int lagAmt;
		ArrayList<Object> lagValues;

		@Override
		public void initialize(int lagAmt) {
			this.lagAmt = lagAmt;
			lagValues = new ArrayList<Object>();
			values = new ArrayList<Object>();
		}

		@Override
		public void addRow(Object currValue, Object defaultValue) {
			int i = values.size() - 1;
			int noEquals = 0;
			for (; i >= 0; i--) {
				if (!currValue.equals(values.get(i))) {
					if (++noEquals == lagAmt) {
						break;
					}
				}
			}
			lagValues.add(i == -1 ? defaultValue : values.get(i));
			values.add(currValue);
		}

		@Override
		public Object terminate() {

			return lagValues;

		}
	}

	/*
	 * StreamingEval: wrap regular eval. on getNext remove first row from values
	 * and return it.
	 */
	static class GenericUDAFDistinctLagEvaluatorStreaming extends GenericUDAFDistinctLagEvaluator implements ISupportStreamingModeForWindowing {

		protected GenericUDAFDistinctLagEvaluatorStreaming(GenericUDAFDistinctLeadLagEvaluator src) {
			super(src);
		}

		@Override
		public Object getNextResult(AggregationBuffer agg) throws HiveException {
			DistinctLagBuffer lb = (DistinctLagBuffer) agg;

			if (!lb.lagValues.isEmpty()) {
				Object res = lb.lagValues.remove(0);
				if (res == null) {
					return ISupportStreamingModeForWindowing.NULL_RESULT;
				}
				return res;
			} else if (!lb.values.isEmpty()) {
				Object res = lb.values.remove(0);
				if (res == null) {
					return ISupportStreamingModeForWindowing.NULL_RESULT;
				}
				return res;
			}
			return null;
		}

		@Override
		public int getRowsRemainingAfterTerminate() throws HiveException {
			return getAmt();
		}
	}
}

 

import java.lang.reflect.Field;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFLead;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFParameterInfo;
import org.apache.hadoop.hive.serde2.objectinspector.ConstantObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorConverters;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorConverters.Converter;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.io.IntWritable;

public abstract class GenericUDAFDistinctLeadLag extends AbstractGenericUDAFResolver {

	static final Log LOG = LogFactory.getLog(GenericUDAFLead.class.getName());

	@Override
	public GenericUDAFEvaluator getEvaluator(GenericUDAFParameterInfo parameters) throws SemanticException {

		ObjectInspector[] paramOIs = parameters.getParameterObjectInspectors();
		String fNm = functionName();

		if (!(paramOIs.length >= 1 && paramOIs.length <= 3)) {
			throw new UDFArgumentTypeException(paramOIs.length - 1, "Incorrect invocation of " + fNm + ": _FUNC_(expr, amt, default)");
		}

		int amt = 1;
		if (paramOIs.length > 1) {
			ObjectInspector amtOI = paramOIs[1];

			if (!ObjectInspectorUtils.isConstantObjectInspector(amtOI) || (amtOI.getCategory() != ObjectInspector.Category.PRIMITIVE)
					|| ((PrimitiveObjectInspector) amtOI).getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.INT) {
				throw new UDFArgumentTypeException(1, fNm + " amount must be a integer value " + amtOI.getTypeName() + " was passed as parameter 1.");
			}
			Object o = ((ConstantObjectInspector) amtOI).getWritableConstantValue();
			amt = ((IntWritable) o).get();
			if (amt < 0) {
				throw new UDFArgumentTypeException(1, fNm + " amount can not be nagative. Specified: " + amt);
			}
		}

		if (paramOIs.length == 3) {
			ObjectInspectorConverters.getConverter(paramOIs[2], paramOIs[0]);
		}

		GenericUDAFDistinctLeadLagEvaluator eval = createLLEvaluator();
		eval.setAmt(amt);
		return eval;
	}

	protected abstract String functionName();

	protected abstract GenericUDAFDistinctLeadLagEvaluator createLLEvaluator();

	public static abstract class GenericUDAFDistinctLeadLagEvaluator extends GenericUDAFEvaluator {

		private transient ObjectInspector[] inputOI;
		private int amt;
		String fnName;
		private transient Converter defaultValueConverter;

		public GenericUDAFDistinctLeadLagEvaluator() {
		}

		/*
		 * used to initialize Streaming Evaluator.
		 */
		protected GenericUDAFDistinctLeadLagEvaluator(GenericUDAFDistinctLeadLagEvaluator src) {
			this.inputOI = src.inputOI;
			this.amt = src.amt;
			this.fnName = src.fnName;
			this.defaultValueConverter = src.defaultValueConverter;
			try {
				Field mode = GenericUDAFEvaluator.class.getDeclaredField("mode");
				mode.setAccessible(true);
				mode.set(this, mode.get(src));
				mode.setAccessible(false);
			} catch (IllegalArgumentException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IllegalAccessException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (NoSuchFieldException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (SecurityException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		@Override
		public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {
			super.init(m, parameters);
			if (m != Mode.COMPLETE) {
				throw new HiveException("Only COMPLETE mode supported for " + fnName + " function");
			}

			inputOI = parameters;

			if (parameters.length == 3) {
				defaultValueConverter = ObjectInspectorConverters.getConverter(parameters[2], parameters[0]);
			}

			return ObjectInspectorFactory.getStandardListObjectInspector(ObjectInspectorUtils.getStandardObjectInspector(parameters[0]));
		}

		public int getAmt() {
			return amt;
		}

		public void setAmt(int amt) {
			this.amt = amt;
		}

		public String getFnName() {
			return fnName;
		}

		public void setFnName(String fnName) {
			this.fnName = fnName;
		}

		protected abstract DistinctLeadLagBuffer getNewLLBuffer() throws HiveException;

		@Override
		public AggregationBuffer getNewAggregationBuffer() throws HiveException {
			DistinctLeadLagBuffer lb = getNewLLBuffer();
			lb.initialize(amt);
			return lb;
		}

		@Override
		public void reset(AggregationBuffer agg) throws HiveException {
			((DistinctLeadLagBuffer) agg).initialize(amt);
		}

		@Override
		public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException {
			Object rowExprVal = ObjectInspectorUtils.copyToStandardObject(parameters[0], inputOI[0]);
			Object defaultVal = parameters.length > 2 ? ObjectInspectorUtils.copyToStandardObject(defaultValueConverter.convert(parameters[2]), inputOI[0]) : null;
			((DistinctLeadLagBuffer) agg).addRow(rowExprVal, defaultVal);
		}

		@Override
		public Object terminatePartial(AggregationBuffer agg) throws HiveException {
			throw new HiveException("terminatePartial not supported");
		}

		@Override
		public void merge(AggregationBuffer agg, Object partial) throws HiveException {
			throw new HiveException("merge not supported");
		}

		@Override
		public Object terminate(AggregationBuffer agg) throws HiveException {
			return ((DistinctLeadLagBuffer) agg).terminate();
		}

	}

}

import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.AggregationBuffer;

@SuppressWarnings("deprecation")
interface DistinctLeadLagBuffer extends AggregationBuffer {
  void initialize(int leadAmt);

  void addRow(Object leadExprValue, Object defaultValue);

  Object terminate();

}

and package as a jar,add into hive,create a temporary function.in hige hive vesion,it works, but in hive1.1.0 version ,It reported an error (and in hive1.1.0 if i create that temporary function named lag or lead,it also works as what i want ,but it will cover hive's built-in function lag/lead even if deleted  that temporary function,Only when I quit hive-cli and reenter hive-cli , built-in function lag/lead can work):

hive> SELECT session,sq,distinctLag(channel)over(PARTITION BY session ORDER BY sq) FROM elephant_active;
Query ID = root_20190131195959_8047b4ba-a85c-4f39-8a27-989388316c50
Total jobs = 1
Launching Job 1 out of 1
Number of reduce tasks not specified. Estimated from input data size: 1
In order to change the average load for a reducer (in bytes):
  set hive.exec.reducers.bytes.per.reducer=<number>
In order to limit the maximum number of reducers:
  set hive.exec.reducers.max=<number>
In order to set a constant number of reducers:
  set mapreduce.job.reduces=<number>
Starting Job = job_1546504603780_0492, Tracking URL = http://dce.bi.com:8088/proxy/application_1546504603780_0492/
Kill Command = /opt/cloudera/parcels/CDH-5.10.2-1.cdh5.10.2.p0.5/lib/hadoop/bin/hadoop job  -kill job_1546504603780_0492
Hadoop job information for Stage-1: number of mappers: 1; number of reducers: 1
2019-01-31 20:00:03,639 Stage-1 map = 0%,  reduce = 0%
2019-01-31 20:00:09,972 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 1.38 sec
2019-01-31 20:00:30,795 Stage-1 map = 100%,  reduce = 100%, Cumulative CPU 1.38 sec
MapReduce Total cumulative CPU time: 1 seconds 380 msec
Ended Job = job_1546504603780_0492 with errors
Error during job, obtaining debugging information...
Examining task ID: task_1546504603780_0492_m_000000 (and more) from job job_1546504603780_0492

Task with the most failures(4): 
-----
Task ID:
  task_1546504603780_0492_r_000000

URL:
  http://0.0.0.0:8088/taskdetails.jsp?jobid=job_1546504603780_0492&tipid=task_1546504603780_0492_r_000000
-----
Diagnostic Messages for this Task:
Error: java.lang.RuntimeException: Error in configuring object
	at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:109)
	at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:75)
	at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:133)
	at org.apache.hadoop.mapred.ReduceTask.runOldReducer(ReduceTask.java:409)
	at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:392)
	at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:164)
	at java.security.AccessController.doPrivileged(Native Method)
	at javax.security.auth.Subject.doAs(Subject.java:422)
	at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1920)
	at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)
Caused by: java.lang.reflect.InvocationTargetException
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:497)
	at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:106)
	... 9 more
Caused by: java.lang.RuntimeException: Reduce operator initialization failed
	at org.apache.hadoop.hive.ql.exec.mr.ExecReducer.configure(ExecReducer.java:166)
	... 14 more
Caused by: java.lang.NullPointerException
	at org.apache.hadoop.hive.ql.exec.Registry.getFunctionInfo(Registry.java:306)
	at org.apache.hadoop.hive.ql.exec.Registry.getWindowFunctionInfo(Registry.java:314)
	at org.apache.hadoop.hive.ql.exec.FunctionRegistry.getWindowFunctionInfo(FunctionRegistry.java:504)
	at org.apache.hadoop.hive.ql.udf.ptf.WindowingTableFunction.streamingPossible(WindowingTableFunction.java:151)
	at org.apache.hadoop.hive.ql.udf.ptf.WindowingTableFunction.setCanAcceptInputAsStream(WindowingTableFunction.java:222)
	at org.apache.hadoop.hive.ql.udf.ptf.WindowingTableFunction.initializeStreaming(WindowingTableFunction.java:256)
	at org.apache.hadoop.hive.ql.exec.PTFOperator$PTFInvocation.initializeStreaming(PTFOperator.java:291)
	at org.apache.hadoop.hive.ql.exec.PTFOperator.initializeOp(PTFOperator.java:86)
	at org.apache.hadoop.hive.ql.exec.Operator.initialize(Operator.java:385)
	at org.apache.hadoop.hive.ql.exec.Operator.initialize(Operator.java:469)
	at org.apache.hadoop.hive.ql.exec.Operator.initializeChildren(Operator.java:425)
	at org.apache.hadoop.hive.ql.exec.SelectOperator.initializeOp(SelectOperator.java:65)
	at org.apache.hadoop.hive.ql.exec.Operator.initialize(Operator.java:385)
	at org.apache.hadoop.hive.ql.exec.mr.ExecReducer.configure(ExecReducer.java:159)
	... 14 more


FAILED: Execution Error, return code 2 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask
MapReduce Jobs Launched: 
Stage-Stage-1: Map: 1  Reduce: 1   Cumulative CPU: 1.38 sec   HDFS Read: 2958 HDFS Write: 0 FAIL
Total MapReduce CPU Time Spent: 1 seconds 380 msec
hive> 

I guess it's FunctionRegistry problem.
I am a beginner. I hope someone can tell me the correct way to realize this special function. Thank you very much. I use Hive 1.1.0 + cdh5.10.2 + 945.

<repositories>
<repository>
<id>cloudera</id>
<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>1.1.0-cdh5.10.2</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.9</version>
<scope>test</scope>
</dependency>
</dependencies>