Batching of data sent from SingleStore to an external TVF

SingleStore always sends data in batches to an external TVF.

Batching of data sent from SingleStore to an external UDF

External UDFs are rewritten by SingleStore to external TVFs to allow external UDFs to be pushed down to a table scan to support batch processing of inputs.

However, not all external UDFs can be rewritten as TVFs. Only UDFs contained in the projection list (the columns that are selected by a query) or the WHERE clause of a SELECT can be rewritten as external TVFs. Furthermore, these external UDFs must not receive aggregate functions, window functions, or nested SELECT statements as input.

The value of the engine variable batch_external_function controls the rewrites of external UDFs to external TVFs for batching. The possible values are as follows:

• ALWAYS: Any query, which contains an external UDF that SingleStore cannot convert to an external TVF, will not be executed. External UDFs used in all other types of queries will run and will be rewritten to TVFs.

• AUTO: SingleStore rewrites all external UDFs that it can to external TVFs for batching. Any external UDFs that cannot be rewritten will be called one row at a time.

• NEVER: All external UDFs will be called without batching (i.e., one row at a time).

Send Delimited Batch Sets

Specify the USING DELIMITED BATCHES clause to keep the connection open after all the batches have been processed by SingleStore and accept a final batch from the external function. For example:

CREATE EXTERNAL FUNCTION extFunc(a TEXT) 
   RETURNS TEXT AS COLLOCATED SERVICE '/xfunc_pipe' 
   FORMAT ROWDAT_1 USING DELIMITED BATCHES;

By default, once SingleStore processes all the rows in the batch, it sends a final batch of zero size and closes the connection to the external function. However, the external function may sometimes send additional rows after all the rows have been processed. Enabling delimited batch sets allows the external function to acknowledge that all the rows have been received and send additional rows within the last response if required.

Benefits of Batching

• Batching reduces network and CPU utilization.

• Batching does increase memory utilization. However, the benefits of the decrease in network and CPU utilization far outweigh the increase in memory utilization.

• A global variable which assists with batching is external_functions_batch_size. It controls the batch size, in rows, for calls to the external process that services external functions. It has a default size of 512 rows. Most applications should simply use the default, but if the overhead of calling the external process becomes significant, you may wish to increase the batch size.

JSON

If batching is used, the data format is serialized as a JSON array of arrays as follows:

{"data":
    [[<row id>, <argument 1>, <argument 2>, …],
    [<row id>, <argument 1>, <argument 2>, …],
    ...
]}

If batching is not used, the data format is serialized as a JSON array of one array. The one array contains the row.

{"data":
    [[<row id>, <argument 1>, <argument 2>, …]]
}

Following is a description of the fields, in order, of the JSON sub-array:

• <row id>: A row id, which is an integer greater than or equal to 0. The row id is generated for the rows sent as input to the external function. The output rows are listed in ascending order. There can be repeating row ids.

• <argument n>: The serialization format supports three types:

  • integer

  • double

  • variable-length string

  All other types are serialized as strings.

ROWDAT_1

ROWDAT_1 is a SingleStore binary format. It is used for serializing data that is the input to and the output from an external function. The data is in a rowstore format which contains one row, or multiple rows if batching is used.

Following is a description of the fields, in order, in each row:

• A row id, which is an integer greater than or equal to 0. The row id in a given row must be larger than the row id in the previous row. Multiple rows may contain the same row id.

• The remaining fields contain the arguments to the external function. The serialization format supports three types:

  • 64-bit signed integer

  • 64-bit double (IEEE 754)

  • length-encoded variable-length string

  The serialization format is little-endian.  A single byte denoting whether a value is NULL (1: NULL, 0: not-NULL) precedes all type values (except the row id which cannot be NULL).

  All other types are serialized as string.

HTTP request

Example: Create a HTTP Link Using the CREDENTIALS and CONFIG Clauses

The following example shows how to create an HTTP link, which you can use in a CREATE [OR REPLACE] EXTERNAL FUNCTION statement.

CREATE LINK HTTP_link AS HTTP
CREDENTIALS '{"headers":{"authorization":"Bearer <token>"}}'
CONFIG '{"headers":{"version":"2.3"}}';

Example: Create an HTTP Link Using Only the CREDENTIALS Clause

The following example shows how to create an HTTP link, which you can use in a CREATE [OR REPLACE] EXTERNAL FUNCTION statement.

CREATE LINK HTTP_link AS HTTP
CREDENTIALS '{"headers":{"authorization":"Bearer <token>", "version":"2.3"}}';

Encrypting HTTP Connection Links

To create an encrypted HTTP connection (encrypts the traffic between the database and the remote endpoint), set the following security-specific CONFIG/CREDENTIALS JSON:

{
  "security": {
    "security.protocol": "ssl",
    "ssl.key.location": "/path/to/key.pem",
    "ssl.key.password": "<password_for_private_key>",
    "ssl.certificate.location": "/path/to/cert.pem"
    "ssl.ca.location": "/path/to/ca.pem"
  }
}

Where:

• security.protocol: must be "ssl". No other protocols are supported/implemented.

• ssl.key.location: required. Specifies the path to the private key that will be used during SSL negotiation with a remote server.

• ssl.key.password: optional. Used if ssl.key.location points to a password-protected private key.

• ssl.certificate.location: required. Specifies the path to the certificate with the signature of the public part of the key during the SSL handshake.

• ssl.ca.location: optional. If specified, the SingleStore Helios engine will verify the peer’s certificate against the CA bundle loaded from the provided path.

You can omit the headers object in the JSON when configuring encrypted connections.

-- Default Values
{
  "security": {
    "security.protocol": "ssl",
    "ssl.key.location": "/etc/memsql/ssl/server-key.pem",
    "ssl.key.password": "<password_for_private_key>",
    "ssl.certificate.location": "/etc/memsql/ssl/server-cert.pem"
    "ssl.ca.location": "/etc/memsql/extra/<.crt file>"
  }
}

-- Locations for certificates manually uploaded via the Cloud Portal

{
  "security": {
    "security.protocol": "ssl",
    "ssl.key.location": "/etc/memsql/extra/<.crt file>",
    "ssl.key.password": "<password_for_private_key>",
    "ssl.certificate.location": "/etc/memsql/extra/<.crt file>"
    "ssl.ca.location": "/etc/memsql/extra/<.crt file>"
  }
}

CREATE LINK link_1 AS HTTP 
CREDENTIALS '{"security":{"ssl.key.location":"<path_to_private_key>/server-key.pem"}}' 
CONFIG '{"security":{
  "security.protocol":"ssl", 
  "ssl.certificate.location":"<path_to_pem>/server-cert.pem", 
  "ssl.ca.location":"/<path_to_ca_bundle>/ca-cert.pem"}}';

CREATE OR REPLACE EXTERNAL FUNCTION 'ef1' (input INT) RETURNS TEXT AS REMOTE SERVICE 
  'https://localhost:8123/func1' FORMAT JSON LINK link_1;

SELECT ef1(42);

CREATE LINK link_2 AS HTTP 
CREDENTIALS '' 
CONFIG '{"security":{"ssl.ca.location":"<path_to_pe>/ca-cert.pem"}}';

CREATE LINK l_public AS HTTP
  credentials ''
  config '{
    "security":{"ssl.ca.location":"/etc/memsql/extra/ca-93057.crt"}
    }';

CREATE OR REPLACE EXTERNAL FUNCTION 'ef_public'(input int) RETURNS text AS REMOTE SERVICE 
    'https://example.com:8881/func1' FORMAT JSON LINK l_public;

CREATE LINK l_public_s2 AS HTTP
    credentials '{
      "security":{"ssl.key.location":"/etc/memsql/ssl/server-key.pem"}
      }'
    config '{"security":{
      "security.protocol":"ssl",
      "ssl.certificate.location":"/etc/memsql/ssl/server-cert.pem",
      "ssl.ca.location":"/etc/memsql/extra/ca-93057.crt"}
      }';
      
CREATE OR REPLACE EXTERNAL FUNCTION 'ef_public_s2'(input int) RETURNS text AS REMOTE SERVICE 
    'https://example.com:8882/func1' FORMAT JSON LINK l_public_s2;

-----BEGIN CERTIFICATE-----
MIIDODCCAiACAQEwDQYJKoZIhvcNAQELBQAwYDELMAkGA1UEBhMCVVMxCzAJBgN
[...]
vp9IHlbbNgAk4DpWvP8zYlAraVAClS0/sWAwhwkyU4OkB3VB8ti6KPBlc0V8ZoO
-----END CERTIFICATE-----


-----BEGIN RSA PRIVATE KEY-----
Proc-Type: 4,ENCRYPTED
DEK-Info: AES-128-CBC,3EB18A38696F0637E720801EA2A1FAEF

azQ5ApyXYoL4SsUFjPJEld4qKYl4p8rVKyiZ7RHihhmtct6jgFOD8h6R3GvifdD
[...]
peT4q+YCrys36Z7ZH/aYJbGzQ305XYvstlL5BK5pKVxTxRucgIGRMa56gbMGdCK
-----END RSA PRIVATE KEY-----

CREATE LINK l_private AS HTTP
    credentials '{"security":{
      "ssl.key.location":"/etc/memsql/extra/ca-12c5d.crt",
      "ssl.key.password":"muchsecret"}
      }'
    config '{"security":{
      "security.protocol":"ssl",
      "ssl.certificate.location":"/etc/memsql/extra/ca-12c5d.crt",
      "ssl.ca.location":"/etc/memsql/extra/ca-4860e.crt"}
      }';

CREATE OR REPLACE EXTERNAL FUNCTION 'ef_private'(input int) RETURNS text AS REMOTE SERVICE 
    'https://example.com:8883/func1' FORMAT JSON LINK l_private;

CREATE LINK l_private_le AS HTTP
    credentials '{"security":{
      "ssl.key.location":"/etc/memsql/extra/ca-12c5d.crt",
      "ssl.key.password":"muchsecret"}
      }'
    config '{"security":{
      "security.protocol":"ssl",
      "ssl.certificate.location":"/etc/memsql/extra/ca-12c5d.crt",
      "ssl.ca.location":"/etc/memsql/extra/ca-93057.crt"}
      }';

CREATE OR REPLACE EXTERNAL FUNCTION 'ef_private_le'(input int) RETURNS text AS REMOTE SERVICE 
    'https://example.com:8884/func1' FORMAT JSON LINK l_private_le;

Example: An External UDF

Suppose you are writing an external function that implements an elementary grammar checker. The checker scans one paragraph at a time. The checker corrects each paragraph by adding missing periods at the end of sentences and capitalizes the first letter of sentences (if the first letter is lowercase). Also, the checker optionally replaces digits, that are standalone words, in the sentence with the corresponding numbers spelled out. If no corrections to a paragraph are needed, the checker outputs the paragraph, unchanged.

Create the paragraphs table which stores the paragraphs to be checked and whether to check the digits in each paragraph.

CREATE TABLE paragraphs(paragraph TEXT, check_digits BOOL);

The table contains the following data.

+---------------------------------------------------------------+----------------+
| paragraph                                                     | check_digits   |
+---------------------------------------------------------------+----------------+
| this paragraph has 1 sentence                                 |              1 |
| this paragraph has 2 sentences. this is the second sentence   |              0 |
| This paragraph has one sentence and doesn't need corrections. |              1 |
+-------------------------------------------------------------+------------------+

Create the definition for an external function that checks a paragraph for grammar mistakes. If 1 is sent as input to the check_digits parameter, then digits are included as part of the grammar check.

CREATE EXTERNAL FUNCTION check_paragraph(paragraph TEXT, check_digits BOOL)
RETURNS TEXT
AS REMOTE SERVICE '<endpoint_URL>'
FORMAT JSON;

Implement the body of the external function and store the implementation on the HTTP server. An example implementation using an AWS Lambda function written in Python follows. Note that the implementation of the helper functions convert_numbers_to_words, capitalize_sentences and add_missing_periods_to_sentences is not shown.

import json

def lambda_handler(event, context):
    request = json.loads(event['body'])
    response = []
    for row_id, paragraph, check_digits in request["data"]:
        if check_digits:
            paragraph = convert_numbers_to_words(paragraph)
        paragraph = capitalize_sentences(paragraph)
        paragraph = add_missing_periods_to_sentences(paragraph)
        response.append([row_id, paragraph])
    return { 'statusCode': 200, 'body': json.dumps({"data": response}) }

Query the paragraphs table and call the external function as part of the query.

SELECT check_paragraph(paragraph, check_digits) as corrected_paragraph,
check_digits FROM paragraphs;

The external UDF is rewritten to an external TVF; the query calls the external function only once with the following input:

{"data":
    [[0, "this paragraph has 1 sentence", 1],
    [1, "this paragraph has 2 sentences. this is the second sentence", 0],
    [2, "This paragraph has one sentence and doesn't need corrections.", 1]]
}

After processing the input data, the external function returns the following data to the calling query. The data contains the corrected paragraphs. Paragraphs not needing any corrections are also included.

{"data":
    [[0, "This paragraph has one sentence."],
    [1, "This paragraph has 2 sentences. This is the second sentence."],
    [2, "This paragraph has one sentence and doesn't need corrections."]]
}

The query returns the following output.

+----------------------------------------------------------------+--------------+
| corrected_paragraph                                            | check_digits |
+----------------------------------------------------------------+--------------+
| This paragraph has one sentence.                               |            1 |
| This paragraph has 2 sentences. This is the second sentence.   |            0 |
| This paragraph has one sentence and doesn't need corrections.  |            1 |
+----------------------------------------------------------------+--------------+

CREATE EXTERNAL FUNCTION ext_func() 
RETURNS TEXT DEFINER ='root'@'%' 
AS REMOTE SERVICE  '...' FORMAT JSON;

SHOW FUNCTIONS;

+------------------------+------------------+----------+-------------+-----------------+------+
| Functions_in_func_proc | Function Type    | Definer  | Data Format | Runtime Type    | Link |
+------------------------+------------------+----------+-------------+-----------------+------+
| ext_func               | External User    | root@%   | JSON        | Remote Service  |      |
|                        | Defined Function |          |             |                 |      |
+------------------------+------------------+----------+-------------+-----------------+------+

Example: An External TVF

Suppose you want to read the contents of an external file and store each line in a row in a table.

Create the definition for the external function, which gets the contents of a file that is passed as input.

CREATE EXTERNAL FUNCTION get_file_lines(filename TEXT)
RETURNS TABLE(line TEXT)
AS REMOTE SERVICE '<endpoint_URL>'
FORMAT JSON;

Implement the body of the external function and store the implementation on the HTTP server. The implementation is not shown here.

Query the external function:

SELECT line FROM get_file_lines("simple_file.txt");

When the external function runs, it returns a JSON array of arrays, which contains the data in simple_file.txt:

{"data":
    [[0, "This is the first sentence."],
    [0, "This is the second sentence."],
    [0, "This is the third sentence."]]
}

When the query that calls the external function completes, the query returns the following output.

+------------------------------+
| line                         |
+------------------------------+
| This is the first sentence.  |
| This is the second sentence. |
| This is the third sentence.  |
+------------------------------+

Example: Using Remote Service

The default service type is REMOTE and the endpoint is expected to be the URL of a remote HTTP service.  The following example uses an AWS endpoint as the remote service:

CREATE OR REPLACE EXTERNAL FUNCTION uppercase_string(string_col TEXT)
     RETURNS TEXT    	
     AS REMOTE SERVICE 	
     'https://fnc.execute-api.us-east-1.amazonaws.com/prod/upcase_string' 
     FORMAT JSON;

Example: Using localhost as Remote Service

CREATE EXTERNAL FUNCTION ext_udf2(col_a INT, col_b INT)
     RETURNS TABLE(col_1 TEXT, col_2 BOOL)
     AS REMOTE SERVICE 'http://localhost:3305/cool_table_func'
     FORMAT JSON;

Using Collocated Service

When the COLLOCATED option is specified, commands are sent via Unix domain socket while input/output is sent via shared memory files.

For each collocated external function, the engine will first connect to the Unix domain socket listened to by the collocated service.  First, the engine will send the version of the protocol (64-bit little-endian int, the current version is 1). Then the length-embedded function name (the length is also 64-bit little-endian int). The engine will create two in-memory files via memfd_create, both sealed with F_SEAL_SHRINK  so the files can only grow in size but not shrink and thus is safe to mmap. The input/output files use sendmsg to transmit to the Unix domain socket as ancillary data.

For each input batch, the input and output will be written to their corresponding shared files using either ROWDAT_1 or JSON format (similar to remote service). Each new batch is going to overwrite the data from the previous batch. The engine will send 64-bit int as the length of the input to the Unix domain socket. The collocated service will send two 64-bit int as the status code and output length to the Unix domain socket. The status code follows HTTP and 200 indicates success. Any code other than 200 indicates an error.

CREATE EXTERNAL FUNCTION lowercase_string(string_col TEXT)
     RETURNS TEXT
     AS COLLOCATED SERVICE '/usr/bin/fd'
     FORMAT ROWDAT_1;