You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by cs...@apache.org on 2023/03/09 14:13:27 UTC
[impala] 05/07: IMPALA-11854: ImpalaStringWritable's underlying array can't be changed in UDFs
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit afe59f7f0d5658a6639e3e6f1f9ff9ed2f76d658
Author: Peter Rozsa <pr...@cloudera.com>
AuthorDate: Wed Feb 1 08:55:39 2023 +0100
IMPALA-11854: ImpalaStringWritable's underlying array can't be changed in UDFs
This change fixes the behavior of BytesWritable and TextWritable's
getBytes() method. Now the returned byte array could be handled as
the underlying buffer as it gets loaded before the UDF's evaluation,
and tracks the changes as a regular Java byte array; the resizing
operation still resets the reference. The operations that wrote back
to the native heap were also removed as these operations are now
handled in the byte array. ImpalaStringWritable class is also removed,
writables that used it before now store the data directly.
Tests:
- Test UDFs added as BufferAlteringUdf and GenericBufferAlteringUdf
- E2E test ran for UDFs
Change-Id: Ifb28bd0dce7b0482c7abe1f61f245691fcbfe212
Reviewed-on: http://gerrit.cloudera.org:8080/19507
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../impala/compat/HiveEsriGeospatialBuiltins.java | 37 +--
.../impala/hive/executor/HiveUdfExecutor.java | 5 +-
.../hive/executor/HiveUdfExecutorGeneric.java | 64 ++---
.../hive/executor/HiveUdfExecutorLegacy.java | 11 +-
.../impala/hive/executor/ImpalaBytesWritable.java | 34 +--
.../impala/hive/executor/ImpalaStringWritable.java | 117 --------
.../impala/hive/executor/ImpalaTextWritable.java | 22 +-
.../impala/hive/executor/JavaUdfDataType.java | 315 ++++++++++-----------
.../{ImpalaTextWritable.java => Reloadable.java} | 27 +-
.../impala/hive/executor/UdfExecutorTest.java | 45 +--
.../java/org/apache/impala/BufferAlteringUdf.java | 85 ++++++
.../apache/impala/GenericBufferAlteringUdf.java | 105 +++++++
.../queries/QueryTest/generic-java-udf.test | 28 ++
.../queries/QueryTest/java-udf.test | 35 +++
.../queries/QueryTest/load-generic-java-udfs.test | 8 +
.../queries/QueryTest/load-java-udfs.test | 12 +
16 files changed, 513 insertions(+), 437 deletions(-)
diff --git a/fe/src/compat-hive-3/java/org/apache/impala/compat/HiveEsriGeospatialBuiltins.java b/fe/src/compat-hive-3/java/org/apache/impala/compat/HiveEsriGeospatialBuiltins.java
index 8ce149942..578dd6d0f 100644
--- a/fe/src/compat-hive-3/java/org/apache/impala/compat/HiveEsriGeospatialBuiltins.java
+++ b/fe/src/compat-hive-3/java/org/apache/impala/compat/HiveEsriGeospatialBuiltins.java
@@ -64,7 +64,6 @@ public class HiveEsriGeospatialBuiltins {
addLegacyUDFs(db);
addGenericUDFs(db);
addVarargsUDFs(db);
- addWorkaroundForStSetSrid(db);
}
private static void addLegacyUDFs(Db db) {
@@ -84,7 +83,7 @@ public class HiveEsriGeospatialBuiltins {
new ST_NumInteriorRing(), new ST_NumPoints(), new ST_Point(),
new ST_PointFromWKB(), new ST_PointN(), new ST_PointZ(), new ST_PolyFromWKB(),
new ST_Relate(), new ST_SRID(), new ST_StartPoint(), new ST_SymmetricDiff(),
- new ST_X(), new ST_Y(), new ST_Z());
+ new ST_X(), new ST_Y(), new ST_Z(), new ST_SetSRID());
for (UDF udf : legacyUDFs) {
for (Function fn : extractFromLegacyHiveBuiltin(udf, db.getName())) {
@@ -207,38 +206,4 @@ public class HiveEsriGeospatialBuiltins {
})
.collect(Collectors.toList());
}
-
- /*
- TODO: IMPALA-11854: A workaround must be applied for ST_SetSRID UDF because the
- GeometryUtils.setWKID method assumes that the incoming geomref's buffer can
- be modified through the array returned by ImpalaBytesWritable.getBytes.
- */
- private static void addWorkaroundForStSetSrid(Db db) {
- db.addBuiltin(
- createScalarFunction(ST_SetSRID_Wrapper.class, ST_SetSRID.class.getSimpleName(),
- Type.BINARY, new Type[] {Type.BINARY, Type.INT}));
- }
-
- public static class ST_SetSRID_Wrapper extends ST_SetSRID {
- private static final Logger LOG = LoggerFactory.getLogger(ST_SetSRID_Wrapper.class);
-
- @Override
- public BytesWritable evaluate(BytesWritable geomref, IntWritable wkwrap) {
- if (geomref != null && geomref.getLength() != 0) {
- if (wkwrap != null) {
- int wkid = wkwrap.get();
- if (GeometryUtils.getWKID(geomref) != wkid) {
- ByteBuffer bb = ByteBuffer.allocate(geomref.getLength());
- bb.putInt(wkid);
- bb.put(Arrays.copyOfRange(geomref.getBytes(), 4, geomref.getLength()));
- return new BytesWritable(bb.array());
- }
- }
- return geomref;
- } else {
- LogUtils.Log_ArgumentsNull(LOG);
- return null;
- }
- }
- }
}
diff --git a/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutor.java b/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutor.java
index 31721f666..ee54066e9 100644
--- a/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutor.java
+++ b/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutor.java
@@ -115,7 +115,7 @@ public abstract class HiveUdfExecutor {
}
/**
- * Evalutes the UDF with 'args' as the input to the UDF. This is exposed
+ * Evaluates the UDF with 'args' as the input to the UDF. This is exposed
* for testing and not the version of evaluate() the backend uses.
*/
public long evaluateForTesting(Object... args) throws ImpalaRuntimeException {
@@ -234,8 +234,7 @@ public abstract class HiveUdfExecutor {
}
UnsafeUtil.Copy(outBufferStringPtr_, bytes, 0, bytes.length);
UnsafeUtil.UNSAFE.putInt(
- outputBufferPtr_ + ImpalaStringWritable.STRING_VALUE_LEN_OFFSET,
- bytes.length);
+ outputBufferPtr_ + JavaUdfDataType.STRING_VALUE_LEN_OFFSET, bytes.length);
}
// Preallocate the input objects that will be passed to the underlying UDF.
diff --git a/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutorGeneric.java b/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutorGeneric.java
index 5d292e17e..9024ab536 100644
--- a/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutorGeneric.java
+++ b/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutorGeneric.java
@@ -17,46 +17,18 @@
package org.apache.impala.hive.executor;
-import sun.misc.Unsafe;
-
-import java.io.File;
-import java.io.IOException;
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.net.URLClassLoader;
-import java.util.ArrayList;
-import java.util.List;
-import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF.DeferredJavaObject;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF.DeferredObject;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
-import org.apache.hadoop.io.BooleanWritable;
-import org.apache.hadoop.io.BytesWritable;
-import org.apache.hadoop.io.FloatWritable;
-import org.apache.hadoop.io.IntWritable;
-import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
-import org.apache.impala.catalog.Type;
-import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.ImpalaRuntimeException;
-import org.apache.impala.common.JniUtil;
import org.apache.impala.thrift.THiveUdfExecutorCtorParams;
-import org.apache.impala.thrift.TPrimitiveType;
import org.apache.impala.util.UnsafeUtil;
import org.apache.log4j.Logger;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
/**
* Wrapper object to run hive GenericUDFs. This class works with UdfCallExpr in the
@@ -98,17 +70,20 @@ public class HiveUdfExecutorGeneric extends HiveUdfExecutor {
}
/**
- * Evalutes the UDF with 'args' as the input to the UDF.
+ * Evaluates the UDF with 'args' as the input to the UDF.
*/
@Override
protected Object evaluateDerived(JavaUdfDataType[] argTypes,
long inputNullsPtr, Object[] inputObjectArgs) throws ImpalaRuntimeException {
try {
for (int i = 0; i < runtimeDeferredParameters_.length; ++i) {
- runtimeDeferredParameters_[i] =
- (UnsafeUtil.UNSAFE.getByte(inputNullsPtr + i) == 0)
- ? deferredParameters_[i]
- : deferredNullParameter_;
+ if (UnsafeUtil.UNSAFE.getByte(inputNullsPtr + i) == 0) {
+ runtimeDeferredParameters_[i] = deferredParameters_[i];
+ // argument 'i' is unused in DeferredJavaObject and in DeferredWritable as well
+ runtimeDeferredParameters_[i].prepare(0);
+ } else {
+ runtimeDeferredParameters_[i] = deferredNullParameter_;
+ }
}
return genericUDF_.evaluate(runtimeDeferredParameters_);
} catch (HiveException e) {
@@ -140,8 +115,29 @@ public class HiveUdfExecutorGeneric extends HiveUdfExecutor {
private DeferredObject[] createDeferredObjects() {
DeferredObject[] deferredObjects = new DeferredObject[getNumParams()];
for (int i = 0; i < deferredObjects.length; ++i) {
- deferredObjects[i] = new DeferredJavaObject(getInputObject(i));
+ Object inputObject = getInputObject(i);
+ if (inputObject instanceof Reloadable) {
+ deferredObjects[i] = new DeferredWritable<>((Reloadable) inputObject);
+ } else {
+ deferredObjects[i] = new DeferredJavaObject(inputObject);
+ }
}
return deferredObjects;
}
+
+ private static class DeferredWritable<T extends Reloadable> implements DeferredObject {
+ private final T writable;
+
+ public DeferredWritable(T writable) { this.writable = writable; }
+
+ @Override
+ public void prepare(int ignored) throws HiveException {
+ writable.reload();
+ }
+
+ @Override
+ public Object get() throws HiveException {
+ return writable;
+ }
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutorLegacy.java b/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutorLegacy.java
index 0f9c59328..c665997d9 100644
--- a/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutorLegacy.java
+++ b/fe/src/main/java/org/apache/impala/hive/executor/HiveUdfExecutorLegacy.java
@@ -82,10 +82,14 @@ public class HiveUdfExecutorLegacy extends HiveUdfExecutor {
case INT_WRITABLE:
case LONG_WRITABLE:
case FLOAT_WRITABLE:
- case DOUBLE_WRITABLE:
+ case DOUBLE_WRITABLE: inputArgs_[i] = inputObjects[i]; break;
case BYTE_ARRAY:
case BYTES_WRITABLE:
+ ((ImpalaBytesWritable) inputObjects[i]).reload();
+ inputArgs_[i] = inputObjects[i];
+ break;
case TEXT:
+ ((ImpalaTextWritable) inputObjects[i]).reload();
inputArgs_[i] = inputObjects[i];
break;
case BOOLEAN:
@@ -111,8 +115,9 @@ public class HiveUdfExecutorLegacy extends HiveUdfExecutor {
break;
case STRING:
Preconditions.checkState(inputObjects[i] instanceof ImpalaBytesWritable);
- inputArgs_[i] =
- new String(((ImpalaBytesWritable)inputObjects[i]).getBytes());
+ ImpalaBytesWritable inputObject = (ImpalaBytesWritable) inputObjects[i];
+ inputObject.reload();
+ inputArgs_[i] = new String(inputObject.getBytes());
break;
}
} else {
diff --git a/fe/src/main/java/org/apache/impala/hive/executor/ImpalaBytesWritable.java b/fe/src/main/java/org/apache/impala/hive/executor/ImpalaBytesWritable.java
index 94824e02c..f223f1194 100644
--- a/fe/src/main/java/org/apache/impala/hive/executor/ImpalaBytesWritable.java
+++ b/fe/src/main/java/org/apache/impala/hive/executor/ImpalaBytesWritable.java
@@ -21,36 +21,18 @@ import org.apache.hadoop.io.BytesWritable;
/**
* Impala writable type that implements the BytesWritable interface. The data
- * marshalling is handled by the underlying {@link ImpalaStringWritable} object.
+ * marshalling is handled by {@link JavaUdfDataType#loadStringValueFromNativeHeap(long)}.
*/
-public class ImpalaBytesWritable extends BytesWritable {
- private final ImpalaStringWritable string_;
+public class ImpalaBytesWritable extends BytesWritable implements Reloadable {
+ private final long ptr_;
- public ImpalaBytesWritable(long ptr) {
- string_ = new ImpalaStringWritable(ptr);
- }
+ public ImpalaBytesWritable(long ptr) { this.ptr_ = ptr; }
@Override
- public byte[] copyBytes() {
- byte[] src = getBytes();
- return src.clone();
+ public void reload() {
+ byte[] bytes = JavaUdfDataType.loadStringValueFromNativeHeap(ptr_);
+ super.setCapacity(bytes.length);
+ super.set(bytes, 0, bytes.length);
}
- @Override
- public byte[] get() { return getBytes(); }
- @Override
- public byte[] getBytes() { return string_.getBytes(); }
- @Override
- public int getCapacity() { return string_.getCapacity(); }
- @Override
- public int getLength() { return string_.getLength(); }
-
- public ImpalaStringWritable getStringWritable() { return string_; }
-
- @Override
- public void set(byte[] v, int offset, int len) { string_.set(v, offset, len); }
- @Override
- public void setCapacity(int newCap) { string_.setCapacity(newCap); }
- @Override
- public void setSize(int size) { string_.setSize(size); }
}
diff --git a/fe/src/main/java/org/apache/impala/hive/executor/ImpalaStringWritable.java b/fe/src/main/java/org/apache/impala/hive/executor/ImpalaStringWritable.java
deleted file mode 100644
index 2ef6adf47..000000000
--- a/fe/src/main/java/org/apache/impala/hive/executor/ImpalaStringWritable.java
+++ /dev/null
@@ -1,117 +0,0 @@
-// 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.impala.hive.executor;
-
-import java.nio.ByteBuffer;
-
-import org.apache.impala.util.UnsafeUtil;
-
-@SuppressWarnings("restriction")
-/**
- * Underlying class for Text and Bytes writable. This class understands marshalling
- * values that map to StringValue in the BE.
- * StringValue is replicated here:
- * struct StringValue {
- * char* ptr;
- * int len;
- * };
- */
-public class ImpalaStringWritable {
- // The length is 8 bytes into the struct.
- static public final int STRING_VALUE_LEN_OFFSET = 8;
-
- // Ptr (to native heap) where the value should be read from and written to.
- // This needs to be ABI compatible with the BE StringValue class
- private final long stringValPtr_;
-
- // Array object to convert between native and java heap (i.e. byte[]).
- private ByteBuffer array_;
-
- // Set if this object had to allocate from the native heap on the java side. If this
- // is set, it will always be stringValPtr_->ptr
- // We only need to allocate from the java side if we are trying to set the
- // StringValue to a bigger size than what the native side allocated.
- // If this object is used as a read-only input argument, this value will stay
- // 0.
- private long bufferPtr_;
-
- // Allocation size of stringValPtr_'s ptr.
- private int bufferCapacity_;
-
- // Creates a string writable backed by a StringValue object. Ptr must be a valid
- // StringValue (in the native heap).
- public ImpalaStringWritable(long ptr) {
- stringValPtr_ = ptr;
- bufferPtr_= 0;
- bufferCapacity_ = getLength();
- array_ = ByteBuffer.allocate(0);
- }
-
- /*
- * Implement finalize() to clean up any allocations from the native heap.
- */
- @Override
- protected void finalize() throws Throwable {
- UnsafeUtil.UNSAFE.freeMemory(bufferPtr_);
- super.finalize();
- }
-
- // Returns the underlying bytes as a byte[]
- public byte[] getBytes() {
- int len = getLength();
- // TODO: reuse this array.
- array_ = ByteBuffer.allocate(len);
- byte[] buffer = array_.array();
-
- long srcPtr = UnsafeUtil.UNSAFE.getLong(stringValPtr_);
- UnsafeUtil.Copy(buffer, 0, srcPtr, len);
- return buffer;
- }
-
- // Returns the capacity of the underlying array
- public int getCapacity() {
- return bufferCapacity_;
- }
-
- // Updates the new capacity. No-op if the new capacity is smaller.
- public void setCapacity(int newCap) {
- if (newCap <= bufferCapacity_) return;
- bufferPtr_ = UnsafeUtil.UNSAFE.reallocateMemory(bufferPtr_, newCap);
- UnsafeUtil.UNSAFE.putLong(stringValPtr_, bufferPtr_);
- bufferCapacity_ = newCap;
- }
-
- // Returns the length of the string
- public int getLength() {
- return UnsafeUtil.UNSAFE.getInt(stringValPtr_ + STRING_VALUE_LEN_OFFSET);
- }
-
- // Updates the length of the string. If the new length is bigger,
- // the additional bytes are undefined.
- public void setSize(int s) {
- setCapacity(s);
- UnsafeUtil.UNSAFE.putInt(stringValPtr_ + 8, s);
- }
-
- // Sets (v[offset], len) to the underlying buffer, growing it as necessary.
- public void set(byte[] v, int offset, int len) {
- setSize(len);
- long strPtr = UnsafeUtil.UNSAFE.getLong(stringValPtr_);
- UnsafeUtil.Copy(strPtr, v, offset, len);
- }
-}
diff --git a/fe/src/main/java/org/apache/impala/hive/executor/ImpalaTextWritable.java b/fe/src/main/java/org/apache/impala/hive/executor/ImpalaTextWritable.java
index 2a8745418..860014b8d 100644
--- a/fe/src/main/java/org/apache/impala/hive/executor/ImpalaTextWritable.java
+++ b/fe/src/main/java/org/apache/impala/hive/executor/ImpalaTextWritable.java
@@ -21,25 +21,17 @@ import org.apache.hadoop.io.Text;
/**
* Impala writable type that implements the Text interface. The data marshalling is
- * handled by the underlying {@link ImpalaStringWritable} object.
+ * handled by {@link JavaUdfDataType#loadStringValueFromNativeHeap(long)}.
*/
-public class ImpalaTextWritable extends Text {
- private final ImpalaStringWritable string_;
+public class ImpalaTextWritable extends Text implements Reloadable {
+ private final long ptr_;
- public ImpalaTextWritable(long ptr) {
- string_ = new ImpalaStringWritable(ptr);
- }
-
- @Override
- public String toString() { return new String(getBytes()); }
- @Override
- public byte[] getBytes() { return string_.getBytes(); }
- @Override
- public int getLength() { return string_.getLength(); }
+ public ImpalaTextWritable(long ptr) { this.ptr_ = ptr; }
@Override
- public void set(byte[] v, int offset, int len) {
- string_.set(v, offset, len);
+ public void reload() {
+ byte[] bytes = JavaUdfDataType.loadStringValueFromNativeHeap(ptr_);
+ super.set(bytes);
}
}
diff --git a/fe/src/main/java/org/apache/impala/hive/executor/JavaUdfDataType.java b/fe/src/main/java/org/apache/impala/hive/executor/JavaUdfDataType.java
index d0824cfbe..981c1d37c 100644
--- a/fe/src/main/java/org/apache/impala/hive/executor/JavaUdfDataType.java
+++ b/fe/src/main/java/org/apache/impala/hive/executor/JavaUdfDataType.java
@@ -23,193 +23,192 @@ import org.apache.hadoop.hive.serde2.io.ShortWritable;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.AbstractPrimitiveWritableObjectInspector;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
import org.apache.impala.catalog.Type;
import org.apache.impala.thrift.TPrimitiveType;
import com.google.common.base.Preconditions;
+import org.apache.impala.util.UnsafeUtil;
+
+// Data types that are supported as return or argument types in Java UDFs.
+public enum JavaUdfDataType {
+ INVALID_TYPE("INVALID_TYPE", TPrimitiveType.INVALID_TYPE),
+ BOOLEAN("BOOLEAN", TPrimitiveType.BOOLEAN),
+ BOOLEAN_WRITABLE("BOOLEAN_WRITABLE", TPrimitiveType.BOOLEAN),
+ TINYINT("TINYINT", TPrimitiveType.TINYINT),
+ BYTE_WRITABLE("BYTE_WRITABLE", TPrimitiveType.TINYINT),
+ SMALLINT("SMALLINT", TPrimitiveType.SMALLINT),
+ SHORT_WRITABLE("SHORT_WRITABLE", TPrimitiveType.SMALLINT),
+ INT("INT", TPrimitiveType.INT),
+ INT_WRITABLE("INT_WRITABLE", TPrimitiveType.INT),
+ BIGINT("BIGINT", TPrimitiveType.BIGINT),
+ LONG_WRITABLE("LONG_WRITABLE", TPrimitiveType.BIGINT),
+ FLOAT("FLOAT", TPrimitiveType.FLOAT),
+ FLOAT_WRITABLE("FLOAT_WRITABLE", TPrimitiveType.FLOAT),
+ DOUBLE("DOUBLE", TPrimitiveType.DOUBLE),
+ DOUBLE_WRITABLE("DOUBLE", TPrimitiveType.DOUBLE),
+ STRING("STRING", TPrimitiveType.STRING),
+ TEXT("TEXT", TPrimitiveType.STRING),
+ BYTES_WRITABLE("BYTES_WRITABLE", TPrimitiveType.STRING),
+ BYTE_ARRAY("BYTE_ARRAY", TPrimitiveType.STRING);
+
+ public static final int STRING_VALUE_LEN_OFFSET = 8;
+
+ private final String description_;
+ private final TPrimitiveType thriftType_;
+
+ private JavaUdfDataType(String description, TPrimitiveType thriftType) {
+ description_ = description;
+ thriftType_ = thriftType;
+ }
- // Data types that are supported as return or argument types in Java UDFs.
- public enum JavaUdfDataType {
- INVALID_TYPE("INVALID_TYPE", TPrimitiveType.INVALID_TYPE),
- BOOLEAN("BOOLEAN", TPrimitiveType.BOOLEAN),
- BOOLEAN_WRITABLE("BOOLEAN_WRITABLE", TPrimitiveType.BOOLEAN),
- TINYINT("TINYINT", TPrimitiveType.TINYINT),
- BYTE_WRITABLE("BYTE_WRITABLE", TPrimitiveType.TINYINT),
- SMALLINT("SMALLINT", TPrimitiveType.SMALLINT),
- SHORT_WRITABLE("SHORT_WRITABLE", TPrimitiveType.SMALLINT),
- INT("INT", TPrimitiveType.INT),
- INT_WRITABLE("INT_WRITABLE", TPrimitiveType.INT),
- BIGINT("BIGINT", TPrimitiveType.BIGINT),
- LONG_WRITABLE("LONG_WRITABLE", TPrimitiveType.BIGINT),
- FLOAT("FLOAT", TPrimitiveType.FLOAT),
- FLOAT_WRITABLE("FLOAT_WRITABLE", TPrimitiveType.FLOAT),
- DOUBLE("DOUBLE", TPrimitiveType.DOUBLE),
- DOUBLE_WRITABLE("DOUBLE", TPrimitiveType.DOUBLE),
- STRING("STRING", TPrimitiveType.STRING),
- TEXT("TEXT", TPrimitiveType.STRING),
- BYTES_WRITABLE("BYTES_WRITABLE", TPrimitiveType.STRING),
- BYTE_ARRAY("BYTE_ARRAY", TPrimitiveType.STRING);
-
- private final String description_;
- private final TPrimitiveType thriftType_;
-
- private JavaUdfDataType(String description, TPrimitiveType thriftType) {
- description_ = description;
- thriftType_ = thriftType;
- }
-
- @Override
- public String toString() { return description_; }
+ @Override
+ public String toString() { return description_; }
- public String getDescription() { return description_; }
+ public String getDescription() { return description_; }
- public TPrimitiveType getPrimitiveType() { return thriftType_; }
+ public TPrimitiveType getPrimitiveType() { return thriftType_; }
- public static JavaUdfDataType[] getTypes(Type typeArray[]) {
- JavaUdfDataType[] types = new JavaUdfDataType[typeArray.length];
- for (int i = 0; i < typeArray.length; ++i) {
- types[i] = getType(typeArray[i]);
- }
- return types;
+ public static JavaUdfDataType[] getTypes(Type[] typeArray) {
+ JavaUdfDataType[] types = new JavaUdfDataType[typeArray.length];
+ for (int i = 0; i < typeArray.length; ++i) {
+ types[i] = getType(typeArray[i]);
}
+ return types;
+ }
- public static JavaUdfDataType[] getTypes(Class<?>[] typeArray) {
- JavaUdfDataType[] types = new JavaUdfDataType[typeArray.length];
- for (int i = 0; i < typeArray.length; ++i) {
- types[i] = getType(typeArray[i]);
- }
- return types;
+ public static JavaUdfDataType[] getTypes(Class<?>[] typeArray) {
+ JavaUdfDataType[] types = new JavaUdfDataType[typeArray.length];
+ for (int i = 0; i < typeArray.length; ++i) {
+ types[i] = getType(typeArray[i]);
}
+ return types;
+ }
- public static JavaUdfDataType getType(Type t) {
- switch (t.getPrimitiveType().toThrift()) {
- case BOOLEAN:
- return JavaUdfDataType.BOOLEAN_WRITABLE;
- case TINYINT:
- return JavaUdfDataType.BYTE_WRITABLE;
- case SMALLINT:
- return JavaUdfDataType.SHORT_WRITABLE;
- case INT:
- return JavaUdfDataType.INT_WRITABLE;
- case BIGINT:
- return JavaUdfDataType.LONG_WRITABLE;
- case FLOAT:
- return JavaUdfDataType.FLOAT_WRITABLE;
- case DOUBLE:
- return JavaUdfDataType.DOUBLE_WRITABLE;
- case STRING:
- return JavaUdfDataType.TEXT;
- case BINARY:
- return JavaUdfDataType.BYTES_WRITABLE;
- default:
- return null;
- }
+ public static JavaUdfDataType getType(Type t) {
+ switch (t.getPrimitiveType().toThrift()) {
+ case BOOLEAN: return JavaUdfDataType.BOOLEAN_WRITABLE;
+ case TINYINT: return JavaUdfDataType.BYTE_WRITABLE;
+ case SMALLINT: return JavaUdfDataType.SHORT_WRITABLE;
+ case INT: return JavaUdfDataType.INT_WRITABLE;
+ case BIGINT: return JavaUdfDataType.LONG_WRITABLE;
+ case FLOAT: return JavaUdfDataType.FLOAT_WRITABLE;
+ case DOUBLE: return JavaUdfDataType.DOUBLE_WRITABLE;
+ case STRING: return JavaUdfDataType.TEXT;
+ case BINARY: return JavaUdfDataType.BYTES_WRITABLE;
+ default: return null;
}
+ }
- public static JavaUdfDataType getType(ObjectInspector oi) {
- // Only primitive objects are supported currently.
- Preconditions.checkState(oi instanceof PrimitiveObjectInspector);
- PrimitiveObjectInspector primOi = (PrimitiveObjectInspector) oi;
- PrimitiveCategory cat = primOi.getPrimitiveCategory();
- boolean writable = primOi.preferWritable();
- switch (cat) {
- case BOOLEAN:
- return writable ? JavaUdfDataType.BOOLEAN_WRITABLE : JavaUdfDataType.BOOLEAN;
- case BYTE:
- return writable ? JavaUdfDataType.BYTE_WRITABLE : JavaUdfDataType.TINYINT;
- case SHORT:
- return writable ? JavaUdfDataType.SHORT_WRITABLE : JavaUdfDataType.SMALLINT;
- case INT:
- return writable ? JavaUdfDataType.INT_WRITABLE : JavaUdfDataType.INT;
- case LONG:
- return writable ? JavaUdfDataType.LONG_WRITABLE : JavaUdfDataType.BIGINT;
- case FLOAT:
- return writable ? JavaUdfDataType.FLOAT_WRITABLE : JavaUdfDataType.FLOAT;
- case DOUBLE:
- return writable ? JavaUdfDataType.DOUBLE_WRITABLE : JavaUdfDataType.DOUBLE;
- case STRING:
- return writable ? JavaUdfDataType.TEXT : JavaUdfDataType.STRING;
- case BINARY:
- return writable ? JavaUdfDataType.BYTES_WRITABLE : JavaUdfDataType.BYTE_ARRAY;
- default:
- return null;
- }
+ public static JavaUdfDataType getType(ObjectInspector oi) {
+ // Only primitive objects are supported currently.
+ Preconditions.checkState(oi instanceof PrimitiveObjectInspector);
+ PrimitiveObjectInspector primOi = (PrimitiveObjectInspector) oi;
+ PrimitiveCategory cat = primOi.getPrimitiveCategory();
+ boolean writable = primOi.preferWritable();
+ switch (cat) {
+ case BOOLEAN:
+ return writable ? JavaUdfDataType.BOOLEAN_WRITABLE : JavaUdfDataType.BOOLEAN;
+ case BYTE:
+ return writable ? JavaUdfDataType.BYTE_WRITABLE : JavaUdfDataType.TINYINT;
+ case SHORT:
+ return writable ? JavaUdfDataType.SHORT_WRITABLE : JavaUdfDataType.SMALLINT;
+ case INT:
+ return writable ? JavaUdfDataType.INT_WRITABLE : JavaUdfDataType.INT;
+ case LONG:
+ return writable ? JavaUdfDataType.LONG_WRITABLE : JavaUdfDataType.BIGINT;
+ case FLOAT:
+ return writable ? JavaUdfDataType.FLOAT_WRITABLE : JavaUdfDataType.FLOAT;
+ case DOUBLE:
+ return writable ? JavaUdfDataType.DOUBLE_WRITABLE : JavaUdfDataType.DOUBLE;
+ case STRING:
+ return writable ? JavaUdfDataType.TEXT : JavaUdfDataType.STRING;
+ case BINARY:
+ return writable ? JavaUdfDataType.BYTES_WRITABLE : JavaUdfDataType.BYTE_ARRAY;
+ default:
+ return null;
}
+ }
- public static JavaUdfDataType getType(Class<?> c) {
- if (c == BooleanWritable.class) {
- return JavaUdfDataType.BOOLEAN_WRITABLE;
- } else if (c == boolean.class || c == Boolean.class) {
- return JavaUdfDataType.BOOLEAN;
- } else if (c == ByteWritable.class) {
- return JavaUdfDataType.BYTE_WRITABLE;
- } else if (c == byte.class || c == Byte.class) {
- return JavaUdfDataType.TINYINT;
- } else if (c == ShortWritable.class) {
- return JavaUdfDataType.SHORT_WRITABLE;
- } else if (c == short.class || c == Short.class) {
- return JavaUdfDataType.SMALLINT;
- } else if (c == IntWritable.class) {
- return JavaUdfDataType.INT_WRITABLE;
- } else if (c == int.class || c == Integer.class) {
- return JavaUdfDataType.INT;
- } else if (c == LongWritable.class) {
- return JavaUdfDataType.LONG_WRITABLE;
- } else if (c == long.class || c == Long.class) {
- return JavaUdfDataType.BIGINT;
- } else if (c == FloatWritable.class) {
- return JavaUdfDataType.FLOAT_WRITABLE;
- } else if (c == float.class || c == Float.class) {
- return JavaUdfDataType.FLOAT;
- } else if (c == DoubleWritable.class) {
- return JavaUdfDataType.DOUBLE_WRITABLE;
- } else if (c == double.class || c == Double.class) {
- return JavaUdfDataType.DOUBLE;
- } else if (c == byte[].class) {
- return JavaUdfDataType.BYTE_ARRAY;
- } else if (c == BytesWritable.class) {
- return JavaUdfDataType.BYTES_WRITABLE;
- } else if (c == Text.class) {
- return JavaUdfDataType.TEXT;
- } else if (c == String.class) {
- return JavaUdfDataType.STRING;
- }
- return JavaUdfDataType.INVALID_TYPE;
+ public static JavaUdfDataType getType(Class<?> c) {
+ if (c == BooleanWritable.class) {
+ return JavaUdfDataType.BOOLEAN_WRITABLE;
+ } else if (c == boolean.class || c == Boolean.class) {
+ return JavaUdfDataType.BOOLEAN;
+ } else if (c == ByteWritable.class) {
+ return JavaUdfDataType.BYTE_WRITABLE;
+ } else if (c == byte.class || c == Byte.class) {
+ return JavaUdfDataType.TINYINT;
+ } else if (c == ShortWritable.class) {
+ return JavaUdfDataType.SHORT_WRITABLE;
+ } else if (c == short.class || c == Short.class) {
+ return JavaUdfDataType.SMALLINT;
+ } else if (c == IntWritable.class) {
+ return JavaUdfDataType.INT_WRITABLE;
+ } else if (c == int.class || c == Integer.class) {
+ return JavaUdfDataType.INT;
+ } else if (c == LongWritable.class) {
+ return JavaUdfDataType.LONG_WRITABLE;
+ } else if (c == long.class || c == Long.class) {
+ return JavaUdfDataType.BIGINT;
+ } else if (c == FloatWritable.class) {
+ return JavaUdfDataType.FLOAT_WRITABLE;
+ } else if (c == float.class || c == Float.class) {
+ return JavaUdfDataType.FLOAT;
+ } else if (c == DoubleWritable.class) {
+ return JavaUdfDataType.DOUBLE_WRITABLE;
+ } else if (c == double.class || c == Double.class) {
+ return JavaUdfDataType.DOUBLE;
+ } else if (c == byte[].class) {
+ return JavaUdfDataType.BYTE_ARRAY;
+ } else if (c == BytesWritable.class) {
+ return JavaUdfDataType.BYTES_WRITABLE;
+ } else if (c == Text.class) {
+ return JavaUdfDataType.TEXT;
+ } else if (c == String.class) {
+ return JavaUdfDataType.STRING;
}
+ return JavaUdfDataType.INVALID_TYPE;
+ }
- public static boolean isSupported(Type t) {
- if (TPrimitiveType.INVALID_TYPE == t.getPrimitiveType().toThrift()) {
- return false;
- }
+ public static boolean isSupported(Type t) {
+ if (TPrimitiveType.INVALID_TYPE == t.getPrimitiveType().toThrift()) {
+ return false;
+ }
- // While BYTES_WRITABLE and BYTE_ARRAY maps to STRING to keep compatibility,
- // BINARY is also accepted (IMPALA-11340).
- if (t.isBinary()) return true;
+ // While BYTES_WRITABLE and BYTE_ARRAY maps to STRING to keep compatibility,
+ // BINARY is also accepted (IMPALA-11340).
+ if (t.isBinary()) return true;
- for(JavaUdfDataType javaType: JavaUdfDataType.values()) {
- if (javaType.getPrimitiveType() == t.getPrimitiveType().toThrift()) {
- return true;
- }
+ for (JavaUdfDataType javaType : JavaUdfDataType.values()) {
+ if (javaType.getPrimitiveType() == t.getPrimitiveType().toThrift()) {
+ return true;
}
- return false;
}
+ return false;
+ }
- public boolean isCompatibleWith(TPrimitiveType t) {
- if (t == getPrimitiveType()) return true;
- if (t == TPrimitiveType.BINARY) {
- // While BYTES_WRITABLE and BYTE_ARRAY maps to STRING to keep compatibility,
- // BINARY is also accepted (IMPALA-11340).
- if (this == BYTE_ARRAY || this == BYTES_WRITABLE) return true;
- }
- return false;
+ public boolean isCompatibleWith(TPrimitiveType t) {
+ if (t == getPrimitiveType()) return true;
+ if (t == TPrimitiveType.BINARY) {
+ // While BYTES_WRITABLE and BYTE_ARRAY maps to STRING to keep compatibility,
+ // BINARY is also accepted (IMPALA-11340).
+ if (this == BYTE_ARRAY || this == BYTES_WRITABLE) return true;
}
+ return false;
}
+ // Returns a backend-allocated string as a Java byte array
+ public static byte[] loadStringValueFromNativeHeap(long ptr) {
+ int length = UnsafeUtil.UNSAFE.getInt(ptr + STRING_VALUE_LEN_OFFSET);
+ byte[] buffer = new byte[length];
+ long srcPtr = UnsafeUtil.UNSAFE.getLong(ptr);
+ UnsafeUtil.Copy(buffer, 0, srcPtr, length);
+ return buffer;
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/hive/executor/ImpalaTextWritable.java b/fe/src/main/java/org/apache/impala/hive/executor/Reloadable.java
similarity index 54%
copy from fe/src/main/java/org/apache/impala/hive/executor/ImpalaTextWritable.java
copy to fe/src/main/java/org/apache/impala/hive/executor/Reloadable.java
index 2a8745418..cb0b9c430 100644
--- a/fe/src/main/java/org/apache/impala/hive/executor/ImpalaTextWritable.java
+++ b/fe/src/main/java/org/apache/impala/hive/executor/Reloadable.java
@@ -17,29 +17,4 @@
package org.apache.impala.hive.executor;
-import org.apache.hadoop.io.Text;
-
-/**
- * Impala writable type that implements the Text interface. The data marshalling is
- * handled by the underlying {@link ImpalaStringWritable} object.
- */
-public class ImpalaTextWritable extends Text {
- private final ImpalaStringWritable string_;
-
- public ImpalaTextWritable(long ptr) {
- string_ = new ImpalaStringWritable(ptr);
- }
-
- @Override
- public String toString() { return new String(getBytes()); }
- @Override
- public byte[] getBytes() { return string_.getBytes(); }
- @Override
- public int getLength() { return string_.getLength(); }
-
- @Override
- public void set(byte[] v, int offset, int len) {
- string_.set(v, offset, len);
- }
-
-}
+public interface Reloadable { void reload(); }
diff --git a/fe/src/test/java/org/apache/impala/hive/executor/UdfExecutorTest.java b/fe/src/test/java/org/apache/impala/hive/executor/UdfExecutorTest.java
index 0ea62bbe2..16f4debda 100644
--- a/fe/src/test/java/org/apache/impala/hive/executor/UdfExecutorTest.java
+++ b/fe/src/test/java/org/apache/impala/hive/executor/UdfExecutorTest.java
@@ -57,7 +57,6 @@ import org.apache.hadoop.hive.ql.udf.UDFUnbase64;
import org.apache.hadoop.hive.ql.udf.UDFUnhex;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBRound;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFUpper;
-import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.impala.catalog.PrimitiveType;
@@ -158,21 +157,30 @@ public class UdfExecutorTest {
Writable createDouble(double v) { return createObject(PrimitiveType.DOUBLE, v); }
Writable createBytes(String v) {
- long ptr = allocate(16);
- UnsafeUtil.UNSAFE.putInt(ptr + 8, 0);
- ImpalaBytesWritable tw = new ImpalaBytesWritable(ptr);
- byte[] array = v.getBytes();
- tw.set(array, 0, array.length);
- return tw;
+ long ptr = allocateStringValue(v);
+ ImpalaBytesWritable bytesWritable = new ImpalaBytesWritable(ptr);
+ bytesWritable.reload();
+ return bytesWritable;
}
Writable createText(String v) {
- long ptr = allocate(16);
- UnsafeUtil.UNSAFE.putInt(ptr + 8, 0);
- ImpalaTextWritable tw = new ImpalaTextWritable(ptr);
- byte[] array = v.getBytes();
- tw.set(array, 0, array.length);
- return tw;
+ long ptr = allocateStringValue(v);
+ ImpalaTextWritable textWritable = new ImpalaTextWritable(ptr);
+ textWritable.reload();
+ return textWritable;
+ }
+
+ private long allocateStringValue(String v) {
+ // Allocate StringValue: sizeof(StringValue) = 8 (pointer) + 4 (length)
+ long ptr = allocate(12);
+ // Setting length
+ UnsafeUtil.UNSAFE.putInt(ptr + 8, v.length());
+ // Allocate buffer for v
+ long stringPtr = allocate(v.length());
+ // Setting string pointer
+ UnsafeUtil.UNSAFE.putLong(ptr, stringPtr);
+ UnsafeUtil.Copy(stringPtr, v.getBytes(), 0, v.length());
+ return ptr;
}
// Returns the primitive type for w
@@ -402,17 +410,16 @@ public class UdfExecutorTest {
} else {
Preconditions.checkState(false);
}
- ImpalaStringWritable sw = new ImpalaStringWritable(r);
- if (Arrays.equals(expectedBytes, sw.getBytes())) break;
+ byte[] bytes = JavaUdfDataType.loadStringValueFromNativeHeap(r);
+ if (Arrays.equals(expectedBytes, bytes)) break;
errMsgs.add("Expected string: " + Bytes.toString(expectedBytes));
- errMsgs.add("Actual string: " + Bytes.toString(sw.getBytes()));
+ errMsgs.add("Actual string: " + Bytes.toString(bytes));
errMsgs.add("Expected bytes: " + Arrays.toString(expectedBytes));
- errMsgs.add("Actual bytes: " + Arrays.toString(sw.getBytes()));
+ errMsgs.add("Actual bytes: " + Arrays.toString(bytes));
break;
}
- default:
- Preconditions.checkArgument(false);
+ default: Preconditions.checkArgument(false);
}
}
diff --git a/java/test-hive-udfs/src/main/java/org/apache/impala/BufferAlteringUdf.java b/java/test-hive-udfs/src/main/java/org/apache/impala/BufferAlteringUdf.java
new file mode 100644
index 000000000..9fbe07a04
--- /dev/null
+++ b/java/test-hive-udfs/src/main/java/org/apache/impala/BufferAlteringUdf.java
@@ -0,0 +1,85 @@
+// 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.impala;
+
+import java.text.ParseException;
+import org.apache.hadoop.hive.ql.exec.UDF;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.Text;
+
+/**
+ * This class intends to test a UDF that manipulates array-backed writables and checks
+ * their data retention behavior. Before IMPALA-11854, every getBytes() method call
+ * directly read the native heap and there was no option to store intermediate results in
+ * the writable buffers. IMPALA-11854 changed this behavior, now every data manipulation
+ * operation on array-backed writables loads the native heap data before the 'evaluate'
+ * phase and stores it in the writable, any subsequent manipulations use the writable's
+ * interface.
+ */
+public class BufferAlteringUdf extends UDF {
+ /**
+ * Increments the first byte by one in a Text and returns the result as a Text
+ */
+ public Text evaluate(Text text) throws ParseException {
+ if ((null == text) || ("".equals(text.toString()))) {
+ return null;
+ }
+ byte[] bytes = text.getBytes();
+
+ incrementByteArray(bytes);
+ return text;
+ }
+
+ /**
+ * Increments the first byte by one in a BytesWritable and returns the result as a
+ * BytesWritable
+ */
+ public BytesWritable evaluate(BytesWritable bytesWritable) throws ParseException {
+ if (null == bytesWritable) {
+ return null;
+ }
+ byte[] bytes = bytesWritable.getBytes();
+
+ incrementByteArray(bytes);
+ return bytesWritable;
+ }
+
+ /**
+ * Copies the source BytesWritable to the target BytesWritable, implicitly resizing it.
+ * After the copy, the first byte of the target BytesWritable is incremented by one.
+ */
+ public BytesWritable evaluate(BytesWritable target, BytesWritable source)
+ throws ParseException {
+ if (null == source || null == target) {
+ return null;
+ }
+ byte[] sourceArray = source.getBytes();
+ target.set(sourceArray, 0, source.getLength());
+
+ byte[] targetArray = target.getBytes();
+ incrementByteArray(targetArray);
+
+ return target;
+ }
+
+ private void incrementByteArray(byte[] array) {
+ if (array.length > 0) {
+ array[0] += 1;
+ }
+ }
+}
diff --git a/java/test-hive-udfs/src/main/java/org/apache/impala/GenericBufferAlteringUdf.java b/java/test-hive-udfs/src/main/java/org/apache/impala/GenericBufferAlteringUdf.java
new file mode 100644
index 000000000..3a6844324
--- /dev/null
+++ b/java/test-hive-udfs/src/main/java/org/apache/impala/GenericBufferAlteringUdf.java
@@ -0,0 +1,105 @@
+// 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.impala;
+
+import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.Text;
+
+/**
+ * This class is the generic version of BufferAlteringUdf. The main purpose is to check
+ * data retention capability after IMPALA-11854.
+ */
+public class GenericBufferAlteringUdf extends GenericUDF {
+ public static final String ARGUMENT_LIST_LENGTH_FORMAT =
+ "This function takes 1 argument, %d argument(s) provided";
+ private PrimitiveCategory argAndRetType_;
+
+ @Override
+ public ObjectInspector initialize(ObjectInspector[] objectInspectors)
+ throws UDFArgumentException {
+ if (objectInspectors.length != 1) {
+ throw new UDFArgumentException(
+ String.format(ARGUMENT_LIST_LENGTH_FORMAT, objectInspectors.length));
+ }
+ if (!(objectInspectors[0] instanceof PrimitiveObjectInspector)) {
+ throw new UDFArgumentException("Found an input that is not a primitive.");
+ }
+ PrimitiveObjectInspector objectInspector =
+ (PrimitiveObjectInspector) objectInspectors[0];
+ argAndRetType_ = objectInspector.getPrimitiveCategory();
+
+ // Return type is same as the input parameter
+ return objectInspector;
+ }
+
+ /**
+ * This function expects a Text or a BytesWritable and increments their underlying byte
+ * array's first element by one.
+ */
+ @Override
+ public Object evaluate(DeferredObject[] deferredObjects) throws HiveException {
+ if (deferredObjects.length != 1) {
+ throw new UDFArgumentException(
+ String.format(ARGUMENT_LIST_LENGTH_FORMAT, deferredObjects.length));
+ }
+ DeferredObject argument = deferredObjects[0];
+ if (argument.get() == null) {
+ return null;
+ }
+ Object object = argument.get();
+
+ switch (argAndRetType_) {
+ case STRING: {
+ if (!(object instanceof Text)) {
+ throw new HiveException("Expected Text but got " + object.getClass());
+ }
+ Text text = (Text) object;
+ byte[] bytes = text.getBytes();
+ incrementByteArray(bytes);
+ return text;
+ }
+ case BINARY: {
+ if (!(object instanceof BytesWritable)) {
+ throw new HiveException("Expected BytesWritable but got " + object.getClass());
+ }
+ BytesWritable bytesWritable = (BytesWritable) object;
+ byte[] bytes = bytesWritable.getBytes();
+ incrementByteArray(bytes);
+ return bytesWritable;
+ }
+ default: throw new IllegalStateException("Unexpected type: " + argAndRetType_);
+ }
+ }
+
+ @Override
+ public String getDisplayString(String[] strings) {
+ return "GenericBufferAltering";
+ }
+
+ private void incrementByteArray(byte[] array) {
+ if (array.length > 0) {
+ array[0] += 1;
+ }
+ }
+}
diff --git a/testdata/workloads/functional-query/queries/QueryTest/generic-java-udf.test b/testdata/workloads/functional-query/queries/QueryTest/generic-java-udf.test
index 3c9fca948..3bd35e8ed 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/generic-java-udf.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/generic-java-udf.test
@@ -335,3 +335,31 @@ symbol='org.apache.impala.TestUdf';
---- CATCH
CatalogException: Variable arguments not supported in Hive UDFs.
====
+---- QUERY
+select increment("a");
+---- TYPES
+STRING
+---- RESULTS
+'b'
+====
+---- QUERY
+select increment(NULL);
+---- TYPES
+STRING
+---- RESULTS
+'NULL'
+====
+---- QUERY
+select increment("");
+---- TYPES
+STRING
+---- RESULTS
+''
+====
+---- QUERY
+select increment(cast("a" as binary));
+---- TYPES
+BINARY
+---- RESULTS
+'b'
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/java-udf.test b/testdata/workloads/functional-query/queries/QueryTest/java-udf.test
index 96769dc77..fb2bc4f2a 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/java-udf.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/java-udf.test
@@ -349,3 +349,38 @@ symbol='org.apache.impala.TestUdf';
---- CATCH
CatalogException: Variable arguments not supported in Hive UDFs.
====
+---- QUERY
+select increment("a");
+---- TYPES
+STRING
+---- RESULTS
+'b'
+====
+---- QUERY
+select increment(NULL);
+---- TYPES
+STRING
+---- RESULTS
+'NULL'
+====
+---- QUERY
+select increment("");
+---- TYPES
+STRING
+---- RESULTS
+''
+====
+---- QUERY
+select increment(cast("a" as binary));
+---- TYPES
+BINARY
+---- RESULTS
+'b'
+====
+---- QUERY
+select copy_and_increment(cast("bbb" as binary), cast("aaaa" as binary));
+---- TYPES
+BINARY
+---- RESULTS
+'baaa'
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/load-generic-java-udfs.test b/testdata/workloads/functional-query/queries/QueryTest/load-generic-java-udfs.test
index ca3415c9c..e029a1a59 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/load-generic-java-udfs.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/load-generic-java-udfs.test
@@ -139,4 +139,12 @@ symbol='org.apache.impala.TestGenericUdfWithJavaReturnTypes';
create function generic_add_java_ret_type(string, string, string) returns string
location '$FILESYSTEM_PREFIX/test-warehouse/impala-hive-udfs.jar'
symbol='org.apache.impala.TestGenericUdfWithJavaReturnTypes';
+
+create function increment(binary) returns binary
+location '$FILESYSTEM_PREFIX/test-warehouse/impala-hive-udfs.jar'
+symbol='org.apache.impala.GenericBufferAlteringUdf';
+
+create function increment(string) returns string
+location '$FILESYSTEM_PREFIX/test-warehouse/impala-hive-udfs.jar'
+symbol='org.apache.impala.GenericBufferAlteringUdf';
====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/load-java-udfs.test b/testdata/workloads/functional-query/queries/QueryTest/load-java-udfs.test
index 45827011b..81c8d579a 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/load-java-udfs.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/load-java-udfs.test
@@ -88,4 +88,16 @@ symbol='org.apache.impala.ReplaceStringUdf';
create function import_nearby_classes(string) returns string
location '$FILESYSTEM_PREFIX/test-warehouse/impala-hive-udfs.jar'
symbol='org.apache.impala.ImportsNearbyClassesUdf';
+
+create function increment(binary) returns binary
+location '$FILESYSTEM_PREFIX/test-warehouse/impala-hive-udfs.jar'
+symbol='org.apache.impala.BufferAlteringUdf';
+
+create function increment(string) returns string
+location '$FILESYSTEM_PREFIX/test-warehouse/impala-hive-udfs.jar'
+symbol='org.apache.impala.BufferAlteringUdf';
+
+create function copy_and_increment(binary, binary) returns binary
+location '$FILESYSTEM_PREFIX/test-warehouse/impala-hive-udfs.jar'
+symbol='org.apache.impala.BufferAlteringUdf';
====