This is an automated email from the ASF dual-hosted git repository. imaxon pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/asterixdb.git
The following commit(s) were added to refs/heads/master by this push: new 8d6cf75 [ASTERIXDB-2895][RT] Vsize buffers in PyUDF IPC 8d6cf75 is described below commit 8d6cf7574430de4f4049fbf5e6726cf2ede5d8aa Author: Ian Maxon <ian@maxons.email> AuthorDate: Tue Oct 26 00:15:48 2021 -0700 [ASTERIXDB-2895][RT] Vsize buffers in PyUDF IPC - user mode changes: no - storage format changes: no - interface changes: no Details: - Convert most uses of ByteBuffer to ArrayBackedValueStorage so that the size of the buffer can grow arbitrarily with the data - Convert ADM-to-Msgpack serialiation to use IVisitablePointable - Convert all serialization interfaces that used ByteBuffer to use DataOutput instead - Fix UTF8 encoding bugs by using StandardToModifiedUTF8DataOutput - Adapt some of the UTF8 printing code to be used for UTF8 output to msgpack - Fix CSV output printer to not ignore surrogate pairs - Fix ASTERIXDB-29773 (returned records from PyUDF aren't sorted) Change-Id: Ic95e592b42139b4750af8bb20291f926b3c973e2 Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/12643 Reviewed-by: Wael Alkowaileet <wael....@gmail.com> Integration-Tests: Jenkins <jenk...@fulliautomatix.ics.uci.edu> Tested-by: Jenkins <jenk...@fulliautomatix.ics.uci.edu> Contrib: Ian Maxon <ima...@uci.edu> --- .../big_object_roundtrip_20M.1.ddl.sqlpp} | 15 +- .../big_object_roundtrip_20M.2.update.sqlpp} | 11 +- .../big_object_roundtrip_20M.3.lib.sqlpp} | 8 +- .../big_object_roundtrip_20M.4.ddl.sqlpp} | 9 +- .../big_object_roundtrip_20M.5.query.sqlpp} | 9 +- .../py_nested_access.4.query.sqlpp | 3 +- .../type_validation.2.ddl.sqlpp | 3 + .../big_object_roundtrip_20M.1.adm.template | 1 + .../mysentiment_twitter/mysentiment_twitter.13.adm | 42 +-- .../resources/runtimets/testsuite_it_python.xml | 5 + .../asterix/common/exceptions/ErrorCode.java | 1 + .../src/main/resources/asx_errormsg/en.properties | 1 + .../builders/ListLikeNumericArrayFactory.java} | 25 +- .../StandardToModifiedUTF8DataOutputFactory.java | 37 +++ .../asterix/external/ipc/PythonIPCProto.java | 163 ++++++++-- .../asterix/external/ipc/PythonMessageBuilder.java | 63 ++-- .../ExternalScalarPythonFunctionEvaluator.java | 25 +- .../external/library/JavaFunctionHelper.java | 8 +- .../library/{TypeInfo.java => JavaTypeInfo.java} | 4 +- .../asterix/external/library/PyTypeInfo.java} | 43 ++- .../external/library/PythonLibraryEvaluator.java | 36 +- .../external/library/java/JObjectAccessors.java | 10 +- .../library/java/JObjectPointableVisitor.java | 13 +- .../library/msgpack/IMsgPackAccessor.java} | 11 +- .../library/msgpack/IMsgPackListAccessor.java} | 13 +- .../library/msgpack/IMsgPackRecordAccessor.java} | 13 +- .../external/library/msgpack/MessagePackUtils.java | 73 +++++ .../library/msgpack/MessagePackerFromADM.java | 362 --------------------- .../library/msgpack/MessageUnpackerToADM.java | 191 ++++++----- .../external/library/msgpack/MsgPackAccessors.java | 275 ++++++++++++++++ .../library/msgpack/MsgPackPointableVisitor.java | 77 +++++ .../ExternalAssignBatchRuntimeFactory.java | 77 ++--- .../data/nontagged/printers/PrintTools.java | 45 ++- 33 files changed, 980 insertions(+), 692 deletions(-) diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.1.ddl.sqlpp similarity index 75% copy from asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp copy to asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.1.ddl.sqlpp index 8041b5e..d3bf31c 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.1.ddl.sqlpp @@ -16,10 +16,15 @@ * specific language governing permissions and limitations * under the License. */ -use externallibtest; -create function typeValidation(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib; +drop dataverse testdv2 if exists; +create dataverse testdv2; +use testdv2; -create function typeValidationNullCall(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib with {"null-call": true}; +create type testtype as closed { + id: int64, + name: string, + hobbies: {{string}} +}; + +create dataset testds(testtype) primary key id; diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.2.update.sqlpp similarity index 75% copy from asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp copy to asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.2.update.sqlpp index 8041b5e..0353284 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.2.update.sqlpp @@ -16,10 +16,11 @@ * specific language governing permissions and limitations * under the License. */ -use externallibtest; -create function typeValidation(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib; +use testdv2; -create function typeValidationNullCall(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib with {"null-call": true}; +set `compiler.sortmemory` "64MB"; + +load dataset testds +using localfs +(("path"="asterix_nc1://target/data/big-object/big_object_20M.adm"),("format"="adm")); diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.3.lib.sqlpp similarity index 75% copy from asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp copy to asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.3.lib.sqlpp index 8041b5e..923dcb2 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.3.lib.sqlpp @@ -16,10 +16,4 @@ * specific language governing permissions and limitations * under the License. */ -use externallibtest; - -create function typeValidation(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib; - -create function typeValidationNullCall(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib with {"null-call": true}; +install testdv2 testlib python admin admin target/TweetSent.pyz diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.4.ddl.sqlpp similarity index 75% copy from asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp copy to asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.4.ddl.sqlpp index 8041b5e..8b304fc 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.4.ddl.sqlpp @@ -16,10 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -use externallibtest; + USE testdv2; -create function typeValidation(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib; - -create function typeValidationNullCall(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib with {"null-call": true}; +create function roundtrip(s) as "roundtrip", + "Tests.roundstr" at testlib; diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.5.query.sqlpp similarity index 75% copy from asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp copy to asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.5.query.sqlpp index 8041b5e..fdb285e 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.5.query.sqlpp @@ -16,10 +16,9 @@ * specific language governing permissions and limitations * under the License. */ -use externallibtest; -create function typeValidation(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib; +use testdv2; -create function typeValidationNullCall(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib with {"null-call": true}; +select value roundtrip(d) +from testds d +where d.id=1; diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/py_nested_access/py_nested_access.4.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/py_nested_access/py_nested_access.4.query.sqlpp index e17f03f..72dc83c 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/py_nested_access/py_nested_access.4.query.sqlpp +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/py_nested_access/py_nested_access.4.query.sqlpp @@ -28,4 +28,5 @@ use test; select element result from Animals as test with result as roundtrip(test)[0].class.fullClassification.lower.lower.lower.lower.lower.lower.Species -order by result; +order by result +; \ No newline at end of file diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp index 8041b5e..ff61810 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp @@ -23,3 +23,6 @@ create function typeValidation(a, b, c, d, e, f, g, h) create function typeValidationNullCall(a, b, c, d, e, f, g, h) as "roundtrip", "Tests.roundtrip" at testlib with {"null-call": true}; + +create function stringTest(s) as "roundtrip", + "Tests.roundtrip" at testlib; diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.1.adm.template b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.1.adm.template new file mode 100644 index 0000000..a0cc7aa --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/big_object_roundtrip_20M/big_object_roundtrip_20M.1.adm.template @@ -0,0 +1 @@ +{ "id": 1, "name": "Person One", "hobbies": [ "%lorembytes:20971520%" ] } diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/mysentiment_twitter/mysentiment_twitter.13.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/mysentiment_twitter/mysentiment_twitter.13.adm index 65a7c81..c960344 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/mysentiment_twitter/mysentiment_twitter.13.adm +++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-library/mysentiment_twitter/mysentiment_twitter.13.adm @@ -1,9 +1,9 @@ { "id": 670301227662491648, "len": 20, "sent": 1 } { "id": 670301227553566720, "len": 139, "sent": 0 } { "id": 670301227041857536, "len": 112, "sent": 0 } -{ "id": 670301227037519876, "len": 33, "sent": 0 } +{ "id": 670301227037519876, "len": 34, "sent": 0 } { "id": 670301226987159552, "len": 57, "sent": 0 } -{ "id": 670301226513391616, "len": 28, "sent": 1 } +{ "id": 670301226513391616, "len": 29, "sent": 1 } { "id": 670301226337202180, "len": 77, "sent": 1 } { "id": 670301226190278656, "len": 25, "sent": 0 } { "id": 670301225959579648, "len": 112, "sent": 1 } @@ -15,25 +15,25 @@ { "id": 670301225162661889, "len": 28, "sent": 1 } { "id": 670301224885837824, "len": 63, "sent": 0 } { "id": 670301224814698496, "len": 59, "sent": 0 } -{ "id": 670301224709849090, "len": 33, "sent": 1 } +{ "id": 670301224709849090, "len": 37, "sent": 1 } { "id": 670301224684556288, "len": 21, "sent": 0 } { "id": 670301224680480768, "len": 39, "sent": 0 } { "id": 670301224348946433, "len": 64, "sent": 1 } { "id": 670301224261058560, "len": 61, "sent": 1 } -{ "id": 670301224231690240, "len": 33, "sent": 0 } -{ "id": 670301224214794240, "len": 33, "sent": 0 } +{ "id": 670301224231690240, "len": 34, "sent": 0 } +{ "id": 670301224214794240, "len": 41, "sent": 0 } { "id": 670301223753351168, "len": 105, "sent": 1 } { "id": 670301223426367488, "len": 23, "sent": 0 } -{ "id": 670301223216545792, "len": 31, "sent": 0 } +{ "id": 670301223216545792, "len": 34, "sent": 0 } { "id": 670301223182974976, "len": 34, "sent": 1 } { "id": 670301223128535041, "len": 21, "sent": 0 } { "id": 670301222759301121, "len": 132, "sent": 0 } { "id": 670301222734307329, "len": 110, "sent": 1 } { "id": 670301222717419520, "len": 81, "sent": 0 } { "id": 670301222318936064, "len": 110, "sent": 1 } -{ "id": 670301222302150657, "len": 131, "sent": 0 } +{ "id": 670301222302150657, "len": 135, "sent": 0 } { "id": 670301222222602240, "len": 43, "sent": 1 } -{ "id": 670301222113517568, "len": 27, "sent": 0 } +{ "id": 670301222113517568, "len": 29, "sent": 0 } { "id": 670301221836615680, "len": 44, "sent": 1 } { "id": 670301221719310336, "len": 28, "sent": 0 } { "id": 670301221442486272, "len": 34, "sent": 0 } @@ -44,13 +44,13 @@ { "id": 670301220305821696, "len": 140, "sent": 0 } { "id": 670301220247072770, "len": 83, "sent": 1 } { "id": 670301220196626432, "len": 36, "sent": 0 } -{ "id": 670301220079312901, "len": 31, "sent": 1 } -{ "id": 670301219949305857, "len": 70, "sent": 1 } +{ "id": 670301220079312901, "len": 32, "sent": 1 } +{ "id": 670301219949305857, "len": 94, "sent": 1 } { "id": 670301219739574273, "len": 131, "sent": 1 } { "id": 670301219206877184, "len": 27, "sent": 0 } { "id": 670301219139620864, "len": 124, "sent": 0 } -{ "id": 670301218737123328, "len": 124, "sent": 0 } -{ "id": 670301218640531458, "len": 31, "sent": 1 } +{ "id": 670301218737123328, "len": 126, "sent": 0 } +{ "id": 670301218640531458, "len": 33, "sent": 1 } { "id": 670301218598756352, "len": 47, "sent": 0 } { "id": 670301218565156865, "len": 44, "sent": 0 } { "id": 670301218414206976, "len": 71, "sent": 1 } @@ -58,14 +58,14 @@ { "id": 670301218078629888, "len": 9, "sent": 0 } { "id": 670301217851990017, "len": 111, "sent": 0 } { "id": 670301217793269760, "len": 113, "sent": 0 } -{ "id": 670301217508036608, "len": 47, "sent": 0 } +{ "id": 670301217508036608, "len": 55, "sent": 0 } { "id": 670301217369657344, "len": 137, "sent": 0 } -{ "id": 670301217311088641, "len": 28, "sent": 0 } +{ "id": 670301217311088641, "len": 29, "sent": 0 } { "id": 670301217231347712, "len": 123, "sent": 0 } { "id": 670301216891473920, "len": 44, "sent": 0 } { "id": 670301216874721280, "len": 68, "sent": 0 } { "id": 670301216799232000, "len": 50, "sent": 1 } -{ "id": 670301216669171713, "len": 54, "sent": 0 } +{ "id": 670301216669171713, "len": 55, "sent": 0 } { "id": 670301216493060097, "len": 113, "sent": 1 } { "id": 670301216400924676, "len": 35, "sent": 1 } { "id": 670301216371552258, "len": 58, "sent": 0 } @@ -78,21 +78,21 @@ { "id": 670301214958055424, "len": 58, "sent": 1 } { "id": 670301214605733888, "len": 139, "sent": 1 } { "id": 670301214509129728, "len": 114, "sent": 1 } -{ "id": 670301214442041344, "len": 18, "sent": 1 } +{ "id": 670301214442041344, "len": 19, "sent": 1 } { "id": 670301214295392256, "len": 47, "sent": 0 } -{ "id": 670301213737529344, "len": 9, "sent": 0 } +{ "id": 670301213737529344, "len": 10, "sent": 0 } { "id": 670301213544595457, "len": 63, "sent": 1 } { "id": 670301213515235333, "len": 107, "sent": 0 } { "id": 670301213464899584, "len": 105, "sent": 1 } { "id": 670301213120942080, "len": 39, "sent": 0 } { "id": 670301212961603585, "len": 63, "sent": 0 } -{ "id": 670301212961603584, "len": 20, "sent": 0 } -{ "id": 670301212856737792, "len": 51, "sent": 0 } +{ "id": 670301212961603584, "len": 25, "sent": 0 } +{ "id": 670301212856737792, "len": 55, "sent": 0 } { "id": 670301212760117248, "len": 133, "sent": 1 } { "id": 670301211808010240, "len": 103, "sent": 0 } -{ "id": 670301211774468096, "len": 40, "sent": 0 } +{ "id": 670301211774468096, "len": 41, "sent": 0 } { "id": 670301211703144450, "len": 138, "sent": 1 } -{ "id": 670301211581685761, "len": 25, "sent": 1 } +{ "id": 670301211581685761, "len": 26, "sent": 1 } { "id": 670301211560685568, "len": 12, "sent": 1 } { "id": 670301211090751490, "len": 140, "sent": 0 } { "id": 670301210654699520, "len": 13, "sent": 0 } diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_it_python.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_it_python.xml index 5fc7316..686ede2 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_it_python.xml +++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_it_python.xml @@ -85,5 +85,10 @@ ArithmeticError: oof <output-dir compare="Text">big_object_pyudf</output-dir> </compilation-unit> </test-case> + <test-case FilePath="external-library"> + <compilation-unit name="big_object_roundtrip_20M"> + <output-dir compare="Text">big_object_roundtrip_20M</output-dir> + </compilation-unit> + </test-case> </test-group> </test-suite> diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java index d49dffc..84e5c22 100644 --- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java +++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java @@ -85,6 +85,7 @@ public enum ErrorCode implements IError { EXTERNAL_UDF_RESULT_TYPE_ERROR(200), EXTERNAL_UDF_EXCEPTION(201), + EXTERNAL_UDF_PROTO_RETURN_EXCEPTION(202), // Compilation errors PARSE_ERROR(1001), diff --git a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties index 136e169..be4f512 100644 --- a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties +++ b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties @@ -92,6 +92,7 @@ 200 = External UDF cannot produce expected result. Please check the UDF configuration 201 = External UDF returned exception. Returned exception was: %1$s +202 = External UDF protocol encountered an unexpected return result. # Compile-time check errors 1001 = Syntax error: %1$s diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/py_nested_access/py_nested_access.4.query.sqlpp b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/builders/ListLikeNumericArrayFactory.java similarity index 61% copy from asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/py_nested_access/py_nested_access.4.query.sqlpp copy to asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/builders/ListLikeNumericArrayFactory.java index e17f03f..e53f157 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/py_nested_access/py_nested_access.4.query.sqlpp +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/builders/ListLikeNumericArrayFactory.java @@ -16,16 +16,21 @@ * specific language governing permissions and limitations * under the License. */ -/* -* Description : Access a records nested records at each level. -* Expected Res : Success -* Date : 04 Jun 2015 -*/ +package org.apache.asterix.external.input.stream.builders; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import org.apache.asterix.om.util.container.IObjectFactory; -use test; +public class ListLikeNumericArrayFactory<T extends Number> implements IObjectFactory<List<T>, T> { + @Override + public List<T> create(T arg) { + List<T> list = new ArrayList<>(arg.intValue()); + list.addAll(Collections.nCopies(arg.intValue(), arg)); + return list; + } -select element result -from Animals as test -with result as roundtrip(test)[0].class.fullClassification.lower.lower.lower.lower.lower.lower.Species -order by result; +} diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/builders/StandardToModifiedUTF8DataOutputFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/builders/StandardToModifiedUTF8DataOutputFactory.java new file mode 100644 index 0000000..606aee2 --- /dev/null +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/builders/StandardToModifiedUTF8DataOutputFactory.java @@ -0,0 +1,37 @@ +/* + * 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.asterix.external.input.stream.builders; + +import org.apache.asterix.dataflow.data.nontagged.serde.AStringSerializerDeserializer; +import org.apache.asterix.external.input.stream.StandardUTF8ToModifiedUTF8DataOutput; +import org.apache.asterix.om.types.ATypeTag; +import org.apache.asterix.om.util.container.IObjectFactory; +import org.apache.hyracks.util.string.UTF8StringReader; +import org.apache.hyracks.util.string.UTF8StringWriter; + +public class StandardToModifiedUTF8DataOutputFactory + implements IObjectFactory<StandardUTF8ToModifiedUTF8DataOutput, ATypeTag> { + + @Override + public StandardUTF8ToModifiedUTF8DataOutput create(ATypeTag type) { + return new StandardUTF8ToModifiedUTF8DataOutput( + new AStringSerializerDeserializer(new UTF8StringWriter(), new UTF8StringReader())); + } + +} diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/PythonIPCProto.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/PythonIPCProto.java index cd7ec18..c803517 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/PythonIPCProto.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/ipc/PythonIPCProto.java @@ -18,14 +18,28 @@ package org.apache.asterix.external.ipc; import static org.apache.hyracks.ipc.impl.Message.HEADER_SIZE; +import java.io.DataOutput; +import java.io.DataOutputStream; import java.io.IOException; import java.io.OutputStream; import java.nio.ByteBuffer; import org.apache.asterix.common.exceptions.AsterixException; import org.apache.asterix.common.exceptions.ErrorCode; +import org.apache.asterix.external.library.msgpack.MsgPackPointableVisitor; +import org.apache.asterix.om.pointables.AFlatValuePointable; +import org.apache.asterix.om.pointables.AListVisitablePointable; +import org.apache.asterix.om.pointables.ARecordVisitablePointable; +import org.apache.asterix.om.pointables.PointableAllocator; +import org.apache.asterix.om.pointables.base.IVisitablePointable; +import org.apache.asterix.om.types.ATypeTag; +import org.apache.asterix.om.types.EnumDeserializer; +import org.apache.asterix.om.types.IAType; +import org.apache.asterix.om.types.TypeTagUtil; import org.apache.hyracks.algebricks.common.utils.Pair; import org.apache.hyracks.api.exceptions.HyracksDataException; +import org.apache.hyracks.data.std.api.IValueReference; +import org.apache.hyracks.data.std.util.ArrayBackedValueStorage; import org.apache.hyracks.ipc.impl.Message; import org.msgpack.core.MessagePack; import org.msgpack.core.MessageUnpacker; @@ -33,26 +47,32 @@ import org.msgpack.core.buffer.ArrayBufferInput; public class PythonIPCProto { - private PythonMessageBuilder messageBuilder; - private OutputStream sockOut; - private ByteBuffer headerBuffer = ByteBuffer.allocate(21); + private final PythonMessageBuilder messageBuilder; + private final DataOutputStream sockOut; + private final ByteBuffer headerBuffer = ByteBuffer.allocate(21); private ByteBuffer recvBuffer = ByteBuffer.allocate(32768); - private ExternalFunctionResultRouter router; + private final ExternalFunctionResultRouter router; private long routeId; private Pair<ByteBuffer, Exception> bufferBox; - private Process pythonProc; + private final Process pythonProc; private long maxFunctionId; - private ArrayBufferInput unpackerInput; - private MessageUnpacker unpacker; + private final ArrayBufferInput unpackerInput; + private final MessageUnpacker unpacker; + private final ArrayBackedValueStorage argsStorage; + private final PointableAllocator pointableAllocator; + private final MsgPackPointableVisitor pointableVisitor; public PythonIPCProto(OutputStream sockOut, ExternalFunctionResultRouter router, Process pythonProc) { - this.sockOut = sockOut; + this.sockOut = new DataOutputStream(sockOut); messageBuilder = new PythonMessageBuilder(); this.router = router; this.pythonProc = pythonProc; - this.maxFunctionId = 0l; + this.maxFunctionId = 0L; unpackerInput = new ArrayBufferInput(new byte[0]); unpacker = MessagePack.newDefaultUnpacker(unpackerInput); + this.argsStorage = new ArrayBackedValueStorage(); + this.pointableAllocator = new PointableAllocator(); + this.pointableVisitor = new MsgPackPointableVisitor(); } public void start() { @@ -65,10 +85,10 @@ public class PythonIPCProto { recvBuffer.clear(); recvBuffer.position(0); recvBuffer.limit(0); - messageBuilder.buf.clear(); - messageBuilder.buf.position(0); + messageBuilder.reset(); messageBuilder.hello(); - sendMsg(routeId); + sendHeader(routeId, messageBuilder.getLength()); + sendMsg(); receiveMsg(); if (getResponseType() != MessageType.HELO) { throw HyracksDataException.create(org.apache.hyracks.api.exceptions.ErrorCode.ILLEGAL_STATE, @@ -81,10 +101,10 @@ public class PythonIPCProto { recvBuffer.clear(); recvBuffer.position(0); recvBuffer.limit(0); - messageBuilder.buf.clear(); - messageBuilder.buf.position(0); + messageBuilder.reset(); messageBuilder.init(module, clazz, fn); - sendMsg(functionId); + sendHeader(functionId, messageBuilder.getLength()); + sendMsg(); receiveMsg(); if (getResponseType() != MessageType.INIT_RSP) { throw HyracksDataException.create(org.apache.hyracks.api.exceptions.ErrorCode.ILLEGAL_STATE, @@ -93,14 +113,21 @@ public class PythonIPCProto { return functionId; } - public ByteBuffer call(long functionId, ByteBuffer args, int numArgs) throws IOException, AsterixException { + public ByteBuffer call(long functionId, IAType[] argTypes, IValueReference[] argValues, boolean nullCall) + throws IOException, AsterixException { recvBuffer.clear(); recvBuffer.position(0); recvBuffer.limit(0); - messageBuilder.buf.clear(); - messageBuilder.buf.position(0); - messageBuilder.call(args.array(), args.position(), numArgs); - sendMsg(functionId); + messageBuilder.reset(); + argsStorage.reset(); + for (int i = 0; i < argTypes.length; i++) { + visitValueRef(argTypes[i], argsStorage.getDataOutput(), argValues[i], pointableAllocator, pointableVisitor, + nullCall); + } + int len = argsStorage.getLength() + 5; + sendHeader(functionId, len); + messageBuilder.call(argValues.length, len); + sendMsg(argsStorage); receiveMsg(); if (getResponseType() != MessageType.CALL_RSP) { throw HyracksDataException.create(org.apache.hyracks.api.exceptions.ErrorCode.ILLEGAL_STATE, @@ -109,14 +136,16 @@ public class PythonIPCProto { return recvBuffer; } - public ByteBuffer callMulti(long key, ByteBuffer args, int numTuples) throws IOException, AsterixException { + public ByteBuffer callMulti(long key, ArrayBackedValueStorage args, int numTuples) + throws IOException, AsterixException { recvBuffer.clear(); recvBuffer.position(0); recvBuffer.limit(0); - messageBuilder.buf.clear(); - messageBuilder.buf.position(0); - messageBuilder.callMulti(args.array(), args.position(), numTuples); - sendMsg(key); + messageBuilder.reset(); + int len = args.getLength() + 4; + sendHeader(key, len); + messageBuilder.callMulti(0, numTuples); + sendMsg(args); receiveMsg(); if (getResponseType() != MessageType.CALL_RSP) { throw HyracksDataException.create(org.apache.hyracks.api.exceptions.ErrorCode.ILLEGAL_STATE, @@ -132,7 +161,7 @@ public class PythonIPCProto { } public void receiveMsg() throws IOException, AsterixException { - Exception except = null; + Exception except; try { synchronized (bufferBox) { while ((bufferBox.getFirst().limit() == 0 || bufferBox.getSecond() != null) && pythonProc.isAlive()) { @@ -162,15 +191,27 @@ public class PythonIPCProto { } } - public void sendMsg(long key) throws IOException { + public void sendHeader(long key, int msgLen) throws IOException { headerBuffer.clear(); headerBuffer.position(0); - headerBuffer.putInt(HEADER_SIZE + Integer.BYTES + messageBuilder.buf.position()); + headerBuffer.putInt(HEADER_SIZE + Integer.BYTES + msgLen); headerBuffer.putLong(key); headerBuffer.putLong(routeId); headerBuffer.put(Message.NORMAL); sockOut.write(headerBuffer.array(), 0, HEADER_SIZE + Integer.BYTES); - sockOut.write(messageBuilder.buf.array(), 0, messageBuilder.buf.position()); + sockOut.flush(); + } + + public void sendMsg(ArrayBackedValueStorage content) throws IOException { + sockOut.write(messageBuilder.getBuf().array(), messageBuilder.getBuf().arrayOffset(), + messageBuilder.getBuf().position()); + sockOut.write(content.getByteArray(), content.getStartOffset(), content.getLength()); + sockOut.flush(); + } + + public void sendMsg() throws IOException { + sockOut.write(messageBuilder.getBuf().array(), messageBuilder.getBuf().arrayOffset(), + messageBuilder.getBuf().position()); sockOut.flush(); } @@ -182,4 +223,68 @@ public class PythonIPCProto { return routeId; } + public DataOutputStream getSockOut() { + return sockOut; + } + + public static void visitValueRef(IAType type, DataOutput out, IValueReference valueReference, + PointableAllocator pointableAllocator, MsgPackPointableVisitor pointableVisitor, boolean visitNull) + throws IOException { + IVisitablePointable pointable; + switch (type.getTypeTag()) { + case OBJECT: + pointable = pointableAllocator.allocateRecordValue(type); + pointable.set(valueReference); + pointableVisitor.visit((ARecordVisitablePointable) pointable, pointableVisitor.getTypeInfo(type, out)); + break; + case ARRAY: + case MULTISET: + pointable = pointableAllocator.allocateListValue(type); + pointable.set(valueReference); + pointableVisitor.visit((AListVisitablePointable) pointable, pointableVisitor.getTypeInfo(type, out)); + break; + case ANY: + ATypeTag rtTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER + .deserialize(valueReference.getByteArray()[valueReference.getStartOffset()]); + IAType rtType = TypeTagUtil.getBuiltinTypeByTag(rtTypeTag); + switch (rtTypeTag) { + case OBJECT: + pointable = pointableAllocator.allocateRecordValue(rtType); + pointable.set(valueReference); + pointableVisitor.visit((ARecordVisitablePointable) pointable, + pointableVisitor.getTypeInfo(rtType, out)); + break; + case ARRAY: + case MULTISET: + pointable = pointableAllocator.allocateListValue(rtType); + pointable.set(valueReference); + pointableVisitor.visit((AListVisitablePointable) pointable, + pointableVisitor.getTypeInfo(rtType, out)); + break; + case MISSING: + case NULL: + if (!visitNull) { + return; + } + default: + pointable = pointableAllocator.allocateFieldValue(rtType); + pointable.set(valueReference); + pointableVisitor.visit((AFlatValuePointable) pointable, + pointableVisitor.getTypeInfo(rtType, out)); + break; + } + break; + case MISSING: + case NULL: + if (!visitNull) { + return; + } + default: + pointable = pointableAllocator.allocateFieldValue(type); + pointable.set(valueReference); + pointableVisitor.visit((AFlatValuePointable) pointable, pointableVisitor.getTypeInfo(type, out)); + break; + } + } + } 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 5052eb4..5429657 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 @@ -26,15 +26,13 @@ import java.net.InetAddress; import java.net.InetSocketAddress; import java.nio.ByteBuffer; -import org.apache.asterix.external.library.msgpack.MessagePackerFromADM; -import org.apache.hyracks.api.exceptions.ErrorCode; +import org.apache.asterix.external.library.msgpack.MessagePackUtils; import org.apache.hyracks.api.exceptions.HyracksDataException; public class PythonMessageBuilder { - private static final int MAX_BUF_SIZE = 64 * 1024 * 1024; //64MB. MessageType type; long dataLength; - ByteBuffer buf; + private final ByteBuffer buf; public PythonMessageBuilder() { this.type = null; @@ -42,12 +40,25 @@ public class PythonMessageBuilder { this.buf = ByteBuffer.allocate(4096); } + public void reset() { + //TODO: should be able to get away w/o clearing buf? + buf.clear(); + } + + public ByteBuffer getBuf() { + return buf; + } + + public int getLength() { + return buf.position() - buf.arrayOffset(); + } + public void setType(MessageType type) { this.type = type; } public void packHeader() throws HyracksDataException { - MessagePackerFromADM.packFixPos(buf, (byte) type.ordinal()); + MessagePackUtils.packFixPos(buf, (byte) type.ordinal()); } //TODO: this is wrong for any multibyte chars @@ -75,7 +86,7 @@ public class PythonMessageBuilder { this.type = MessageType.QUIT; dataLength = getStringLength("QUIT"); packHeader(); - MessagePackerFromADM.packFixStr(buf, "QUIT"); + MessagePackUtils.packFixStr(buf, "QUIT"); } public void init(final String module, final String clazz, final String fn) throws HyracksDataException { @@ -89,46 +100,27 @@ public class PythonMessageBuilder { } packHeader(); int numArgs = clazz == null ? 2 : 3; - MessagePackerFromADM.packFixArrayHeader(buf, (byte) numArgs); - MessagePackerFromADM.packStr(buf, module); + MessagePackUtils.packFixArrayHeader(buf, (byte) numArgs); + MessagePackUtils.packStr(buf, module); if (clazz != null) { - MessagePackerFromADM.packStr(buf, clazz); + MessagePackUtils.packStr(buf, clazz); } - MessagePackerFromADM.packStr(buf, fn); + MessagePackUtils.packStr(buf, fn); } - public void call(byte[] args, int lim, int numArgs) throws HyracksDataException { - if (args.length > buf.capacity()) { - int growTo = ExternalFunctionResultRouter.closestPow2(args.length); - if (growTo > MAX_BUF_SIZE) { - throw HyracksDataException.create(ErrorCode.ILLEGAL_STATE, - "Unable to allocate message buffer larger than:" + MAX_BUF_SIZE + " bytes"); - } - buf = ByteBuffer.allocate(growTo); - } + public void call(int numArgs, int len) throws HyracksDataException { buf.clear(); buf.position(0); this.type = MessageType.CALL; - dataLength = 5 + 1 + lim; + dataLength = 5 + 1 + len; packHeader(); //TODO: make this switch between fixarray/array16/array32 buf.put((byte) (FIXARRAY_PREFIX + 1)); - buf.put(ARRAY32); - buf.putInt(numArgs); - if (numArgs > 0) { - buf.put(args, 0, lim); - } + buf.put(ARRAY16); + buf.putShort((short) numArgs); } - public void callMulti(byte[] args, int lim, int numArgs) throws HyracksDataException { - if (args.length > buf.capacity()) { - int growTo = ExternalFunctionResultRouter.closestPow2(args.length); - if (growTo > MAX_BUF_SIZE) { - throw HyracksDataException.create(ErrorCode.ILLEGAL_STATE, - "Unable to allocate message buffer larger than:" + MAX_BUF_SIZE + " bytes"); - } - buf = ByteBuffer.allocate(growTo); - } + public void callMulti(int lim, int numArgs) throws HyracksDataException { buf.clear(); buf.position(0); this.type = MessageType.CALL; @@ -137,9 +129,6 @@ public class PythonMessageBuilder { //TODO: make this switch between fixarray/array16/array32 buf.put(ARRAY16); buf.putShort((short) numArgs); - if (numArgs > 0) { - buf.put(args, 0, lim); - } } //this is used to send a serialized java inetaddress to the entrypoint so it can send it back diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalScalarPythonFunctionEvaluator.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalScalarPythonFunctionEvaluator.java index 7c860a2..94a4dd2 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalScalarPythonFunctionEvaluator.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalScalarPythonFunctionEvaluator.java @@ -61,6 +61,7 @@ class ExternalScalarPythonFunctionEvaluator extends ExternalScalarFunctionEvalua private MessageUnpacker unpacker; private ArrayBufferInput unpackerInput; + private MessageUnpackerToADM unpackerToADM; private long fnId; @@ -87,6 +88,7 @@ class ExternalScalarPythonFunctionEvaluator extends ExternalScalarFunctionEvalua this.sourceLocation = sourceLoc; this.unpackerInput = new ArrayBufferInput(new byte[0]); this.unpacker = MessagePack.newDefaultUnpacker(unpackerInput); + this.unpackerToADM = new MessageUnpackerToADM(); } @Override @@ -107,18 +109,13 @@ class ExternalScalarPythonFunctionEvaluator extends ExternalScalarFunctionEvalua hasNullArg = true; } } - try { - PythonLibraryEvaluator.setArgument(argTypes[i], argValues[i], argHolder, nullCall); - } catch (IOException e) { - throw new HyracksDataException("Error evaluating Python UDF", e); - } } if (!nullCall && hasNullArg) { PointableHelper.setNull(result); return; } try { - ByteBuffer res = libraryEvaluator.callPython(fnId, argHolder, argTypes.length); + ByteBuffer res = libraryEvaluator.callPython(fnId, argTypes, argValues, nullCall); resultBuffer.reset(); wrap(res, resultBuffer.getDataOutput()); } catch (Exception e) { @@ -133,30 +130,28 @@ class ExternalScalarPythonFunctionEvaluator extends ExternalScalarFunctionEvalua outputWrapper.position(0); try { if (resultWrapper == null) { - outputWrapper.put(ATypeTag.SERIALIZED_NULL_TYPE_TAG); - out.write(outputWrapper.array(), 0, outputWrapper.position() + outputWrapper.arrayOffset()); + out.writeByte(ATypeTag.SERIALIZED_NULL_TYPE_TAG); return; } if ((resultWrapper.get() ^ FIXARRAY_PREFIX) != (byte) 2) { - throw HyracksDataException.create(AsterixException.create(ErrorCode.EXTERNAL_UDF_EXCEPTION, - "Returned result missing outer wrapper")); + throw HyracksDataException + .create(AsterixException.create(ErrorCode.EXTERNAL_UDF_PROTO_RETURN_EXCEPTION)); } int numresults = resultWrapper.get() ^ FIXARRAY_PREFIX; if (numresults > 0) { - MessageUnpackerToADM.unpack(resultWrapper, outputWrapper, true); + unpackerToADM.unpack(resultWrapper, out, true); } unpackerInput.reset(resultWrapper.array(), resultWrapper.position() + resultWrapper.arrayOffset(), resultWrapper.remaining()); unpacker.reset(unpackerInput); - int numEntries = unpacker.unpackArrayHeader(); - for (int j = 0; j < numEntries; j++) { - outputWrapper.put(ATypeTag.SERIALIZED_NULL_TYPE_TAG); + int numErrors = unpacker.unpackArrayHeader(); + for (int j = 0; j < numErrors; j++) { + out.writeByte(ATypeTag.SERIALIZED_NULL_TYPE_TAG); if (evaluatorContext.getWarningCollector().shouldWarn()) { evaluatorContext.getWarningCollector().warn( Warning.of(sourceLocation, ErrorCode.EXTERNAL_UDF_EXCEPTION, unpacker.unpackString())); } } - out.write(outputWrapper.array(), 0, outputWrapper.position() + outputWrapper.arrayOffset()); } catch (IOException e) { throw HyracksDataException.create(e); } diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/JavaFunctionHelper.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/JavaFunctionHelper.java index d1b2685..98755fc 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/JavaFunctionHelper.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/JavaFunctionHelper.java @@ -56,7 +56,7 @@ public class JavaFunctionHelper implements IFunctionHelper { private final IObjectPool<IJObject, IAType> objectPool = new ListObjectPool<>(JTypeObjectFactory.INSTANCE); private final JObjectPointableVisitor pointableVisitor; private final PointableAllocator pointableAllocator; - private final Map<Integer, TypeInfo> poolTypeInfo; + private final Map<Integer, JavaTypeInfo> poolTypeInfo; private final Map<String, String> parameters; private final IAType[] argTypes; @@ -164,10 +164,10 @@ public class JavaFunctionHelper implements IFunctionHelper { arguments[index] = jObject; } - private TypeInfo getTypeInfo(int index, IAType type) { - TypeInfo typeInfo = poolTypeInfo.get(index); + private JavaTypeInfo getTypeInfo(int index, IAType type) { + JavaTypeInfo typeInfo = poolTypeInfo.get(index); if (typeInfo == null) { - typeInfo = new TypeInfo(objectPool, type, type.getTypeTag()); + typeInfo = new JavaTypeInfo(objectPool, type, type.getTypeTag()); poolTypeInfo.put(index, typeInfo); } return typeInfo; diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/TypeInfo.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/JavaTypeInfo.java similarity index 93% rename from asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/TypeInfo.java rename to asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/JavaTypeInfo.java index 453cf39..e60cc5d 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/TypeInfo.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/JavaTypeInfo.java @@ -23,13 +23,13 @@ import org.apache.asterix.om.types.ATypeTag; import org.apache.asterix.om.types.IAType; import org.apache.asterix.om.util.container.IObjectPool; -public class TypeInfo { +public class JavaTypeInfo { private IObjectPool<IJObject, IAType> objectPool; private IAType atype; private ATypeTag typeTag; - public TypeInfo(IObjectPool<IJObject, IAType> objectPool, IAType atype, ATypeTag typeTag) { + public JavaTypeInfo(IObjectPool<IJObject, IAType> objectPool, IAType atype, ATypeTag typeTag) { this.objectPool = objectPool; this.atype = atype; this.typeTag = typeTag; diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/py_nested_access/py_nested_access.4.query.sqlpp b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/PyTypeInfo.java similarity index 54% copy from asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/py_nested_access/py_nested_access.4.query.sqlpp copy to asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/PyTypeInfo.java index e17f03f..e2b7ad4 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/py_nested_access/py_nested_access.4.query.sqlpp +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/PyTypeInfo.java @@ -16,16 +16,39 @@ * specific language governing permissions and limitations * under the License. */ -/* -* Description : Access a records nested records at each level. -* Expected Res : Success -* Date : 04 Jun 2015 -*/ +package org.apache.asterix.external.library; + +import java.io.DataOutput; +import java.util.Objects; + +import org.apache.asterix.om.types.IAType; + +public class PyTypeInfo { + + private final IAType type; + private final DataOutput out; + + public PyTypeInfo(IAType type, DataOutput out) { + this.type = type; + this.out = out; + } + + public DataOutput getDataOutput() { + return out; + } + + public IAType getType() { + return type; + } -use test; + @Override + public int hashCode() { + return Objects.hash(out, type); + } + @Override + public boolean equals(Object obj) { + return out.equals(out) && type.equals(type); + } -select element result -from Animals as test -with result as roundtrip(test)[0].class.fullClassification.lower.lower.lower.lower.lower.lower.Species -order by result; +} diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/PythonLibraryEvaluator.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/PythonLibraryEvaluator.java index 457b86a..f82b30d 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/PythonLibraryEvaluator.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/PythonLibraryEvaluator.java @@ -34,7 +34,7 @@ import org.apache.asterix.common.exceptions.AsterixException; import org.apache.asterix.common.library.ILibraryManager; import org.apache.asterix.external.ipc.ExternalFunctionResultRouter; import org.apache.asterix.external.ipc.PythonIPCProto; -import org.apache.asterix.external.library.msgpack.MessagePackerFromADM; +import org.apache.asterix.external.library.msgpack.MessagePackUtils; import org.apache.asterix.om.functions.IExternalFunctionInfo; import org.apache.asterix.om.types.ATypeTag; import org.apache.asterix.om.types.EnumDeserializer; @@ -50,6 +50,7 @@ import org.apache.hyracks.api.job.JobId; import org.apache.hyracks.api.resources.IDeallocatable; import org.apache.hyracks.data.std.api.IValueReference; import org.apache.hyracks.data.std.primitive.TaggedValuePointable; +import org.apache.hyracks.data.std.util.ArrayBackedValueStorage; import org.apache.hyracks.dataflow.std.base.AbstractStateObject; import org.apache.hyracks.ipc.impl.IPCSystem; @@ -86,7 +87,6 @@ public class PythonLibraryEvaluator extends AbstractStateObject implements IDeal this.ipcSys = ipcSys; this.warningCollector = warningCollector; this.sourceLoc = sourceLoc; - } private void initialize() throws IOException, AsterixException { @@ -128,10 +128,11 @@ public class PythonLibraryEvaluator extends AbstractStateObject implements IDeal return proto.init(packageModule, clazz, fn); } - public ByteBuffer callPython(long id, ByteBuffer arguments, int numArgs) throws IOException { + public ByteBuffer callPython(long id, IAType[] argTypes, IValueReference[] valueReferences, boolean nullCall) + throws IOException { ByteBuffer ret = null; try { - ret = proto.call(id, arguments, numArgs); + ret = proto.call(id, argTypes, valueReferences, nullCall); } catch (AsterixException e) { if (warningCollector.shouldWarn()) { warningCollector.warn(Warning.of(sourceLoc, EXTERNAL_UDF_EXCEPTION, e.getMessage())); @@ -140,7 +141,7 @@ public class PythonLibraryEvaluator extends AbstractStateObject implements IDeal return ret; } - public ByteBuffer callPythonMulti(long id, ByteBuffer arguments, int numTuples) throws IOException { + public ByteBuffer callPythonMulti(long id, ArrayBackedValueStorage arguments, int numTuples) throws IOException { ByteBuffer ret = null; try { ret = proto.callMulti(id, arguments, numTuples); @@ -169,20 +170,6 @@ public class PythonLibraryEvaluator extends AbstractStateObject implements IDeal router.removeRoute(proto.getRouteId()); } - public static ATypeTag setArgument(IAType type, IValueReference valueReference, ByteBuffer argHolder, - boolean nullCall) throws IOException { - ATypeTag tag = type.getTypeTag(); - if (tag == ATypeTag.ANY) { - TaggedValuePointable pointy = TaggedValuePointable.FACTORY.createPointable(); - pointy.set(valueReference); - ATypeTag rtTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(pointy.getTag()); - IAType rtType = TypeTagUtil.getBuiltinTypeByTag(rtTypeTag); - return MessagePackerFromADM.pack(valueReference, rtType, argHolder, nullCall); - } else { - return MessagePackerFromADM.pack(valueReference, type, argHolder, nullCall); - } - } - public static ATypeTag peekArgument(IAType type, IValueReference valueReference) throws HyracksDataException { ATypeTag tag = type.getTypeTag(); if (tag == ATypeTag.ANY) { @@ -190,15 +177,15 @@ public class PythonLibraryEvaluator extends AbstractStateObject implements IDeal pointy.set(valueReference); ATypeTag rtTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(pointy.getTag()); IAType rtType = TypeTagUtil.getBuiltinTypeByTag(rtTypeTag); - return MessagePackerFromADM.peekUnknown(rtType); + return MessagePackUtils.peekUnknown(rtType); } else { - return MessagePackerFromADM.peekUnknown(type); + return MessagePackUtils.peekUnknown(type); } } - public static void setVoidArgument(ByteBuffer argHolder) { - argHolder.put(ARRAY16); - argHolder.putShort((short) 0); + public static void setVoidArgument(ArrayBackedValueStorage argHolder) throws IOException { + argHolder.getDataOutput().writeByte(ARRAY16); + argHolder.getDataOutput().writeShort((short) 0); } public static PythonLibraryEvaluator getInstance(IExternalFunctionInfo finfo, ILibraryManager libMgr, @@ -218,4 +205,5 @@ public class PythonLibraryEvaluator extends AbstractStateObject implements IDeal } return evaluator; } + } diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/java/JObjectAccessors.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/java/JObjectAccessors.java index 230627f..5479b56 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/java/JObjectAccessors.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/java/JObjectAccessors.java @@ -42,7 +42,7 @@ import org.apache.asterix.external.api.IJListAccessor; import org.apache.asterix.external.api.IJObject; import org.apache.asterix.external.api.IJObjectAccessor; import org.apache.asterix.external.api.IJRecordAccessor; -import org.apache.asterix.external.library.TypeInfo; +import org.apache.asterix.external.library.JavaTypeInfo; import org.apache.asterix.external.library.java.base.JBoolean; import org.apache.asterix.external.library.java.base.JByte; import org.apache.asterix.external.library.java.base.JDate; @@ -356,14 +356,14 @@ public class JObjectAccessors { public static class JRecordAccessor implements IJRecordAccessor { - private final TypeInfo typeInfo; + private final JavaTypeInfo typeInfo; private final JRecord jRecord; private final IJObject[] jObjects; private final LinkedHashMap<String, IJObject> openFields; private final UTF8StringReader reader = new UTF8StringReader(); public JRecordAccessor(ARecordType recordType, IObjectPool<IJObject, IAType> objectPool) { - this.typeInfo = new TypeInfo(objectPool, null, null); + this.typeInfo = new JavaTypeInfo(objectPool, null, null); this.jObjects = new IJObject[recordType.getFieldNames().length]; this.openFields = new LinkedHashMap<>(); this.jRecord = new JRecord(recordType, jObjects, openFields); @@ -439,10 +439,10 @@ public class JObjectAccessors { public static class JListAccessor implements IJListAccessor { - private final TypeInfo typeInfo; + private final JavaTypeInfo typeInfo; public JListAccessor(IObjectPool<IJObject, IAType> objectPool) { - this.typeInfo = new TypeInfo(objectPool, null, null); + this.typeInfo = new JavaTypeInfo(objectPool, null, null); } @Override diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/java/JObjectPointableVisitor.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/java/JObjectPointableVisitor.java index 9edc569..bd64922 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/java/JObjectPointableVisitor.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/java/JObjectPointableVisitor.java @@ -25,7 +25,7 @@ import org.apache.asterix.external.api.IJListAccessor; import org.apache.asterix.external.api.IJObject; import org.apache.asterix.external.api.IJObjectAccessor; import org.apache.asterix.external.api.IJRecordAccessor; -import org.apache.asterix.external.library.TypeInfo; +import org.apache.asterix.external.library.JavaTypeInfo; import org.apache.asterix.external.library.java.JObjectAccessors.JListAccessor; import org.apache.asterix.external.library.java.JObjectAccessors.JRecordAccessor; import org.apache.asterix.om.pointables.AFlatValuePointable; @@ -37,7 +37,7 @@ import org.apache.asterix.om.types.ARecordType; import org.apache.asterix.om.types.ATypeTag; import org.apache.hyracks.api.exceptions.HyracksDataException; -public class JObjectPointableVisitor implements IVisitablePointableVisitor<IJObject, TypeInfo> { +public class JObjectPointableVisitor implements IVisitablePointableVisitor<IJObject, JavaTypeInfo> { private final Map<ATypeTag, IJObjectAccessor> flatJObjectAccessors = new HashMap<ATypeTag, IJObjectAccessor>(); private final Map<IVisitablePointable, IJRecordAccessor> raccessorToJObject = @@ -46,7 +46,7 @@ public class JObjectPointableVisitor implements IVisitablePointableVisitor<IJObj new HashMap<IVisitablePointable, IJListAccessor>(); @Override - public IJObject visit(AListVisitablePointable accessor, TypeInfo arg) throws HyracksDataException { + public IJObject visit(AListVisitablePointable accessor, JavaTypeInfo arg) throws HyracksDataException { IJObject result = null; IJListAccessor jListAccessor = laccessorToPrinter.get(accessor); if (jListAccessor == null) { @@ -58,7 +58,7 @@ public class JObjectPointableVisitor implements IVisitablePointableVisitor<IJObj } @Override - public IJObject visit(ARecordVisitablePointable accessor, TypeInfo arg) throws HyracksDataException { + public IJObject visit(ARecordVisitablePointable accessor, JavaTypeInfo arg) throws HyracksDataException { IJObject result = null; IJRecordAccessor jRecordAccessor = raccessorToJObject.get(accessor); if (jRecordAccessor == null) { @@ -70,7 +70,7 @@ public class JObjectPointableVisitor implements IVisitablePointableVisitor<IJObj } @Override - public IJObject visit(AFlatValuePointable accessor, TypeInfo arg) throws HyracksDataException { + public IJObject visit(AFlatValuePointable accessor, JavaTypeInfo arg) throws HyracksDataException { ATypeTag typeTag = arg.getTypeTag(); IJObject result = null; IJObjectAccessor jObjectAccessor = flatJObjectAccessors.get(typeTag); @@ -83,7 +83,8 @@ public class JObjectPointableVisitor implements IVisitablePointableVisitor<IJObj return result; } - public IJObject visit(AFlatValuePointable accessor, ATypeTag typeTag, TypeInfo arg) throws HyracksDataException { + public IJObject visit(AFlatValuePointable accessor, ATypeTag typeTag, JavaTypeInfo arg) + throws HyracksDataException { IJObject result = null; IJObjectAccessor jObjectAccessor = flatJObjectAccessors.get(typeTag); if (jObjectAccessor == null) { diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/IMsgPackAccessor.java similarity index 75% copy from asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp copy to asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/IMsgPackAccessor.java index 8041b5e..6b2f0f8 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/IMsgPackAccessor.java @@ -16,10 +16,11 @@ * specific language governing permissions and limitations * under the License. */ -use externallibtest; +package org.apache.asterix.external.library.msgpack; -create function typeValidation(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib; +import java.io.IOException; -create function typeValidationNullCall(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib with {"null-call": true}; +@FunctionalInterface +public interface IMsgPackAccessor<T, U, R> { + R apply(T t, U u) throws IOException; +} diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/IMsgPackListAccessor.java similarity index 75% copy from asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp copy to asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/IMsgPackListAccessor.java index 8041b5e..a061a06 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/IMsgPackListAccessor.java @@ -16,10 +16,13 @@ * specific language governing permissions and limitations * under the License. */ -use externallibtest; +package org.apache.asterix.external.library.msgpack; -create function typeValidation(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib; +import org.apache.hyracks.api.exceptions.HyracksDataException; -create function typeValidationNullCall(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib with {"null-call": true}; +@FunctionalInterface +public interface IMsgPackListAccessor<P, D, T, V, R> { + + R access(P p, D d, T t, V v) throws HyracksDataException; + +} diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/IMsgPackRecordAccessor.java similarity index 75% copy from asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp copy to asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/IMsgPackRecordAccessor.java index 8041b5e..0ce417d 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-library/python_open_type_validation/type_validation.2.ddl.sqlpp +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/IMsgPackRecordAccessor.java @@ -16,10 +16,13 @@ * specific language governing permissions and limitations * under the License. */ -use externallibtest; +package org.apache.asterix.external.library.msgpack; -create function typeValidation(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib; +import org.apache.hyracks.api.exceptions.HyracksDataException; -create function typeValidationNullCall(a, b, c, d, e, f, g, h) - as "roundtrip", "Tests.roundtrip" at testlib with {"null-call": true}; +@FunctionalInterface +public interface IMsgPackRecordAccessor<P, D, T, V, R> { + + R access(P p, D d, T t, V v) throws HyracksDataException; + +} 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 new file mode 100644 index 0000000..2377f9a --- /dev/null +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessagePackUtils.java @@ -0,0 +1,73 @@ +/* + * 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.asterix.external.library.msgpack; + +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 java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; + +import org.apache.asterix.om.types.ATypeTag; +import org.apache.asterix.om.types.IAType; +import org.apache.hyracks.api.exceptions.HyracksDataException; + +public class MessagePackUtils { + + public static ATypeTag peekUnknown(IAType type) { + switch (type.getTypeTag()) { + case MISSING: + case NULL: + return type.getTypeTag(); + default: + return ATypeTag.TYPE; + } + } + + public static void packFixPos(ByteBuffer buf, byte in) throws HyracksDataException { + byte mask = (byte) (1 << 7); + if ((in & mask) != 0) { + throw HyracksDataException.create(org.apache.hyracks.api.exceptions.ErrorCode.ILLEGAL_STATE, + "fixint7 must be positive"); + } + buf.put(in); + } + + public static void packFixStr(ByteBuffer buf, String in) throws HyracksDataException { + byte[] strBytes = in.getBytes(StandardCharsets.UTF_8); + if (strBytes.length > 31) { + throw HyracksDataException.create(org.apache.hyracks.api.exceptions.ErrorCode.ILLEGAL_STATE, + "fixint7 must be positive"); + } + buf.put((byte) (FIXSTR_PREFIX + strBytes.length)); + buf.put(strBytes); + } + + public static void packStr(ByteBuffer out, String in) { + out.put(STR32); + byte[] strBytes = in.getBytes(StandardCharsets.UTF_8); + out.putInt(strBytes.length); + out.put(strBytes); + } + + public static void packFixArrayHeader(ByteBuffer buf, byte numObj) { + buf.put((byte) (FIXARRAY_PREFIX + (0x0F & numObj))); + } +} diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessagePackerFromADM.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessagePackerFromADM.java deleted file mode 100644 index f0ac56e..0000000 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessagePackerFromADM.java +++ /dev/null @@ -1,362 +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.asterix.external.library.msgpack; - -import static org.msgpack.core.MessagePack.Code.ARRAY32; -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.FLOAT32; -import static org.msgpack.core.MessagePack.Code.FLOAT64; -import static org.msgpack.core.MessagePack.Code.INT16; -import static org.msgpack.core.MessagePack.Code.INT32; -import static org.msgpack.core.MessagePack.Code.INT64; -import static org.msgpack.core.MessagePack.Code.INT8; -import static org.msgpack.core.MessagePack.Code.MAP32; -import static org.msgpack.core.MessagePack.Code.NIL; -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; - -import org.apache.asterix.common.exceptions.AsterixException; -import org.apache.asterix.common.exceptions.ErrorCode; -import org.apache.asterix.om.types.ARecordType; -import org.apache.asterix.om.types.ATypeTag; -import org.apache.asterix.om.types.AUnionType; -import org.apache.asterix.om.types.AbstractCollectionType; -import org.apache.asterix.om.types.IAType; -import org.apache.asterix.om.types.TypeTagUtil; -import org.apache.asterix.om.utils.NonTaggedFormatUtil; -import org.apache.asterix.om.utils.RecordUtil; -import org.apache.hyracks.api.exceptions.HyracksDataException; -import org.apache.hyracks.data.std.api.IValueReference; -import org.apache.hyracks.data.std.primitive.BooleanPointable; -import org.apache.hyracks.data.std.primitive.BytePointable; -import org.apache.hyracks.data.std.primitive.DoublePointable; -import org.apache.hyracks.data.std.primitive.FloatPointable; -import org.apache.hyracks.data.std.primitive.IntegerPointable; -import org.apache.hyracks.data.std.primitive.LongPointable; -import org.apache.hyracks.data.std.primitive.ShortPointable; -import org.apache.hyracks.util.string.UTF8StringUtil; - -public class MessagePackerFromADM { - - private static final int TYPE_TAG_SIZE = 1; - private static final int TYPE_SIZE = 1; - private static final int LENGTH_SIZE = 4; - private static final int ITEM_COUNT_SIZE = 4; - private static final int ITEM_OFFSET_SIZE = 4; - - public static ATypeTag pack(IValueReference ptr, IAType type, ByteBuffer out, boolean packUnknown) - throws HyracksDataException { - return pack(ptr.getByteArray(), ptr.getStartOffset(), type, true, packUnknown, out); - } - - public static ATypeTag pack(byte[] ptr, int offs, IAType type, boolean tagged, boolean packUnknown, ByteBuffer out) - throws HyracksDataException { - int relOffs = tagged ? offs + 1 : offs; - ATypeTag tag = type.getTypeTag(); - switch (tag) { - case STRING: - packStr(ptr, relOffs, out); - break; - case BOOLEAN: - if (BooleanPointable.getBoolean(ptr, relOffs)) { - out.put(TRUE); - } else { - out.put(FALSE); - } - break; - case TINYINT: - packByte(out, BytePointable.getByte(ptr, relOffs)); - break; - case SMALLINT: - packShort(out, ShortPointable.getShort(ptr, relOffs)); - break; - case INTEGER: - packInt(out, IntegerPointable.getInteger(ptr, relOffs)); - break; - case BIGINT: - packLong(out, LongPointable.getLong(ptr, relOffs)); - break; - case FLOAT: - packFloat(out, FloatPointable.getFloat(ptr, relOffs)); - break; - case DOUBLE: - packDouble(out, DoublePointable.getDouble(ptr, relOffs)); - break; - case ARRAY: - case MULTISET: - packArray(ptr, offs, type, out); - break; - case OBJECT: - packObject(ptr, offs, type, out); - break; - case MISSING: - case NULL: - if (packUnknown) { - packNull(out); - break; - } else { - return tag; - } - default: - throw HyracksDataException.create(AsterixException.create(ErrorCode.PARSER_ADM_DATA_PARSER_CAST_ERROR, - tag.name(), "to a msgpack")); - } - return ATypeTag.TYPE; - } - - public static ATypeTag peekUnknown(IAType type) { - switch (type.getTypeTag()) { - case MISSING: - case NULL: - return type.getTypeTag(); - default: - return ATypeTag.TYPE; - } - } - - public static void packNull(ByteBuffer out) { - out.put(NIL); - } - - public static void packByte(ByteBuffer out, byte in) { - out.put(INT8); - out.put(in); - } - - public static void packShort(ByteBuffer out, short in) { - out.put(INT16); - out.putShort(in); - } - - public static void packInt(ByteBuffer out, int in) { - out.put(INT32); - out.putInt(in); - - } - - public static void packLong(ByteBuffer out, long in) { - out.put(INT64); - out.putLong(in); - } - - public static void packFloat(ByteBuffer out, float in) { - out.put(FLOAT32); - out.putFloat(in); - } - - public static void packDouble(ByteBuffer out, double in) { - out.put(FLOAT64); - out.putDouble(in); - } - - public static void packFixPos(ByteBuffer out, byte in) throws HyracksDataException { - byte mask = (byte) (1 << 7); - if ((in & mask) != 0) { - throw HyracksDataException.create(org.apache.hyracks.api.exceptions.ErrorCode.ILLEGAL_STATE, - "fixint7 must be positive"); - } - out.put(in); - } - - public static void packFixStr(ByteBuffer buf, String in) throws HyracksDataException { - byte[] strBytes = in.getBytes(StandardCharsets.UTF_8); - if (strBytes.length > 31) { - throw HyracksDataException.create(org.apache.hyracks.api.exceptions.ErrorCode.ILLEGAL_STATE, - "fixint7 must be positive"); - } - buf.put((byte) (FIXSTR_PREFIX + strBytes.length)); - buf.put(strBytes); - } - - public static void packStr(ByteBuffer out, String in) { - out.put(STR32); - byte[] strBytes = in.getBytes(StandardCharsets.UTF_8); - out.putInt(strBytes.length); - out.put(strBytes); - } - - private static void packStr(byte[] in, int offs, ByteBuffer out) { - out.put(STR32); - //TODO: tagged/untagged. closed support is borked so always tagged rn - String str = UTF8StringUtil.toString(in, offs); - byte[] strBytes = str.getBytes(StandardCharsets.UTF_8); - out.putInt(strBytes.length); - out.put(strBytes); - } - - public static void packStr(String str, ByteBuffer out) { - out.put(STR32); - byte[] strBytes = str.getBytes(StandardCharsets.UTF_8); - out.putInt(strBytes.length); - out.put(strBytes); - } - - private static void packArray(byte[] in, int offs, IAType type, ByteBuffer out) throws HyracksDataException { - //TODO: - could optimize to pack fixarray/array16 for small arrays - // - this code is basically a static version of AListPointable, could be deduped - AbstractCollectionType collType = (AbstractCollectionType) type; - out.put(ARRAY32); - int lenOffs = offs + TYPE_TAG_SIZE + TYPE_SIZE; - int itemCtOffs = LENGTH_SIZE + lenOffs; - int itemCt = IntegerPointable.getInteger(in, itemCtOffs); - boolean fixType = NonTaggedFormatUtil.isFixedSizedCollection(type); - out.putInt(itemCt); - for (int i = 0; i < itemCt; i++) { - if (fixType) { - int itemOffs = itemCtOffs + ITEM_COUNT_SIZE + (i - * NonTaggedFormatUtil.getFieldValueLength(in, 0, collType.getItemType().getTypeTag(), false)); - pack(in, itemOffs, collType.getItemType(), false, true, out); - } else { - int itemOffs = - offs + IntegerPointable.getInteger(in, itemCtOffs + ITEM_COUNT_SIZE + (i * ITEM_OFFSET_SIZE)); - ATypeTag tag = ATypeTag.VALUE_TYPE_MAPPING[BytePointable.getByte(in, itemOffs)]; - pack(in, itemOffs, TypeTagUtil.getBuiltinTypeByTag(tag), true, true, out); - } - } - } - - private static void packObject(byte[] in, int offs, IAType type, ByteBuffer out) throws HyracksDataException { - ARecordType recType = (ARecordType) type; - out.put(MAP32); - int fieldCt = recType.getFieldNames().length + RecordUtils.getOpenFieldCount(in, offs, recType); - out.putInt(fieldCt); - for (int i = 0; i < recType.getFieldNames().length; i++) { - String field = recType.getFieldNames()[i]; - IAType fieldType = RecordUtils.getClosedFieldType(recType, i); - packStr(field, out); - pack(in, RecordUtils.getClosedFieldOffset(in, offs, recType, i), fieldType, false, true, out); - } - if (RecordUtils.isExpanded(in, offs, recType)) { - for (int i = 0; i < RecordUtils.getOpenFieldCount(in, offs, recType); i++) { - packStr(in, RecordUtils.getOpenFieldNameOffset(in, offs, recType, i), out); - ATypeTag tag = ATypeTag.VALUE_TYPE_MAPPING[RecordUtils.getOpenFieldTag(in, offs, recType, i)]; - pack(in, RecordUtils.getOpenFieldValueOffset(in, offs, recType, i), - TypeTagUtil.getBuiltinTypeByTag(tag), true, true, out); - } - } - - } - - public static void packFixArrayHeader(ByteBuffer buf, byte numObj) { - buf.put((byte) (FIXARRAY_PREFIX + (0x0F & numObj))); - } - - private static class RecordUtils { - - static final int TAG_SIZE = 1; - static final int RECORD_LENGTH_SIZE = 4; - static final int EXPANDED_SIZE = 1; - static final int OPEN_OFFSET_SIZE = 4; - static final int CLOSED_COUNT_SIZE = 4; - static final int FIELD_OFFSET_SIZE = 4; - static final int OPEN_COUNT_SIZE = 4; - private static final int OPEN_FIELD_HASH_SIZE = 4; - private static final int OPEN_FIELD_OFFSET_SIZE = 4; - private static final int OPEN_FIELD_HEADER = OPEN_FIELD_HASH_SIZE + OPEN_FIELD_OFFSET_SIZE; - - private static boolean isOpen(ARecordType recordType) { - return recordType == null || recordType.isOpen(); - } - - public static int getLength(byte[] bytes, int start) { - return IntegerPointable.getInteger(bytes, start + TAG_SIZE); - } - - public static boolean isExpanded(byte[] bytes, int start, ARecordType recordType) { - return isOpen(recordType) && BooleanPointable.getBoolean(bytes, start + TAG_SIZE + RECORD_LENGTH_SIZE); - } - - public static int getOpenPartOffset(int start, ARecordType recordType) { - return start + TAG_SIZE + RECORD_LENGTH_SIZE + (isOpen(recordType) ? EXPANDED_SIZE : 0); - } - - public static int getNullBitmapOffset(byte[] bytes, int start, ARecordType recordType) { - return getOpenPartOffset(start, recordType) + (isExpanded(bytes, start, recordType) ? OPEN_OFFSET_SIZE : 0) - + CLOSED_COUNT_SIZE; - } - - public static int getNullBitmapSize(ARecordType recordType) { - return RecordUtil.computeNullBitmapSize(recordType); - } - - public static final IAType getClosedFieldType(ARecordType recordType, int fieldId) { - IAType aType = recordType.getFieldTypes()[fieldId]; - if (NonTaggedFormatUtil.isOptional(aType)) { - // optional field: add the embedded non-null type tag - aType = ((AUnionType) aType).getActualType(); - } - return aType; - } - - public static final int getClosedFieldOffset(byte[] bytes, int start, ARecordType recordType, int fieldId) { - int offset = getNullBitmapOffset(bytes, start, recordType) + getNullBitmapSize(recordType) - + fieldId * FIELD_OFFSET_SIZE; - return start + IntegerPointable.getInteger(bytes, offset); - } - - public static final int getOpenFieldCount(byte[] bytes, int start, ARecordType recordType) { - return isExpanded(bytes, start, recordType) - ? IntegerPointable.getInteger(bytes, getOpenFieldCountOffset(bytes, start, recordType)) : 0; - } - - public static int getOpenFieldCountSize(byte[] bytes, int start, ARecordType recordType) { - return isExpanded(bytes, start, recordType) ? OPEN_COUNT_SIZE : 0; - } - - public static int getOpenFieldCountOffset(byte[] bytes, int start, ARecordType recordType) { - return start + IntegerPointable.getInteger(bytes, getOpenPartOffset(start, recordType)); - } - - public static final int getOpenFieldValueOffset(byte[] bytes, int start, ARecordType recordType, int fieldId) { - return getOpenFieldNameOffset(bytes, start, recordType, fieldId) - + getOpenFieldNameSize(bytes, start, recordType, fieldId); - } - - public static int getOpenFieldNameSize(byte[] bytes, int start, ARecordType recordType, int fieldId) { - int utfleng = UTF8StringUtil.getUTFLength(bytes, getOpenFieldNameOffset(bytes, start, recordType, fieldId)); - return utfleng + UTF8StringUtil.getNumBytesToStoreLength(utfleng); - } - - public static int getOpenFieldNameOffset(byte[] bytes, int start, ARecordType recordType, int fieldId) { - return getOpenFieldOffset(bytes, start, recordType, fieldId); - } - - public static final byte getOpenFieldTag(byte[] bytes, int start, ARecordType recordType, int fieldId) { - return bytes[getOpenFieldValueOffset(bytes, start, recordType, fieldId)]; - } - - public static int getOpenFieldHashOffset(byte[] bytes, int start, ARecordType recordType, int fieldId) { - return getOpenFieldCountOffset(bytes, start, recordType) + getOpenFieldCountSize(bytes, start, recordType) - + fieldId * OPEN_FIELD_HEADER; - } - - public static int getOpenFieldOffset(byte[] bytes, int start, ARecordType recordType, int fieldId) { - return start - + IntegerPointable.getInteger(bytes, getOpenFieldOffsetOffset(bytes, start, recordType, fieldId)); - } - - public static int getOpenFieldOffsetOffset(byte[] bytes, int start, ARecordType recordType, int fieldId) { - return getOpenFieldHashOffset(bytes, start, recordType, fieldId) + OPEN_FIELD_HASH_SIZE; - } - } - -} diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessageUnpackerToADM.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessageUnpackerToADM.java index 4af1121..d6121e8 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessageUnpackerToADM.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MessageUnpackerToADM.java @@ -18,18 +18,39 @@ package org.apache.asterix.external.library.msgpack; import static org.msgpack.core.MessagePack.Code.*; +import java.io.DataOutput; +import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.List; +import org.apache.asterix.builders.AbvsBuilderFactory; import org.apache.asterix.common.exceptions.AsterixException; import org.apache.asterix.common.exceptions.ErrorCode; +import org.apache.asterix.external.input.stream.StandardUTF8ToModifiedUTF8DataOutput; +import org.apache.asterix.external.input.stream.builders.ListLikeNumericArrayFactory; +import org.apache.asterix.external.input.stream.builders.StandardToModifiedUTF8DataOutputFactory; import org.apache.asterix.om.types.ATypeTag; +import org.apache.asterix.om.util.container.IObjectPool; +import org.apache.asterix.om.util.container.ListObjectPool; import org.apache.hyracks.api.exceptions.HyracksDataException; -import org.apache.hyracks.util.encoding.VarLenIntEncoderDecoder; +import org.apache.hyracks.data.std.api.IMutableValueStorage; +import org.apache.hyracks.data.std.primitive.IntegerPointable; +import org.apache.hyracks.data.std.util.ArrayBackedValueStorage; import org.apache.hyracks.util.string.UTF8StringUtil; public class MessageUnpackerToADM { - public static void unpack(ByteBuffer in, ByteBuffer out, boolean tagged) throws HyracksDataException { + private final IObjectPool<IMutableValueStorage, ATypeTag> abvsBuilderPool = + new ListObjectPool<>(new AbvsBuilderFactory()); + private final IObjectPool<StandardUTF8ToModifiedUTF8DataOutput, ATypeTag> utfPool = + new ListObjectPool<>(new StandardToModifiedUTF8DataOutputFactory()); + private final IObjectPool<List<Long>, Long> listPool = new ListObjectPool<>(new ListLikeNumericArrayFactory<>()); + + public MessageUnpackerToADM() { + } + + public void unpack(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { byte tag = NIL; if (in != null) { tag = in.get(); @@ -38,12 +59,12 @@ public class MessageUnpackerToADM { unpackStr(in, out, (tag ^ FIXSTR_PREFIX), tagged); } else if (isFixInt(tag)) { if (tagged) { - out.put(ATypeTag.SERIALIZED_INT8_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_INT8_TYPE_TAG); } if (isPosFixInt(tag)) { - out.put(tag); + out.writeByte(tag); } else if (isNegFixInt(tag)) { - out.put(tag); + out.writeByte(tag); } } else if (isFixedArray(tag)) { unpackArray(in, out, (tag ^ FIXARRAY_PREFIX)); @@ -52,15 +73,15 @@ public class MessageUnpackerToADM { } else { switch (tag) { case TRUE: - out.put(ATypeTag.SERIALIZED_BOOLEAN_TYPE_TAG); - out.put((byte) 1); + out.writeByte(ATypeTag.SERIALIZED_BOOLEAN_TYPE_TAG); + out.writeByte((byte) 1); break; case FALSE: - out.put(ATypeTag.SERIALIZED_BOOLEAN_TYPE_TAG); - out.put((byte) 0); + out.writeByte(ATypeTag.SERIALIZED_BOOLEAN_TYPE_TAG); + out.writeByte((byte) 0); break; case NIL: - out.put(ATypeTag.SERIALIZED_NULL_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_NULL_TYPE_TAG); break; case UINT8: unpackUByte(in, out, tagged); @@ -113,7 +134,6 @@ public class MessageUnpackerToADM { case MAP32: unpackMap(in, out, (int) Integer.toUnsignedLong(in.getInt())); break; - default: throw HyracksDataException.create(AsterixException.create( ErrorCode.PARSER_ADM_DATA_PARSER_CAST_ERROR, "msgpack tag " + tag + " ", "to an ADM type")); @@ -121,154 +141,159 @@ public class MessageUnpackerToADM { } } - public static void unpackByte(ByteBuffer in, ByteBuffer out, boolean tagged) { + public static void unpackByte(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { if (tagged) { - out.put(ATypeTag.SERIALIZED_INT8_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_INT8_TYPE_TAG); } - out.put(in.get()); + out.writeByte(in.get()); } - public static void unpackShort(ByteBuffer in, ByteBuffer out, boolean tagged) { + public static void unpackShort(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { if (tagged) { - out.put(ATypeTag.SERIALIZED_INT16_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_INT16_TYPE_TAG); } - out.putShort(in.getShort()); + out.writeShort(in.getShort()); } - public static void unpackInt(ByteBuffer in, ByteBuffer out, boolean tagged) { + public static void unpackInt(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { if (tagged) { - out.put(ATypeTag.SERIALIZED_INT32_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_INT32_TYPE_TAG); } - out.putInt(in.getInt()); + out.writeInt(in.getInt()); } - public static void unpackLong(ByteBuffer in, ByteBuffer out, boolean tagged) { + public static void unpackLong(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { if (tagged) { - out.put(ATypeTag.SERIALIZED_INT64_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_INT64_TYPE_TAG); } - out.putLong(in.getLong()); + out.writeLong(in.getLong()); } - public static void unpackUByte(ByteBuffer in, ByteBuffer out, boolean tagged) { + public static void unpackUByte(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { if (tagged) { - out.put(ATypeTag.SERIALIZED_INT16_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_INT16_TYPE_TAG); } - out.putShort((short) (in.get() & ((short) 0x00FF))); + out.writeShort((short) (in.get() & ((short) 0x00FF))); } - public static void unpackUShort(ByteBuffer in, ByteBuffer out, boolean tagged) { + public static void unpackUShort(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { if (tagged) { - out.put(ATypeTag.SERIALIZED_INT32_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_INT32_TYPE_TAG); } - out.putInt(in.getShort() & 0x0000FFFF); + out.writeInt(in.getShort() & 0x0000FFFF); } - public static void unpackUInt(ByteBuffer in, ByteBuffer out, boolean tagged) { + public static void unpackUInt(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { if (tagged) { - out.put(ATypeTag.SERIALIZED_INT64_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_INT64_TYPE_TAG); } - out.putLong(in.getInt() & 0x00000000FFFFFFFFl); + out.writeLong(in.getInt() & 0x00000000FFFFFFFFl); } - public static void unpackULong(ByteBuffer in, ByteBuffer out, boolean tagged) { + public static void unpackULong(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { if (tagged) { - out.put(ATypeTag.SERIALIZED_INT64_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_INT64_TYPE_TAG); } long val = in.getLong(); if (val < 0) { throw new IllegalArgumentException("Integer overflow"); } - out.putLong(val); + out.writeLong(val); } - public static void unpackFloat(ByteBuffer in, ByteBuffer out, boolean tagged) { + public static void unpackFloat(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { if (tagged) { - out.put(ATypeTag.SERIALIZED_FLOAT_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_FLOAT_TYPE_TAG); } - out.putFloat(in.getFloat()); + out.writeFloat(in.getFloat()); } - public static void unpackDouble(ByteBuffer in, ByteBuffer out, boolean tagged) { + public static void unpackDouble(ByteBuffer in, DataOutput out, boolean tagged) throws IOException { if (tagged) { - out.put(ATypeTag.SERIALIZED_DOUBLE_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_DOUBLE_TYPE_TAG); } - out.putDouble(in.getDouble()); + out.writeDouble(in.getDouble()); } - public static void unpackArray(ByteBuffer in, ByteBuffer out, long uLen) throws HyracksDataException { + public void unpackArray(ByteBuffer in, DataOutput out, long uLen) throws IOException { if (uLen > Integer.MAX_VALUE) { throw new UnsupportedOperationException("Array is too long"); } + ArrayBackedValueStorage buildBuf = (ArrayBackedValueStorage) abvsBuilderPool.allocate(ATypeTag.ARRAY); + buildBuf.reset(); + DataOutput bufOut = buildBuf.getDataOutput(); int count = (int) uLen; - int offs = out.position(); - out.put(ATypeTag.SERIALIZED_ORDEREDLIST_TYPE_TAG); - out.put(ATypeTag.ANY.serialize()); - int asxLenPos = out.position(); + bufOut.writeByte(ATypeTag.SERIALIZED_ORDEREDLIST_TYPE_TAG); + bufOut.writeByte(ATypeTag.ANY.serialize()); + int asxLenPos = buildBuf.getLength(); //reserve space - out.putInt(-1); - out.putInt(count); - int slotStartOffs = out.position() + out.arrayOffset(); + bufOut.writeInt(-1); + bufOut.writeInt(count); + int slotStartOffs = buildBuf.getLength(); for (int i = 0; i < count; i++) { - out.putInt(0xFFFF); + bufOut.writeInt(0xDEADBEEF); } for (int i = 0; i < count; i++) { - out.putInt(slotStartOffs + (i * 4), (out.position() - offs)); - unpack(in, out, true); + IntegerPointable.setInteger(buildBuf.getByteArray(), ((slotStartOffs) + (i * 4)), buildBuf.getLength()); + //tagged b/c any + unpack(in, bufOut, true); } - int totalLen = out.position() - offs; - out.putInt(asxLenPos, totalLen); + IntegerPointable.setInteger(buildBuf.getByteArray(), asxLenPos, buildBuf.getLength()); + out.write(buildBuf.getByteArray(), buildBuf.getStartOffset(), buildBuf.getLength()); } - public static void unpackMap(ByteBuffer in, ByteBuffer out, int count) throws HyracksDataException { + public void unpackMap(ByteBuffer in, DataOutput out, int count) throws IOException { //TODO: need to handle typed records. this only produces a completely open record. + ArrayBackedValueStorage buildBuf = (ArrayBackedValueStorage) abvsBuilderPool.allocate(ATypeTag.OBJECT); + List<Long> offsets = listPool.allocate((long) count); + DataOutput bufOut = buildBuf.getDataOutput(); //hdr size = 6? - int startOffs = out.position(); - out.put(ATypeTag.SERIALIZED_RECORD_TYPE_TAG); - int totalSizeOffs = out.position(); - out.putInt(-1); + bufOut.writeByte(ATypeTag.SERIALIZED_RECORD_TYPE_TAG); + int totalSizeOffs = buildBuf.getLength(); + bufOut.writeInt(-1); //isExpanded - out.put((byte) 1); - int openPartOffs = out.position(); - out.putInt(-1); + bufOut.writeByte((byte) 1); + int openPartOffs = buildBuf.getLength(); + bufOut.writeInt(-1); //isExpanded, so num of open fields - out.putInt(openPartOffs, out.position() - startOffs); - out.putInt(count); - int offsetAryPos = out.position(); + IntegerPointable.setInteger(buildBuf.getByteArray(), openPartOffs, buildBuf.getLength()); + bufOut.writeInt(count); + int offsetAryPos = buildBuf.getLength(); int offsetArySz = count * 2; //allocate space for open field offsets for (int i = 0; i < offsetArySz; i++) { - out.putInt(0xDEADBEEF); + bufOut.writeInt(0xDEADBEEF); } for (int i = 0; i < count; i++) { - int offs = out.position() + out.arrayOffset(); - int relOffs = offs - startOffs; - unpack(in, out, false); - int hash = UTF8StringUtil.hash(out.array(), offs); - out.putInt(offsetAryPos, hash); + int offs = buildBuf.getLength(); + unpack(in, bufOut, false); + long hash = UTF8StringUtil.hash(buildBuf.getByteArray(), offs); + offsets.set(i, (hash << 32) + offs); + unpack(in, bufOut, true); + } + Collections.sort(offsets); + for (Long l : offsets) { + IntegerPointable.setInteger(buildBuf.getByteArray(), offsetAryPos, (int) (l >> 32)); offsetAryPos += 4; - out.putInt(offsetAryPos, relOffs); + IntegerPointable.setInteger(buildBuf.getByteArray(), offsetAryPos, (int) ((l << 32) >> 32)); offsetAryPos += 4; - unpack(in, out, true); } - out.putInt(totalSizeOffs, out.position() - startOffs); + IntegerPointable.setInteger(buildBuf.getByteArray(), totalSizeOffs, buildBuf.getLength()); + out.write(buildBuf.getByteArray(), buildBuf.getStartOffset(), buildBuf.getLength()); } - public static void unpackStr(ByteBuffer in, ByteBuffer out, long uLen, boolean tag) { - //TODO: this probably breaks for 3 and 4 byte UTF-8 + public void unpackStr(ByteBuffer in, DataOutput out, long uLen, boolean tag) throws IOException { if (tag) { - out.put(ATypeTag.SERIALIZED_STRING_TYPE_TAG); + out.writeByte(ATypeTag.SERIALIZED_STRING_TYPE_TAG); } if (Long.compareUnsigned(uLen, Integer.MAX_VALUE) > 0) { throw new UnsupportedOperationException("String is too long"); } int len = (int) uLen; - int strLen = UTF8StringUtil.getStringLength(in.array(), in.position() + in.arrayOffset(), len); - int adv = VarLenIntEncoderDecoder.encode(strLen, out.array(), out.position() + out.arrayOffset()); - out.position(out.position() + adv); - System.arraycopy(in.array(), in.arrayOffset() + in.position(), out.array(), out.arrayOffset() + out.position(), - len); - out.position(out.position() + len); + StandardUTF8ToModifiedUTF8DataOutput conv = utfPool.allocate(ATypeTag.STRING); + conv.setDataOutput(out); + conv.write(in.array(), in.arrayOffset() + in.position(), len); in.position(in.position() + len); } diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MsgPackAccessors.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MsgPackAccessors.java new file mode 100644 index 0000000..a90a183 --- /dev/null +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MsgPackAccessors.java @@ -0,0 +1,275 @@ +/* + * 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.asterix.external.library.msgpack; + +import static org.apache.hyracks.util.string.UTF8StringUtil.getUTFLength; +import static org.msgpack.core.MessagePack.Code.ARRAY32; +import static org.msgpack.core.MessagePack.Code.FALSE; +import static org.msgpack.core.MessagePack.Code.FLOAT32; +import static org.msgpack.core.MessagePack.Code.FLOAT64; +import static org.msgpack.core.MessagePack.Code.INT16; +import static org.msgpack.core.MessagePack.Code.INT32; +import static org.msgpack.core.MessagePack.Code.INT64; +import static org.msgpack.core.MessagePack.Code.INT8; +import static org.msgpack.core.MessagePack.Code.MAP32; +import static org.msgpack.core.MessagePack.Code.NIL; +import static org.msgpack.core.MessagePack.Code.STR32; +import static org.msgpack.core.MessagePack.Code.TRUE; + +import java.io.DataOutput; +import java.io.IOException; +import java.util.List; + +import org.apache.asterix.common.exceptions.AsterixException; +import org.apache.asterix.common.exceptions.ErrorCode; +import org.apache.asterix.dataflow.data.nontagged.printers.PrintTools; +import org.apache.asterix.dataflow.data.nontagged.serde.ABooleanSerializerDeserializer; +import org.apache.asterix.dataflow.data.nontagged.serde.ADoubleSerializerDeserializer; +import org.apache.asterix.dataflow.data.nontagged.serde.AFloatSerializerDeserializer; +import org.apache.asterix.dataflow.data.nontagged.serde.AInt16SerializerDeserializer; +import org.apache.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer; +import org.apache.asterix.dataflow.data.nontagged.serde.AInt64SerializerDeserializer; +import org.apache.asterix.dataflow.data.nontagged.serde.AInt8SerializerDeserializer; +import org.apache.asterix.external.library.PyTypeInfo; +import org.apache.asterix.om.pointables.AListVisitablePointable; +import org.apache.asterix.om.pointables.ARecordVisitablePointable; +import org.apache.asterix.om.pointables.base.IVisitablePointable; +import org.apache.asterix.om.types.ARecordType; +import org.apache.asterix.om.types.ATypeTag; +import org.apache.asterix.om.types.AbstractCollectionType; +import org.apache.asterix.om.types.BuiltinType; +import org.apache.asterix.om.types.EnumDeserializer; +import org.apache.asterix.om.types.IAType; +import org.apache.asterix.om.types.TypeTagUtil; +import org.apache.hyracks.api.exceptions.HyracksDataException; +import org.apache.hyracks.data.std.api.IPointable; +import org.apache.hyracks.util.encoding.VarLenIntEncoderDecoder; + +public class MsgPackAccessors { + + private MsgPackAccessors() { + } + + public static IMsgPackAccessor<IPointable, DataOutput, Void> createFlatMsgPackAccessor(ATypeTag aTypeTag) + throws HyracksDataException { + switch (aTypeTag) { + case BOOLEAN: + return MsgPackBooleanAccessor::apply; + case TINYINT: + return MsgPackInt8Accessor::apply; + case SMALLINT: + return MsgPackInt16Accessor::apply; + case INTEGER: + return MsgPackInt32Accessor::apply; + case BIGINT: + return MsgPackInt64Accessor::apply; + case FLOAT: + return MsgPackFloatAccessor::apply; + case DOUBLE: + return MsgPackDoubleAccessor::apply; + case STRING: + return MsgPackStringAccessor::apply; + case MISSING: + case NULL: + return MsgPackNullAccessor::apply; + default: + throw HyracksDataException + .create(AsterixException.create(ErrorCode.TYPE_UNSUPPORTED, "msgpack", aTypeTag.name())); + } + } + + public static class MsgPackInt8Accessor { + public static Void apply(IPointable pointable, DataOutput out) throws IOException { + byte[] b = pointable.getByteArray(); + int s = pointable.getStartOffset(); + byte o = AInt8SerializerDeserializer.getByte(b, s + 1); + out.writeByte(INT8); + out.writeByte(o); + return null; + } + } + + public static class MsgPackInt16Accessor { + + public static Void apply(IPointable pointable, DataOutput out) throws IOException { + byte[] b = pointable.getByteArray(); + int s = pointable.getStartOffset(); + short i = AInt16SerializerDeserializer.getShort(b, s + 1); + out.writeByte(INT16); + out.writeShort(i); + return null; + } + } + + public static class MsgPackInt32Accessor { + + public static Void apply(IPointable pointable, DataOutput out) throws IOException { + byte[] b = pointable.getByteArray(); + int s = pointable.getStartOffset(); + int i = AInt32SerializerDeserializer.getInt(b, s + 1); + out.writeByte(INT32); + out.writeByte(i); + return null; + } + } + + public static class MsgPackNullAccessor { + public static Void apply(IPointable pointable, DataOutput out) throws IOException { + out.writeByte(NIL); + return null; + } + } + + public static class MsgPackInt64Accessor { + + public static Void apply(IPointable pointable, DataOutput out) throws IOException { + byte[] b = pointable.getByteArray(); + int s = pointable.getStartOffset(); + long v = AInt64SerializerDeserializer.getLong(b, s + 1); + out.writeByte(INT64); + out.writeLong(v); + return null; + } + } + + public static class MsgPackFloatAccessor { + + public static Void apply(IPointable pointable, DataOutput out) throws IOException { + byte[] b = pointable.getByteArray(); + int s = pointable.getStartOffset(); + float v = AFloatSerializerDeserializer.getFloat(b, s + 1); + out.writeByte(FLOAT32); + out.writeFloat(v); + return null; + } + } + + public static class MsgPackDoubleAccessor { + public static Void apply(IPointable pointable, DataOutput out) throws IOException { + byte[] b = pointable.getByteArray(); + int s = pointable.getStartOffset(); + double v = ADoubleSerializerDeserializer.getDouble(b, s + 1); + out.writeByte(FLOAT64); + out.writeDouble(v); + return null; + } + } + + public static class MsgPackStringAccessor { + public static Void apply(IPointable pointable, DataOutput out) throws IOException { + byte[] b = pointable.getByteArray(); + int s = pointable.getStartOffset(); + out.writeByte(STR32); + final int calculatedLength = getUTFLength(b, s + 1); + out.writeInt(calculatedLength); + PrintTools.writeUTF8StringRaw(b, s + 1, calculatedLength, out); + return null; + } + + } + + public static class MsgPackBooleanAccessor { + public static Void apply(IPointable pointable, DataOutput out) throws IOException { + byte[] b = pointable.getByteArray(); + int s = pointable.getStartOffset(); + boolean v = ABooleanSerializerDeserializer.getBoolean(b, s + 1); + if (v) { + out.writeByte(TRUE); + } else { + out.writeByte(FALSE); + } + return null; + } + } + + public static class MsgPackRecordAccessor { + + public static int getUTFLength(byte[] b, int s) { + return VarLenIntEncoderDecoder.decode(b, s); + } + + public static Void access(ARecordVisitablePointable pointable, PyTypeInfo arg, + MsgPackPointableVisitor pointableVisitor) throws HyracksDataException { + List<IVisitablePointable> fieldPointables = pointable.getFieldValues(); + List<IVisitablePointable> fieldTypeTags = pointable.getFieldTypeTags(); + List<IVisitablePointable> fieldNames = pointable.getFieldNames(); + boolean closedPart; + int index = 0; + DataOutput out = arg.getDataOutput(); + ARecordType recordType = ((ARecordType) arg.getType()); + try { + out.writeByte(MAP32); + out.writeInt(fieldNames.size()); + } catch (IOException e) { + throw HyracksDataException.create(e); + } + try { + for (IVisitablePointable fieldPointable : fieldPointables) { + closedPart = index < recordType.getFieldTypes().length; + IVisitablePointable tt = fieldTypeTags.get(index); + ATypeTag typeTag = + EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(tt.getByteArray()[tt.getStartOffset()]); + IAType fieldType; + fieldType = + closedPart ? recordType.getFieldTypes()[index] : TypeTagUtil.getBuiltinTypeByTag(typeTag); + IPointable fieldName = fieldNames.get(index); + MsgPackAccessors.createFlatMsgPackAccessor(BuiltinType.ASTRING.getTypeTag()).apply(fieldName, + arg.getDataOutput()); + PyTypeInfo fieldTypeInfo = pointableVisitor.getTypeInfo(fieldType, out); + fieldPointable.accept(pointableVisitor, fieldTypeInfo); + index++; + } + } catch (IOException e) { + throw HyracksDataException.create(e); + } + return null; + } + } + + public static class MsgPackListAccessor { + + public static Void access(AListVisitablePointable pointable, PyTypeInfo arg, + MsgPackPointableVisitor pointableVisitor) throws HyracksDataException { + List<IVisitablePointable> items = pointable.getItems(); + List<IVisitablePointable> itemTags = pointable.getItemTags(); + DataOutput out = arg.getDataOutput(); + try { + out.writeByte(ARRAY32); + out.writeInt(items.size()); + } catch (IOException e) { + throw HyracksDataException.create(e); + } + for (int iter1 = 0; iter1 < items.size(); iter1++) { + IVisitablePointable itemPointable = items.get(iter1); + // First, try to get defined type. + IAType fieldType = ((AbstractCollectionType) arg.getType()).getItemType(); + if (fieldType.getTypeTag() == ATypeTag.ANY) { + // Second, if defined type is not available, try to infer it from data + IVisitablePointable itemTagPointable = itemTags.get(iter1); + ATypeTag itemTypeTag = EnumDeserializer.ATYPETAGDESERIALIZER + .deserialize(itemTagPointable.getByteArray()[itemTagPointable.getStartOffset()]); + fieldType = TypeTagUtil.getBuiltinTypeByTag(itemTypeTag); + } + PyTypeInfo fieldTypeInfo = pointableVisitor.getTypeInfo(fieldType, out); + itemPointable.accept(pointableVisitor, fieldTypeInfo); + } + return null; + } + } +} diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MsgPackPointableVisitor.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MsgPackPointableVisitor.java new file mode 100644 index 0000000..08be6ac --- /dev/null +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/msgpack/MsgPackPointableVisitor.java @@ -0,0 +1,77 @@ +/* + * 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.asterix.external.library.msgpack; + +import java.io.DataOutput; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import org.apache.asterix.external.library.PyTypeInfo; +import org.apache.asterix.om.pointables.AFlatValuePointable; +import org.apache.asterix.om.pointables.AListVisitablePointable; +import org.apache.asterix.om.pointables.ARecordVisitablePointable; +import org.apache.asterix.om.pointables.visitor.IVisitablePointableVisitor; +import org.apache.asterix.om.types.IAType; +import org.apache.hyracks.api.exceptions.HyracksDataException; + +public class MsgPackPointableVisitor implements IVisitablePointableVisitor<Void, PyTypeInfo> { + + Map<DataOutput, Map<IAType, PyTypeInfo>> typeInfoMap = new HashMap<>(); + + public Void visit(AListVisitablePointable accessor, PyTypeInfo arg) throws HyracksDataException { + MsgPackAccessors.MsgPackListAccessor.access(accessor, arg, this); + return null; + } + + @Override + public Void visit(ARecordVisitablePointable accessor, PyTypeInfo arg) throws HyracksDataException { + MsgPackAccessors.MsgPackRecordAccessor.access(accessor, arg, this); + return null; + } + + @Override + public Void visit(AFlatValuePointable accessor, PyTypeInfo arg) throws HyracksDataException { + try { + MsgPackAccessors.createFlatMsgPackAccessor(arg.getType().getTypeTag()).apply(accessor, arg.getDataOutput()); + } catch (IOException e) { + throw HyracksDataException.create(e); + } + return null; + } + + public PyTypeInfo getTypeInfo(IAType type, DataOutput out) { + PyTypeInfo tInfo = null; + Map<IAType, PyTypeInfo> type2TypeInfo = typeInfoMap.get(out); + if (type2TypeInfo == null) { + type2TypeInfo = new HashMap<>(); + tInfo = new PyTypeInfo(type, out); + type2TypeInfo.put(type, tInfo); + typeInfoMap.put(out, type2TypeInfo); + } + tInfo = tInfo == null ? type2TypeInfo.get(type) : tInfo; + if (tInfo == null) { + tInfo = new PyTypeInfo(type, out); + type2TypeInfo.put(type, tInfo); + } + return tInfo; + } + +} 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 593bac6..741dad2 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 @@ -33,11 +33,14 @@ import java.util.List; import org.apache.asterix.common.exceptions.AsterixException; import org.apache.asterix.common.exceptions.ErrorCode; import org.apache.asterix.common.exceptions.RuntimeDataException; +import org.apache.asterix.external.ipc.PythonIPCProto; import org.apache.asterix.external.library.PythonLibraryEvaluator; import org.apache.asterix.external.library.PythonLibraryEvaluatorFactory; import org.apache.asterix.external.library.msgpack.MessageUnpackerToADM; -import org.apache.asterix.external.util.ExternalDataUtils; +import org.apache.asterix.external.library.msgpack.MsgPackPointableVisitor; +import org.apache.asterix.external.util.ExternalDataConstants; import org.apache.asterix.om.functions.IExternalFunctionDescriptor; +import org.apache.asterix.om.pointables.PointableAllocator; import org.apache.asterix.om.types.ATypeTag; import org.apache.hyracks.algebricks.common.utils.Pair; import org.apache.hyracks.algebricks.core.algebra.base.Counter; @@ -48,6 +51,7 @@ import org.apache.hyracks.api.context.IHyracksTaskContext; import org.apache.hyracks.api.exceptions.HyracksDataException; import org.apache.hyracks.api.exceptions.Warning; import org.apache.hyracks.data.std.primitive.VoidPointable; +import org.apache.hyracks.data.std.util.ArrayBackedValueStorage; import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder; import org.apache.hyracks.dataflow.common.data.accessors.FrameTupleReference; import org.msgpack.core.MessagePack; @@ -58,12 +62,10 @@ import org.msgpack.core.buffer.ArrayBufferInput; public final class ExternalAssignBatchRuntimeFactory extends AbstractOneInputOneOutputRuntimeFactory { private static final long serialVersionUID = 1L; - private int[] outColumns; + private final int[] outColumns; private final IExternalFunctionDescriptor[] fnDescs; private final int[][] fnArgColumns; - private int rpcBufferSize; - public ExternalAssignBatchRuntimeFactory(int[] outColumns, IExternalFunctionDescriptor[] fnDescs, int[][] fnArgColumns, int[] projectionList) { super(projectionList); @@ -76,17 +78,14 @@ public final class ExternalAssignBatchRuntimeFactory extends AbstractOneInputOne public AbstractOneInputOneOutputPushRuntime createOneOutputPushRuntime(IHyracksTaskContext ctx) { final int[] projectionToOutColumns = new int[projectionList.length]; - //this is a temporary bodge. these buffers need to work like vsize frames, or be absent entirely - int maxArgSz = ExternalDataUtils.getArgBufferSize(); - rpcBufferSize = ExternalDataUtils.roundUpToNearestFrameSize(maxArgSz, ctx.getInitialFrameSize()); for (int j = 0; j < projectionList.length; j++) { projectionToOutColumns[j] = Arrays.binarySearch(outColumns, projectionList[j]); } return new AbstractOneInputOneOutputOneFramePushRuntime() { - private ByteBuffer outputWrapper; - private List<ByteBuffer> argHolders; + private ArrayBackedValueStorage outputWrapper; + private List<ArrayBackedValueStorage> argHolders; ArrayTupleBuilder tupleBuilder; private List<Pair<Long, PythonLibraryEvaluator>> libraryEvaluators; private ATypeTag[][] nullCalls; @@ -95,6 +94,9 @@ public final class ExternalAssignBatchRuntimeFactory extends AbstractOneInputOne private MessageUnpacker unpacker; private ArrayBufferInput unpackerInput; private List<Pair<ByteBuffer, Counter>> batchResults; + private MessageUnpackerToADM unpackerToADM; + private PointableAllocator pointableAllocator; + private MsgPackPointableVisitor pointableVisitor; @Override public void open() throws HyracksDataException { @@ -116,23 +118,26 @@ public final class ExternalAssignBatchRuntimeFactory extends AbstractOneInputOne } argHolders = new ArrayList<>(fnArgColumns.length); for (int i = 0; i < fnArgColumns.length; i++) { - argHolders.add(ctx.allocateFrame(rpcBufferSize)); + argHolders.add(new ArrayBackedValueStorage()); } - outputWrapper = ctx.allocateFrame(); + outputWrapper = new ArrayBackedValueStorage(); nullCalls = new ATypeTag[argHolders.size()][0]; numCalls = new int[fnArgColumns.length]; batchResults = new ArrayList<>(argHolders.size()); for (int i = 0; i < argHolders.size(); i++) { - batchResults.add(new Pair<>(ctx.allocateFrame(rpcBufferSize), new Counter(-1))); + batchResults.add(new Pair<>(ByteBuffer.allocate(ExternalDataConstants.DEFAULT_BUFFER_SIZE), + new Counter(-1))); } unpackerInput = new ArrayBufferInput(new byte[0]); unpacker = MessagePack.newDefaultUnpacker(unpackerInput); + unpackerToADM = new MessageUnpackerToADM(); + pointableAllocator = new PointableAllocator(); + pointableVisitor = new MsgPackPointableVisitor(); } private void resetBuffers(int numTuples, int[] numCalls) { for (int func = 0; func < fnArgColumns.length; func++) { - argHolders.get(func).clear(); - argHolders.get(func).position(0); + argHolders.get(func).reset(); if (nullCalls[func].length < numTuples) { nullCalls[func] = new ATypeTag[numTuples]; } @@ -188,6 +193,8 @@ public final class ExternalAssignBatchRuntimeFactory extends AbstractOneInputOne @Override public void nextFrame(ByteBuffer buffer) throws HyracksDataException { + /*TODO: this whole transposition stuff is not necessary + the evaulator should accept a format that is a collection of rows, logically*/ tAccess.reset(buffer); tupleBuilder.reset(); try { @@ -211,14 +218,15 @@ public final class ExternalAssignBatchRuntimeFactory extends AbstractOneInputOne } if (argumentStatus == ATypeTag.TYPE) { if (cols.length > 0) { - argHolders.get(func).put(ARRAY16); - argHolders.get(func).putShort((short) cols.length); + 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])); - PythonLibraryEvaluator.setArgument(fnDescs[func].getArgumentTypes()[colIdx], ref, - argHolders.get(func), fnDescs[func].getFunctionInfo().getNullCall()); + PythonIPCProto.visitValueRef(fnDescs[func].getArgumentTypes()[colIdx], + argHolders.get(func).getDataOutput(), ref, pointableAllocator, + pointableVisitor, fnDescs[func].getFunctionInfo().getNullCall()); } } else { numCalls[func]--; @@ -228,6 +236,7 @@ public final class ExternalAssignBatchRuntimeFactory extends AbstractOneInputOne } } } + //TODO: maybe this could be done in parallel for each unique library evaluator? for (int argHolderIdx = 0; argHolderIdx < argHolders.size(); argHolderIdx++) { Pair<Long, PythonLibraryEvaluator> fnEval = libraryEvaluators.get(argHolderIdx); @@ -236,15 +245,14 @@ public final class ExternalAssignBatchRuntimeFactory extends AbstractOneInputOne if (columnResult != null) { Pair<ByteBuffer, Counter> resultholder = batchResults.get(argHolderIdx); if (resultholder.getFirst().capacity() < columnResult.capacity()) { - resultholder.setFirst(ctx.allocateFrame(ExternalDataUtils.roundUpToNearestFrameSize( - columnResult.capacity(), ctx.getInitialFrameSize()))); + ByteBuffer realloc = ctx.reallocateFrame(resultholder.getFirst(), + columnResult.capacity() * 2, false); + resultholder.setFirst(realloc); } ByteBuffer resultBuf = resultholder.getFirst(); - resultBuf.clear(); - resultBuf.position(0); //offset 1 to skip message type - System.arraycopy(columnResult.array(), columnResult.arrayOffset() + 1, resultBuf.array(), - resultBuf.arrayOffset(), columnResult.capacity() - 1); + System.arraycopy(columnResult.array(), 1, resultBuf.array(), 0, + columnResult.remaining() - 1); //wrapper for results and warnings arrays. always length 2 consumeAndGetBatchLength(resultBuf); int numResults = (int) consumeAndGetBatchLength(resultBuf); @@ -260,37 +268,30 @@ public final class ExternalAssignBatchRuntimeFactory extends AbstractOneInputOne } } } + //decompose returned function columns into frame tuple format for (int i = 0; i < numTuples; i++) { tupleBuilder.reset(); for (int f = 0; f < projectionList.length; f++) { int k = projectionToOutColumns[f]; if (k >= 0) { - outputWrapper.clear(); - outputWrapper.position(0); + outputWrapper.reset(); Pair<ByteBuffer, Counter> result = batchResults.get(k); - if (result.getFirst() != null) { - if (result.getFirst().capacity() > outputWrapper.capacity()) { - outputWrapper = ctx.allocateFrame(ExternalDataUtils.roundUpToNearestFrameSize( - outputWrapper.capacity(), ctx.getInitialFrameSize())); - } - } - int start = outputWrapper.arrayOffset(); ATypeTag functionCalled = nullCalls[k][i]; if (functionCalled == ATypeTag.TYPE) { if (result.getSecond().get() > 0) { - MessageUnpackerToADM.unpack(result.getFirst(), outputWrapper, true); + unpackerToADM.unpack(result.getFirst(), outputWrapper.getDataOutput(), true); result.getSecond().set(result.getSecond().get() - 1); } else { //emit NULL for functions which failed with a warning - outputWrapper.put(ATypeTag.SERIALIZED_NULL_TYPE_TAG); + outputWrapper.getDataOutput().writeByte(ATypeTag.SERIALIZED_NULL_TYPE_TAG); } } else if (functionCalled == ATypeTag.NULL) { - outputWrapper.put(ATypeTag.SERIALIZED_NULL_TYPE_TAG); + outputWrapper.getDataOutput().writeByte(ATypeTag.SERIALIZED_NULL_TYPE_TAG); } else { - outputWrapper.put(ATypeTag.SERIALIZED_MISSING_TYPE_TAG); + outputWrapper.getDataOutput().writeByte(ATypeTag.SERIALIZED_MISSING_TYPE_TAG); } - tupleBuilder.addField(outputWrapper.array(), start, start + outputWrapper.position()); + tupleBuilder.addField(outputWrapper.getByteArray(), 0, outputWrapper.getLength()); } else { tupleBuilder.addField(tAccess, i, projectionList[f]); } diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/PrintTools.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/PrintTools.java index b337777..449ce13 100644 --- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/PrintTools.java +++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/printers/PrintTools.java @@ -18,6 +18,7 @@ */ package org.apache.asterix.dataflow.data.nontagged.printers; +import java.io.DataOutput; import java.io.IOException; import java.io.OutputStream; import java.io.PrintStream; @@ -285,12 +286,39 @@ public class PrintTools { if (c == '"') { os.write('"'); } - os.write(c); - position += sz; + if (Character.isHighSurrogate(c)) { + position += writeSupplementaryChar(os, b, maxPosition, position, c, sz); + continue; + } + while (sz > 0) { + os.write(b[position]); + ++position; + --sz; + } + break; } os.write('"'); } + public static void writeUTF8StringRaw(byte[] b, int s, int l, DataOutput os) throws IOException { + int utfLength = UTF8StringUtil.getUTFLength(b, s); + int position = s + UTF8StringUtil.getNumBytesToStoreLength(utfLength); // skip 2 bytes containing string size + int maxPosition = position + utfLength; + while (position < maxPosition) { + char c = UTF8StringUtil.charAt(b, position); + int sz = UTF8StringUtil.charSize(b, position); + if (Character.isHighSurrogate(c)) { + position += writeSupplementaryChar(os, b, maxPosition, position, c, sz); + continue; + } + while (sz > 0) { + os.write(b[position]); + ++position; + --sz; + } + } + } + public static void writeUTF8StringAsJSON(byte[] b, int s, int l, OutputStream os) throws IOException { int utfLength = UTF8StringUtil.getUTFLength(b, s); os.write('"'); @@ -420,4 +448,17 @@ public class PrintTools { return highSurrogateSize + lowSurrogateSize; } + //TODO: some way to dedupe this? + private static int writeSupplementaryChar(DataOutput os, byte[] src, int limit, int highSurrogatePos, + char highSurrogate, int highSurrogateSize) throws IOException { + final int lowSurrogatePos = highSurrogatePos + highSurrogateSize; + if (lowSurrogatePos >= limit) { + throw new IllegalStateException("malformed utf8 input"); + } + final char lowSurrogate = UTF8StringUtil.charAt(src, lowSurrogatePos); + final int lowSurrogateSize = UTF8StringUtil.charSize(src, lowSurrogatePos); + os.write(new String(new char[] { highSurrogate, lowSurrogate }).getBytes(StandardCharsets.UTF_8)); + return highSurrogateSize + lowSurrogateSize; + } + }