[RFC] Off-Process Expression Eval

[pedroerp]

Context

Many engines integrating with Velox have requirements to evaluate functions (UDFs) and expressions with some degree of isolation. The requirements range from:

• Programming language diversity - allow execution of say, Java or Python UDFs.
• Resource isolation - prevent UDF code from hogging resources from the main process.
• Release cycle - allow developers to decouple UDF release from main engine release.
• Security/Privacy - prevent malicious UDF code from inspecting other structures in the same memory space.
• Reliability - prevent UDF code from crashing the main process.

Proposal - OffProcessExpressionEvalOperator

Even though dynamic linking could be used to address some of these requirements, some form of off-process execution (either local/sidecar, or remote) could be supported in Velox to provide flexibility to engine developers.

The proposal is to add a new Velox operator to allow expressions to be executed off-process. This operator will consume input data, batch it, then serialize and communicate with a separate process for expression evaluation. Once the call is complete, data will be deserialized and sent downstream to the next operator. The separate process could be running in the same host (sidecar) or in a remote environment, and will communicate via a transport protocol such as thrift or protobuf - the actual transport will be built in a pluggable manner.

The pluggable transport will take a serialized payload and a serialized expression tree, and return another serialized payload with the results. The remote process can run anything, from C++ code also built using Velox, to logic in different programming languages using other frameworks.

The proposed operator will work in the following manner:

1. Consume and batch input data until a buffer size threshold is reached.
2. Then, serialize the batch using PrestoPage and make the transport call to a separate process.
   2.a. UDTs will be serialized according to the physical type they use underneath. If they map to an opaque physical type, a custom serde function needs to be provided.
3. Capture the future from the transport call, continue batching the data (step 1).
   3.a. There will be a parameter to limit how many requests we can have in-flight. If we hit that threshold, we stop consuming input data and just wait until any of the ongoing futures finish. This should provide us with control flow and prevent us from overloading the sidecar process.
4. Once a future is completed, deserialize the data and produce an output buffer.

Other considerations:

• The operator will also have some control logic to decide which input columns need to be serialized/deserialized, and which ones are pass-through.
• The operator should support an entire expression tree to be pushed down for execution in the remote process, though for the initial use cases a single function call (UDF) will be issued.
• We will also need to add support for expression tree serialization in Velox.
• The operator will be first added to velox/experimental until it’s feature-complete and stabilized.

Testing And Benchmarking

Other than regular unit tests, an end-to-end test suite using FuzzerConnector will be built. The test will generate random datasets and expressions and compare a regular expression eval with the new off-process path.

In a similar manner, benchmarks to evaluate the overhead incurred by the serialization should also be built.

Related Work

These are some systems that support UDFs. Java systems usually have different limitations considering the JVM makes the ABI compatibility issues somewhat easier to handle. So these are some of the C++ engine examples I could find:

SQL Server:

Supports UDFs through "language extensions" which are executed in a separate process via a "launchpad" - the component that controls the separate/sidecar process:

https://learn.microsoft.com/en-us/sql/language-extensions/concepts/extensibility-framework?view=sql-server-ver16

DB2:

Supports "external function" that can be written in multiple languages. I could not find the exact architecture, but the documentation suggests it runs on a separate process:

https://www.ibm.com/docs/en/db2-for-zos/11?topic=function-external-functions
https://www.ibm.com/docs/en/db2-for-zos/11?topic=programs-creating-user-defined-function

Vertica:

This one is interesting. It support external scripts (which are just spawned and executed externally), and C++ UDFs either in "fenced" or "unfenced" mode:

"User-Defined Extensions (UDxs) written in the C++ programming language have the option of running in unfenced mode, which means running directly within the Vertica process. Since they run within Vertica, unfenced UDxs have little overhead, and can perform almost as fast as Vertica's own built-in functions. However, since they run within Vertica directly, any bugs in their code (memory leaks, for example) can destabilize the main Vertica process that can bring one or more database nodes down.

You can instead opt to run most C++ UDxs in fenced mode, which runs the UDxs code outside of the main Vertica process in a separate zygote process. UDx code that crashes while running in fenced mode does not impact the core Vertica process. There is a small performance impact when running UDx code in fenced mode. On average, using fenced mode adds about 10% more time to execution compared to unfenced mode."

https://www.vertica.com/docs/9.2.x/HTML/Content/Authoring/ExtendingVertica/UDx/FencedMode/FencedMode.htm

Big Query:

Supports regular SQL and Javascript UDFs, but also remote UDF which are executed in Google's serverless architecture and communicate with the engine via HTTP:

https://cloud.google.com/bigquery/docs/reference/standard-sql/user-defined-functions
https://cloud.google.com/bigquery/docs/reference/standard-sql/remote-functions

This recent paper from Microsoft also contains a performance analysis of different UDF deployment modes:

https://www.microsoft.com/en-us/research/uploads/prod/2022/07/p2549-saur-final.pdf

Redshift

Supports Python UDFs and external AWS lambda-based UDFs:

"You can also create AWS Lambda UDFs that use custom functions defined in Lambda as part of your SQL queries. Lambda UDFs enable you to write complex UDFs and integrate with third-party components. "

https://docs.aws.amazon.com/redshift/latest/dg/user-defined-functions.html

Snowflake

Has external functions and remote UDFs . These are for scalar functions. The system architecture with the Proxy service and API integration seems quite nuanced:

https://docs.snowflake.com/en/sql-reference/external-functions-introduction

Postgres

Has very advanced extensibility options. There is support for scalars, aggregates, table functions, srfs any many options for their optimization also:

https://www.postgresql.org/docs/current/extend.html

--
Looking for feedback!

Cc: @mbasmanova @oerling @xiaoxmeng @bikramSingh91 @Yuhta @kagamiori @spershin @majetideepak @aditi-pandit @frankobe

[mbasmanova]

@pedroerp Pedro, it would be nice to add links to papers / documentation that explains how this problem is being solved in other systems.

[pedroerp]

@mbasmanova I added a "related work" section above with information about the C++ engines I could find. Let me know if you have other particular systems in mind and I can try to find more information.

[mbasmanova]

@pedroerp Thank you, Pedro. Perhaps, mention BigQuery as well: https://cloud.google.com/bigquery/docs/reference/standard-sql/user-defined-functions

[pedroerp]

Good callout. Just added quick info about BigQuery and Redshift. They both provide support for serverless-based remote UDF execution.

[aditi-pandit]

I found 2 references :

• Snowflake has external functions and remote UDFs https://docs.snowflake.com/en/sql-reference/external-functions-introduction. These are for scalar functions. The system architecture with the Proxy service and API integration seems quite nuanced.

• Postgres has very advanced extensibility options in https://www.postgresql.org/docs/current/extend.html. There is support for scalars, aggregates, table functions, srfs any many options for their optimization also.

Do you consider supporting these variations in the future ?

[pedroerp]

Thanks @aditi-pandit , I just added these two to the related work section.

To your question, aggregates and table functions should also be in scope and could be supported with a similar design, but it isn't something we'll be looking at yet since they are much fewer and users write them much less often.