UDTFs, UDAFs and Snowpark Python UDFs in Snowflake

1.User-defined table functions (UDTFs) in SQL

A user-defined table function (UDTF) returns a set of rows: zero, one or many, each with one or more columns. You declare the output shape with RETURNS TABLE(...), listing name-and-type pairs. At least one output column is required. In a SQL UDTF, the body must be a SELECT expression, and its columns must match the declared types position by position. A UDTF can take at most 500 input arguments and return at most 500 output columns.

create or replace function orders_for_product(PROD_ID varchar)
    returns table (Product_ID varchar, Quantity_Sold numeric(11, 2))
    as
    $$
        select product_ID, quantity_sold
            from orders
            where product_ID = PROD_ID
    $$
    ;

select product_id, quantity_sold
    from table(orders_for_product('compostable bags'))
    order by product_id;

You don't call a UDTF in the SELECT list as you would a scalar function. You call it in the FROM clause, with its name and arguments inside TABLE( ... ). Its output columns can then be used anywhere a normal table column can.

UDTFs can also take part in joins. A column from the other table can be passed in as the argument, so the function runs once for each row. In the example below, each year from favorite_years produces that year's favorite colors. If you need extra join conditions, put them in a WHERE clause, not in ON.

select *
    from favorite_years y join table(favorite_colors(y.year)) c
    order by year, color;

select product_id, quantity_sold
    from  ? (orders_for_product('compostable bags'))
    order by product_id;

Sources1

2.User-defined aggregate functions (UDAFs) in Python

A user-defined aggregate function (UDAF) takes one or more rows as input and produces a single row of output. Creating one takes two steps. First, write a handler class. Then register it, either with the Snowpark API or with SQL's CREATE FUNCTION. The same class works for both.

Snowflake calls the handler's methods at runtime. All five are required.

  class PythonSumUDAF:
    def __init__(self):
      # This aggregate state is a primitive Python data type.
      self._partial_sum = 0

    @property
    def aggregate_state(self):
      return self._partial_sum

    def accumulate(self, input_value):
      self._partial_sum += input_value

    def merge(self, other_partial_sum):
      self._partial_sum += other_partial_sum

    def finish(self):
      return self._partial_sum
  sum_udaf = udaf(PythonSumUDAF, name="sum_int", replace=True, return_type=IntegerType(), input_types=[IntegerType()])

Like a scalar UDF, a UDAF can be registered as a named or an anonymous function. There are three ways to register it:

- the udaf function, used directly or as a @udaf decorator on the class - UDAFRegistration.register - register_from_file, which points to a Python file or zip file

After registration, you can call the UDAF from SQL or from a DataFrame, for example df.agg(sum_udaf("a")).

Sources2

3.Snowpark UDFs in Python: anonymous, named, temporary and permanent

With the Snowpark API, you can turn a Python lambda or function into a UDF and use it on DataFrames. Snowpark uploads your function's code to an internal stage. When you call the UDF, it runs on the server where the data is. A UDF created with SQL's CREATE FUNCTION can also be called from Snowpark.

There are two kinds of registration:

- An anonymous UDF is assigned to a variable. You can use it for as long as that variable is in scope. - A named UDF is registered with a name argument. Use this when you need to call it by name, or reuse it in a later session.

Calling register or udf creates a temporary UDF by default, which lasts only for the current session. To make it permanent, set is_permanent=True and give a stage_location where the code and its dependencies are uploaded. In code that may run in several sessions, use the register method rather than udf, so that the default Session object can always be found.

@udf(name="minus_one",  ? =True, stage_location="@my_stage", replace=True)
def minus_one(x: int) -> int:
  return x-1

Dependencies have to be declared:

- Your own modules, zip files, resource files or directories are added with session.add_import(). Snowpark uploads them to an internal stage and imports them when the UDF runs. The Snowpark library itself is not uploaded automatically. Python's built-in libraries are already available on the server. - Third-party packages come from the Snowflake Anaconda channel. Add them for the whole session with session.add_packages or session.add_requirements, or for one UDF with @udf(packages=[...]). UDF-level packages override session-level ones.

Pinning versions matters for production. A permanent UDF resolves its dependencies once, when it is registered. A temporary UDF should list exact versions, or a newer package version may be picked up later.

Sources3

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.A UDTF is called in the SELECT list like a scalar function, and extra join conditions go in an ON clause.Why is that wrong?

   A UDTF is called in the FROM clause, inside TABLE( ... ). When it is joined to another table, extra conditions go in a WHERE clause, not ON.

   Covered in User-defined table functions (UDTFs) in SQL

2. 2.Giving a Snowpark UDF a name makes it permanent.Why is that wrong?

   Named and anonymous UDFs are both temporary by default. A permanent UDF needs is_permanent=True and a stage_location.

   Covered in Snowpark UDFs in Python: anonymous, named, temporary and permanent