Custom scalar logic¶

Part of Streaming Transformations. See all pages in this section.

A PIPE may call your own scalar user-defined functions, so per-row logic that would otherwise need a downstream job can run at ingest time. UDFs are permitted in both positions the pipe subquery offers:

In the SELECT list, to compute a derived column.
In the WHERE predicate, to filter on a computed value.

Supported handler languages are SQL and JavaScript. SECURE, IMMUTABLE, and MEMOIZABLE SQL UDFs are also supported; a memoizable UDF takes no arguments, so it can contribute a looked-up constant but can’t itself receive a streamed value.

Example — A derived column from a SQL UDF¶

Map a site code to its local timezone, and land the event time already converted:

CREATE OR REPLACE FUNCTION site_timezone(site STRING)
  RETURNS STRING
  IMMUTABLE
AS $$
  CASE site
    WHEN 'ber' THEN 'Europe/Berlin'
    WHEN 'sfo' THEN 'America/Los_Angeles'
    WHEN 'blr' THEN 'Asia/Kolkata'
    ELSE 'UTC'
  END
$$;

CREATE OR REPLACE PIPE events_pipe
  AS COPY INTO events (event_id, site, local_ts)
  FROM (
    SELECT
      s.$1:event_id::STRING,
      s.$1:site::STRING,
      CONVERT_TIMEZONE(site_timezone(s.$1:site::STRING), s.$1:ts::TIMESTAMP_NTZ)
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) s
  );

Example — Filtering on a JavaScript UDF result¶

A JavaScript UDF may appear in the WHERE predicate. Here only readings the function classifies as out of range are ingested:

CREATE OR REPLACE FUNCTION is_out_of_range(reading FLOAT)
  RETURNS BOOLEAN
  LANGUAGE JAVASCRIPT
  IMMUTABLE
AS $$
  return READING < -40 || READING > 85;
$$;

CREATE OR REPLACE PIPE alerts_pipe
  AS COPY INTO alerts (event_id, device_id, reading)
  FROM (
    SELECT
      s.$1:event_id::STRING,
      s.$1:device_id::STRING,
      s.$1:reading::FLOAT
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) s
    WHERE is_out_of_range(s.$1:reading::FLOAT)
  );

As with any WHERE clause in a pipe, records that do not satisfy the predicate are discarded silently and do not appear in load history.

Usage notes¶

Avoid long-running UDFs to minimize impact on stream throughput and latency.

Limitations¶

SQL and JavaScript scalar UDFs only; Python, Java, and Scala are currently unsupported as are all UDAFs, UDTFs, and external functions.