>From Ian Maxon <[email protected]>:
Ian Maxon has uploaded this change for review. (
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20883?usp=email )
Change subject: [ASTERIXDB-3704][RT][EXT] Direct batched UDFs
......................................................................
[ASTERIXDB-3704][RT][EXT] Direct batched UDFs
- user model changes: yes
- storage format changes: yes
- interface changes: yes
Details:
Add an optional WITH parameter to CREATTE FUNCTION, "batched",
which is a default false boolean. When true, feed the UDF an
entire buffer's worth of arguments as an array, rather than
having the entrypoint wrapper call the function for each argument
in the buffer. This function should then return an array of
results. Each element in the array will be treated as a result
to its corresponding argument based on position.
Change-Id: Icf52985cd22e1d8ae93c39d1fac187ce23d316b0
---
M
asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
M asterixdb/asterix-app/src/main/resources/entrypoint.py
M asterixdb/asterix-app/src/test/resources/TweetSent/sentiment.py
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.21.query.sqlpp
M
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.3.ddl.sqlpp
R
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.99.ddl.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/mysentiment_twitter/mysentiment_twitter.14.adm
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalLangIPCProto.java
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/AbstractPythonIPCProto.java
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/PythonMessageBuilder.java
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/AbstractLibrarySocketEvaluator.java
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessagePackUtils.java
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/ExternalAssignBatchRuntimeFactory.java
M
asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CreateFunctionStatement.java
M
asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataTransactionContext.java
M
asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataRecordTypes.java
M
asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entities/Function.java
M
asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entitytupletranslators/FunctionTupleTranslator.java
M
asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/functions/ExternalFunctionCompilerUtil.java
M
asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/functions/ExternalScalarFunctionInfo.java
M
asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/ExternalFunctionInfo.java
M
asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/IExternalFunctionInfo.java
22 files changed, 136 insertions(+), 39 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/83/20883/1
diff --git
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index 0c5cd98..040005e 100644
---
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -3405,7 +3405,7 @@
function = new Function(functionSignature, paramNames,
paramTypes, returnTypeSignature, null,
FunctionKind.SCALAR.toString(), library.getLanguage(),
libraryDatabaseName,
libraryDataverseName, libraryName, externalIdentifier,
cfs.getNullCall(),
- cfs.getDeterministic(), cfs.getResources(),
dependencies, creator, false);
+ cfs.getDeterministic(), cfs.getBatched(),
cfs.getResources(), dependencies, creator, false);
} else {
List<Pair<VarIdentifier, TypeExpression>> paramList =
cfs.getParameters();
int paramCount = paramList.size();
@@ -3468,7 +3468,7 @@
newInlineTypes = Collections.emptyMap();
function = new Function(functionSignature, paramNames, null,
null, cfs.getFunctionBody(),
FunctionKind.SCALAR.toString(),
compilationProvider.getParserFactory().getLanguage(), null,
- null, null, null, null, null, null, dependencies,
creator, cfs.isTransform());
+ null, null, null, null, null, null, null,
dependencies, creator, cfs.isTransform());
}
if (existingFunction == null) {
diff --git a/asterixdb/asterix-app/src/main/resources/entrypoint.py
b/asterixdb/asterix-app/src/main/resources/entrypoint.py
index 918596c..4ff0a9a 100755
--- a/asterixdb/asterix-app/src/main/resources/entrypoint.py
+++ b/asterixdb/asterix-app/src/main/resources/entrypoint.py
@@ -76,7 +76,7 @@
readview = memoryview(readbuf)
- def init(self, module_name, class_name, fn_name):
+ def init(self, module_name, class_name, fn_name, batched):
self.wrapped_module = import_module(module_name)
# do not allow modules to be called that are not part of the uploaded
module
wrapped_fn = None
@@ -93,10 +93,10 @@
if wrapped_fn is None:
raise ImportError(
"Could not find class or function in specified module")
- self.wrapped_fns[self.mid] = wrapped_fn
+ self.wrapped_fns[self.mid] = (wrapped_fn, batched)
def next_tuple(self, *args, key=None):
- return self.wrapped_fns[key](*args)
+ return self.wrapped_fns[key][0](*args)
def check_module_path(self, module):
cwd = Path('.').resolve()
@@ -150,13 +150,15 @@
self.response_buf.seek(0)
args = self.unpacked_msg[1]
module = args[0]
- if len(args) == 3:
+ if len(args) == 4:
clazz = args[1]
fn = args[2]
+ batched = args[3];
else:
clazz = None
fn = args[1]
- self.init(module, clazz, fn)
+ batched = args[2];
+ self.init(module, clazz, fn, batched)
self.packer.pack(int(MessageType.INIT_RSP))
dlen = 1 # just the tag.
resp_len = self.write_header(self.response_buf, dlen)
@@ -177,11 +179,17 @@
if len(self.unpacked_msg) > 1:
args = self.unpacked_msg[1]
if args is not None:
- for arg in args:
+ if self.wrapped_fns[self.mid][1] is True: # batchable function
call
try:
- result[0].append(self.next_tuple(*arg, key=self.mid))
+ result[0].append(self.next_tuple(args,key=self.mid))
except BaseException as e:
result[1].append(traceback.format_exc())
+ else:
+ for arg in args:
+ try:
+ result[0].append(self.next_tuple(*arg,
key=self.mid))
+ except BaseException as e:
+ result[1].append(traceback.format_exc())
self.packer.reset()
self.response_buf.seek(0)
body = msgpack.packb(result)
diff --git a/asterixdb/asterix-app/src/test/resources/TweetSent/sentiment.py
b/asterixdb/asterix-app/src/test/resources/TweetSent/sentiment.py
index 66545ae..550f234 100644
--- a/asterixdb/asterix-app/src/test/resources/TweetSent/sentiment.py
+++ b/asterixdb/asterix-app/src/test/resources/TweetSent/sentiment.py
@@ -32,3 +32,8 @@
if args is None:
return 2
return self.pipeline.predict([args])[0].item()
+
+ def sentiment_batch(self, args):
+ if args is None:
+ return 2
+ return self.pipeline.predict(args).tolist()
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.21.query.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.21.query.sqlpp
new file mode 100644
index 0000000..2303020
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.21.query.sqlpp
@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+use externallibtest;
+
+select value count(sentimentBatch(t.text))
+from Tweet t;
+
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.3.ddl.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.3.ddl.sqlpp
index 00838de..c08a5a9 100644
---
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.3.ddl.sqlpp
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.3.ddl.sqlpp
@@ -20,3 +20,7 @@
create function sentiment(s)
as "sentiment", "TweetSent.sentiment" at testlib;
+
+create function sentimentBatch(s)
+ as "sentiment", "TweetSent.sentiment_batch" at testlib
+ with {"batched": true};
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.21.ddl.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.99.ddl.sqlpp
similarity index 100%
rename from
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.21.ddl.sqlpp
rename to
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/mysentiment_twitter/mysentiment_twitter.99.ddl.sqlpp
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/mysentiment_twitter/mysentiment_twitter.14.adm
b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/mysentiment_twitter/mysentiment_twitter.14.adm
new file mode 100644
index 0000000..e9c02da
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/mysentiment_twitter/mysentiment_twitter.14.adm
@@ -0,0 +1 @@
+5000
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalLangIPCProto.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalLangIPCProto.java
index 35e5961..1a14bca 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalLangIPCProto.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalLangIPCProto.java
@@ -79,7 +79,7 @@
void helo() throws IOException, AsterixException;
- long init(String module, String clazz, String fn) throws IOException,
AsterixException;
+ long init(String module, String clazz, String fn, boolean batched) throws
IOException, AsterixException;
ByteBuffer call(long functionId, IAType[] argTypes, IValueReference[]
argValues, boolean nullCall)
throws IOException, AsterixException;
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/AbstractPythonIPCProto.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/AbstractPythonIPCProto.java
index 00d1dcc..aee5b5a 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/AbstractPythonIPCProto.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/AbstractPythonIPCProto.java
@@ -42,6 +42,7 @@
public abstract class AbstractPythonIPCProto {
public static final int HEADER_SIZE_LEN_INCLUSIVE = 21;
+ public static final int DEFAULT_BUF_SIZE = 32768;
protected final PythonMessageBuilder messageBuilder;
protected final DataOutputStream sockOut;
protected final ArrayBufferInput unpackerInput;
@@ -50,7 +51,7 @@
protected final PointableAllocator pointableAllocator;
protected final MsgPackPointableVisitor pointableVisitor;
private final ByteBuffer headerBuffer =
ByteBuffer.allocate(HEADER_SIZE_LEN_INCLUSIVE);
- protected ByteBuffer recvBuffer = ByteBuffer.allocate(32768);
+ protected ByteBuffer recvBuffer = ByteBuffer.allocate(DEFAULT_BUF_SIZE);
protected long routeId;
protected Pair<ByteBuffer, Exception> bufferBox;
protected long maxFunctionId;
@@ -81,13 +82,13 @@
}
}
- public long init(String module, String clazz, String fn) throws
IOException, AsterixException {
+ public long init(String module, String clazz, String fn, boolean batched)
throws IOException, AsterixException {
long functionId = maxFunctionId++;
recvBuffer.clear();
recvBuffer.position(0);
recvBuffer.limit(0);
messageBuilder.reset();
- messageBuilder.init(module, clazz, fn);
+ messageBuilder.init(module, clazz, fn, batched);
sendHeader(functionId, messageBuilder.getLength());
sendMsg();
receiveMsg();
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/PythonMessageBuilder.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/PythonMessageBuilder.java
index 20f8306..0434436 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/PythonMessageBuilder.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/PythonMessageBuilder.java
@@ -99,23 +99,25 @@
MessagePackUtils.packFixStr(buf, "QUIT");
}
- public void init(final String module, final String clazz, final String fn)
throws HyracksDataException {
+ public void init(final String module, final String clazz, final String fn,
final boolean batched)
+ throws HyracksDataException {
this.type = MessageType.INIT;
- // sum(string lengths) + 2 from fix array tag and message type
+ // sum(string lengths) + 3 from fix array tag, message type and
batched flag
if (clazz != null) {
dataLength =
- PythonMessageBuilder.getStringLength(module) +
getStringLength(clazz) + getStringLength(fn) + 2;
+ PythonMessageBuilder.getStringLength(module) +
getStringLength(clazz) + getStringLength(fn) + 3;
} else {
- dataLength = PythonMessageBuilder.getStringLength(module) +
getStringLength(fn) + 2;
+ dataLength = PythonMessageBuilder.getStringLength(module) +
getStringLength(fn) + 3;
}
packHeader();
- int numArgs = clazz == null ? 2 : 3;
+ int numArgs = clazz == null ? 3 : 4;
MessagePackUtils.packFixArrayHeader(buf, (byte) numArgs);
MessagePackUtils.packStr(buf, module);
if (clazz != null) {
MessagePackUtils.packStr(buf, clazz);
}
MessagePackUtils.packStr(buf, fn);
+ MessagePackUtils.packBoolean(buf, batched);
}
public void call(int numArgs, int len) throws HyracksDataException {
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/AbstractLibrarySocketEvaluator.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/AbstractLibrarySocketEvaluator.java
index 6fcfdcf..bbf2166 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/AbstractLibrarySocketEvaluator.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/AbstractLibrarySocketEvaluator.java
@@ -68,7 +68,7 @@
clazz = null;
fn = externalIdent1;
}
- return proto.init(packageModule, clazz, fn);
+ return proto.init(packageModule, clazz, fn, finfo.isBatched());
}
@Override
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessagePackUtils.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessagePackUtils.java
index 2377f9a..e5bd9df 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessagePackUtils.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessagePackUtils.java
@@ -18,9 +18,11 @@
*/
package org.apache.asterix.external.library.msgpack;
+import static org.msgpack.core.MessagePack.Code.FALSE;
import static org.msgpack.core.MessagePack.Code.FIXARRAY_PREFIX;
import static org.msgpack.core.MessagePack.Code.FIXSTR_PREFIX;
import static org.msgpack.core.MessagePack.Code.STR32;
+import static org.msgpack.core.MessagePack.Code.TRUE;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
@@ -70,4 +72,8 @@
public static void packFixArrayHeader(ByteBuffer buf, byte numObj) {
buf.put((byte) (FIXARRAY_PREFIX + (0x0F & numObj)));
}
+
+ public static void packBoolean(ByteBuffer buf, boolean val) {
+ buf.put(val ? TRUE : FALSE);
+ }
}
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/ExternalAssignBatchRuntimeFactory.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/ExternalAssignBatchRuntimeFactory.java
index 5f8a3f0..92e250a 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/ExternalAssignBatchRuntimeFactory.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/ExternalAssignBatchRuntimeFactory.java
@@ -199,6 +199,20 @@
}
}
+ private void writeAdmToMsgpackBuf(int[] cols, ByteBuffer buffer,
int func, boolean argArray)
+ throws IOException {
+ if (argArray && cols.length > 0) {
+ argHolders.get(func).getDataOutput().writeByte(ARRAY16);
+ argHolders.get(func).getDataOutput().writeShort((short)
cols.length);
+ }
+ for (int colIdx = 0; colIdx < cols.length; colIdx++) {
+ ref.set(buffer.array(), tRef.getFieldStart(cols[colIdx]),
tRef.getFieldLength(cols[colIdx]));
+
IExternalLangIPCProto.visitValueRef(fnDescs[func].getArgumentTypes()[colIdx],
+ argHolders.get(func).getDataOutput(), ref,
pointableAllocator, pointableVisitor,
+ fnDescs[func].getFunctionInfo().getNullCall());
+ }
+ }
+
@Override
public void nextFrame(ByteBuffer buffer) throws
HyracksDataException {
/*TODO: this whole transposition stuff is not necessary
@@ -211,6 +225,7 @@
//build columns of arguments for each function
for (int t = 0; t < numTuples; t++) {
for (int func = 0; func < fnArgColumns.length; func++)
{
+ boolean batchedFn =
fnDescs[func].getFunctionInfo().isBatched();
tRef.reset(tAccess, t);
int[] cols = fnArgColumns[func];
//TODO: switch between fixarray/array16/array32
where appropriate
@@ -225,17 +240,7 @@
}
}
if (argumentStatus == ATypeTag.TYPE) {
- if (cols.length > 0) {
-
argHolders.get(func).getDataOutput().writeByte(ARRAY16);
-
argHolders.get(func).getDataOutput().writeShort((short) cols.length);
- }
- for (int colIdx = 0; colIdx < cols.length;
colIdx++) {
- ref.set(buffer.array(),
tRef.getFieldStart(cols[colIdx]),
- tRef.getFieldLength(cols[colIdx]));
-
IExternalLangIPCProto.visitValueRef(fnDescs[func].getArgumentTypes()[colIdx],
-
argHolders.get(func).getDataOutput(), ref, pointableAllocator,
- pointableVisitor,
fnDescs[func].getFunctionInfo().getNullCall());
- }
+ writeAdmToMsgpackBuf(cols, buffer, func,
!batchedFn);
} else {
numCalls[func]--;
}
@@ -268,6 +273,9 @@
//wrapper for results and warnings arrays. always
length 2
consumeAndGetBatchLength(resultBuf);
int numResults = (int)
consumeAndGetBatchLength(resultBuf);
+ if
(fnDescs[argHolderIdx].getFunctionInfo().isBatched()) {
+ numResults = (int)
consumeAndGetBatchLength(resultBuf);
+ }
resultholder.getSecond().set(numResults);
} else {
if (ctx.getWarningCollector().shouldWarn()) {
diff --git
a/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CreateFunctionStatement.java
b/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CreateFunctionStatement.java
index 74b7bbd..25acf91 100644
---
a/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CreateFunctionStatement.java
+++
b/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CreateFunctionStatement.java
@@ -47,6 +47,8 @@
private static final String NULLCALL_FIELD_NAME = "null-call";
private static final boolean NULLCALL_DEFAULT = false;
private static final String DETERMINISTIC_FIELD_NAME = "deterministic";
+ private static final String BATCHED_FIELD_NAME = "batched";
+ private static final boolean BATCHED_DEFAULT = false;
private static final boolean DETERMINISTIC_DEFAULT = true;
private static final String RESOURCES_FIELD_NAME = "resources";
@@ -162,6 +164,11 @@
return deterministic != null ? deterministic : DETERMINISTIC_DEFAULT;
}
+ public boolean getBatched() throws CompilationException {
+ Boolean batched = getBooleanOption(BATCHED_FIELD_NAME);
+ return batched != null ? batched : BATCHED_DEFAULT;
+ }
+
private Boolean getBooleanOption(String optionName) throws
CompilationException {
IAdmNode value = getOption(optionName);
if (value == null) {
diff --git
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataTransactionContext.java
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataTransactionContext.java
index 483c4f2..3efc536 100644
---
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataTransactionContext.java
+++
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataTransactionContext.java
@@ -209,7 +209,7 @@
public void dropFunction(FunctionSignature signature) {
Function function = new Function(signature, null, null, null, null,
null, null, null, null, null, null, false,
- false, null, null, null, false);
+ false, false, null, null, null, false);
droppedCache.addFunctionIfNotExists(function);
logAndApply(new MetadataLogicalOperation(function, false));
}
diff --git
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataRecordTypes.java
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataRecordTypes.java
index 820cbb9..888b48d 100644
---
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataRecordTypes.java
+++
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataRecordTypes.java
@@ -267,6 +267,7 @@
//open types
public static final String FUNCTION_ARECORD_FUNCTION_RESOURCES_FIELD_NAME
= "Resources";
public static final String FUNCTION_ARECORD_FUNCTION_NULLCALL_FIELD_NAME =
"NullCall";
+ public static final String FUNCTION_ARECORD_FUNCTION_BATCHED_FIELD_NAME =
"Batched";
public static final String
FUNCTION_ARECORD_FUNCTION_DETERMINISTIC_FIELD_NAME = "Deterministic";
public static final String FUNCTION_ARECORD_FUNCTION_PARAMTYPES_FIELD_NAME
= "ParamTypes";
public static final String
FUNCTION_ARECORD_FUNCTION_EXTERNAL_IDENTIFIER_FIELD_NAME = "ExternalIdentifier";
diff --git
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entities/Function.java
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entities/Function.java
index c19b542..50eeff08 100644
---
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entities/Function.java
+++
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entities/Function.java
@@ -48,6 +48,7 @@
private final List<String> externalIdentifier;
private final Boolean deterministic; // null for SQL++ and AQL functions
private final Boolean nullCall; // null for SQL++ and AQL functions
+ private final Boolean batched; // null for SQL++ and AQL functions
private final Map<String, String> resources;
private final List<List<DependencyFullyQualifiedName>> dependencies;
private final Creator creator;
@@ -56,8 +57,9 @@
public Function(FunctionSignature signature, List<String> paramNames,
List<TypeSignature> paramTypes,
TypeSignature returnType, String functionBody, String
functionKind, String language,
String libraryDatabaseName, DataverseName libraryDataverseName,
String libraryName,
- List<String> externalIdentifier, Boolean nullCall, Boolean
deterministic, Map<String, String> resources,
- List<List<DependencyFullyQualifiedName>> dependencies, Creator
creator, boolean transform) {
+ List<String> externalIdentifier, Boolean nullCall, Boolean
deterministic, Boolean batched,
+ Map<String, String> resources,
List<List<DependencyFullyQualifiedName>> dependencies, Creator creator,
+ boolean transform) {
this.signature = signature;
this.paramNames = paramNames;
this.paramTypes = paramTypes;
@@ -70,6 +72,7 @@
this.libraryName = libraryName;
this.externalIdentifier = externalIdentifier;
this.nullCall = nullCall;
+ this.batched = batched;
this.deterministic = deterministic;
this.resources = resources == null ? Collections.emptyMap() :
resources;
this.dependencies = dependencies == null
@@ -154,6 +157,10 @@
return nullCall;
}
+ public Boolean getBatched() {
+ return batched;
+ }
+
public Boolean getDeterministic() {
return deterministic;
}
diff --git
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entitytupletranslators/FunctionTupleTranslator.java
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entitytupletranslators/FunctionTupleTranslator.java
index 1eddccc..2dcdbb1 100644
---
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entitytupletranslators/FunctionTupleTranslator.java
+++
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/entitytupletranslators/FunctionTupleTranslator.java
@@ -28,6 +28,7 @@
import static
org.apache.asterix.metadata.bootstrap.MetadataRecordTypes.FIELD_NAME_RETURN_TYPE_DATAVERSE_NAME;
import static
org.apache.asterix.metadata.bootstrap.MetadataRecordTypes.FIELD_NAME_TYPE;
import static
org.apache.asterix.metadata.bootstrap.MetadataRecordTypes.FIELD_NAME_VALUE;
+import static
org.apache.asterix.metadata.bootstrap.MetadataRecordTypes.FUNCTION_ARECORD_FUNCTION_BATCHED_FIELD_NAME;
import static
org.apache.asterix.metadata.bootstrap.MetadataRecordTypes.FUNCTION_ARECORD_FUNCTION_DETERMINISTIC_FIELD_NAME;
import static
org.apache.asterix.metadata.bootstrap.MetadataRecordTypes.FUNCTION_ARECORD_FUNCTION_EXTERNAL_IDENTIFIER_FIELD_NAME;
import static
org.apache.asterix.metadata.bootstrap.MetadataRecordTypes.FUNCTION_ARECORD_FUNCTION_LIBRARY_FIELD_NAME;
@@ -185,9 +186,11 @@
}
Boolean nullCall = null;
+ Boolean batched = null;
Boolean deterministic = null;
if (externalIdentifier != null) {
nullCall = getBoolean(functionRecord,
FUNCTION_ARECORD_FUNCTION_NULLCALL_FIELD_NAME);
+ batched = getBoolean(functionRecord,
FUNCTION_ARECORD_FUNCTION_BATCHED_FIELD_NAME);
deterministic = getBoolean(functionRecord,
FUNCTION_ARECORD_FUNCTION_DETERMINISTIC_FIELD_NAME);
}
@@ -214,7 +217,7 @@
}
return new Function(signature, paramNames, paramTypes, returnType,
definition, functionKind, language,
libraryDatabaseName, libraryDataverseName, libraryName,
externalIdentifier, nullCall, deterministic,
- resources, dependencies, creator, transform);
+ batched, resources, dependencies, creator, transform);
}
private List<TypeSignature> getParamTypes(ARecord functionRecord, String
functionDatabaseName,
@@ -585,6 +588,18 @@
recordBuilder.addField(fieldName, fieldValue);
}
+ protected void writeBatched(Function function) throws HyracksDataException
{
+ if (function.getBatched() == null) {
+ return;
+ }
+ fieldName.reset();
+ aString.setValue(FUNCTION_ARECORD_FUNCTION_BATCHED_FIELD_NAME);
+ stringSerde.serialize(aString, fieldName.getDataOutput());
+ fieldValue.reset();
+ booleanSerde.serialize(ABoolean.valueOf(function.getBatched()),
fieldValue.getDataOutput());
+ recordBuilder.addField(fieldName, fieldValue);
+ }
+
protected void writeDeterministic(Function function) throws
HyracksDataException {
if (function.getDeterministic() == null) {
return;
diff --git
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/functions/ExternalFunctionCompilerUtil.java
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/functions/ExternalFunctionCompilerUtil.java
index 765c2d3..e36c1f9 100644
---
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/functions/ExternalFunctionCompilerUtil.java
+++
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/functions/ExternalFunctionCompilerUtil.java
@@ -80,12 +80,11 @@
throw new AsterixException(ErrorCode.METADATA_ERROR,
function.getSignature().toString());
}
- //TODO(DB): review
return new
ExternalScalarFunctionInfo(function.getSignature().createFunctionIdentifier(),
paramTypes,
returnType, typeComputer, lang,
new Namespace(function.getLibraryDatabaseName(),
function.getLibraryDataverseName()),
function.getLibraryName(), function.getExternalIdentifier(),
function.getResources(), deterministic,
- function.getNullCall());
+ function.getNullCall(), function.getBatched());
}
private static IFunctionInfo getUnnestFunctionInfo(MetadataProvider
metadataProvider, Function function) {
diff --git
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/functions/ExternalScalarFunctionInfo.java
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/functions/ExternalScalarFunctionInfo.java
index 441d118..fb38f8f 100644
---
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/functions/ExternalScalarFunctionInfo.java
+++
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/functions/ExternalScalarFunctionInfo.java
@@ -35,8 +35,9 @@
public ExternalScalarFunctionInfo(FunctionIdentifier fid, List<IAType>
parameterTypes, IAType returnType,
IResultTypeComputer rtc, ExternalFunctionLanguage language,
Namespace libraryNamespace, String libraryName,
- List<String> externalIdentifier, Map<String, String> resources,
boolean deterministic, boolean nullCall) {
+ List<String> externalIdentifier, Map<String, String> resources,
boolean deterministic, boolean nullCall,
+ boolean batched) {
super(fid, FunctionKind.SCALAR, parameterTypes, returnType, rtc,
language, libraryNamespace, libraryName,
- externalIdentifier, resources, deterministic, nullCall);
+ externalIdentifier, resources, deterministic, nullCall,
batched);
}
}
diff --git
a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/ExternalFunctionInfo.java
b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/ExternalFunctionInfo.java
index 700c9f2..e67eac0 100644
---
a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/ExternalFunctionInfo.java
+++
b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/ExternalFunctionInfo.java
@@ -41,11 +41,12 @@
private final List<String> externalIdentifier;
private final Map<String, String> resources;
private final boolean nullCall;
+ private final boolean batched;
public ExternalFunctionInfo(FunctionIdentifier fid, FunctionKind kind,
List<IAType> parameterTypes,
IAType returnType, IResultTypeComputer rtc,
ExternalFunctionLanguage language, Namespace libraryNamespace,
String libraryName, List<String> externalIdentifier, Map<String,
String> resources, boolean deterministic,
- boolean nullCall) {
+ boolean nullCall, boolean batched) {
super(fid, rtc, deterministic);
this.kind = kind;
this.parameterTypes = parameterTypes;
@@ -56,6 +57,7 @@
this.externalIdentifier = externalIdentifier;
this.resources = resources;
this.nullCall = nullCall;
+ this.batched = batched;
}
@Override
@@ -106,4 +108,9 @@
public boolean isExternal() {
return true;
}
+
+ @Override
+ public boolean isBatched() {
+ return batched;
+ }
}
diff --git
a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/IExternalFunctionInfo.java
b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/IExternalFunctionInfo.java
index e2662e6..a189905 100644
---
a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/IExternalFunctionInfo.java
+++
b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/functions/IExternalFunctionInfo.java
@@ -49,4 +49,6 @@
Map<String, String> getResources();
boolean getNullCall();
+
+ boolean isBatched();
}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20883?usp=email
To unsubscribe, or for help writing mail filters, visit
https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: Icf52985cd22e1d8ae93c39d1fac187ce23d316b0
Gerrit-Change-Number: 20883
Gerrit-PatchSet: 1
Gerrit-Owner: Ian Maxon <[email protected]>
