diff --git a/docs/reference/compute-engine/snowflake.md b/docs/reference/compute-engine/snowflake.md index e7b0dc5bd63..f6c633a4e40 100644 --- a/docs/reference/compute-engine/snowflake.md +++ b/docs/reference/compute-engine/snowflake.md @@ -24,5 +24,10 @@ batch_engine: role: sysadmin warehouse: demo_wh database: FEAST + python_udf_runtime_version: "3.10" ``` {% endcode %} + +## Configuration + +* `python_udf_runtime_version` *(optional, default: `"3.10"`)* -- The Snowflake Python UDF `RUNTIME_VERSION` used when Feast deploys its materialization UDFs. Snowflake periodically decommissions old Python UDF runtimes (for example, the 3.9 runtime was decommissioned, requiring Feast to bump its default to 3.10 -- see [#6606](https://github.com/feast-dev/feast/issues/6606)). If Snowflake decommissions the 3.10 runtime in the future, set this field to a still-supported version (e.g. `"3.11"`) instead of waiting for a new Feast release. diff --git a/sdk/python/feast/infra/compute_engines/snowflake/snowflake_engine.py b/sdk/python/feast/infra/compute_engines/snowflake/snowflake_engine.py index d0a1152eb55..ddcdad9578a 100644 --- a/sdk/python/feast/infra/compute_engines/snowflake/snowflake_engine.py +++ b/sdk/python/feast/infra/compute_engines/snowflake/snowflake_engine.py @@ -1,4 +1,5 @@ import os +import re import shutil from datetime import timezone from typing import Literal, Optional, Sequence, Union @@ -7,7 +8,7 @@ import pandas as pd import pyarrow as pa from colorama import Fore, Style -from pydantic import ConfigDict, Field, StrictStr +from pydantic import ConfigDict, Field, StrictStr, field_validator from tqdm import tqdm import feast @@ -87,8 +88,41 @@ class SnowflakeComputeEngineConfig(FeastConfigBaseModel): schema_: Optional[str] = Field("PUBLIC", alias="schema") """ Snowflake schema name """ + + python_udf_runtime_version: StrictStr = "3.10" + """ + Snowflake Python UDF RUNTIME_VERSION used when Feast deploys its materialization + UDFs (see `snowflake_python_udfs_creation.sql`). + + Snowflake periodically decommissions old Python UDF runtimes -- e.g. the 3.9 + runtime was decommissioned, which required Feast to bump its default runtime + from 3.9 to 3.10 (see https://github.com/feast-dev/feast/issues/6606). Rather + than hardcoding a version that will eventually go stale again, this field is + user-configurable so you are not blocked on a new Feast release the next time + Snowflake deprecates a runtime. + + Defaults to "3.10", matching Feast's own minimum supported Python version + (`requires-python` in `pyproject.toml`). If Snowflake decommissions the 3.10 + runtime in the future, override it in your `feature_store.yaml`, e.g.: + + batch_engine: + type: snowflake.engine + ... + python_udf_runtime_version: "3.11" + """ + model_config = ConfigDict(populate_by_name=True, extra="allow") + @field_validator("python_udf_runtime_version") + @classmethod + def validate_python_udf_runtime_version(cls, v: str) -> str: + if not re.fullmatch(r"\d+\.\d+(\.\d+)?", v): + raise ValueError( + "python_udf_runtime_version must be a valid Python version string " + f"such as '3.10' or '3.11.2', got {v!r}" + ) + return v + class SnowflakeComputeEngine(ComputeEngine): def get_historical_features( @@ -110,25 +144,18 @@ def update( entities_to_delete: Sequence[Entity], entities_to_keep: Sequence[Entity], ): - stage_context = f'"{self.repo_config.batch_engine.database}"."{self.repo_config.batch_engine.schema_}"' - stage_path = f'{stage_context}."feast_{project}"' + stage_path = f'"{self.repo_config.batch_engine.database}"."{self.repo_config.batch_engine.schema_}"."feast_{project}"' with GetSnowflakeConnection(self.repo_config.batch_engine) as conn: - query = f"SHOW USER FUNCTIONS LIKE 'FEAST_{project.upper()}%' IN SCHEMA {stage_context}" - cursor = execute_snowflake_statement(conn, query) - function_list = pd.DataFrame( - cursor.fetchall(), - columns=[column.name for column in cursor.description], - ) - - # if the SHOW FUNCTIONS query returns results, - # assumes that the materialization functions have been deployed - if len(function_list.index) > 0: - click.echo( - f"Materialization functions for {Style.BRIGHT + Fore.GREEN}{project}{Style.RESET_ALL} already detected." - ) - click.echo() - return None - + # Always (re)deploy the materialization functions, using + # `CREATE OR REPLACE FUNCTION` below, instead of skipping deployment + # when functions already exist. Previously, an early return here + # (triggered by a `SHOW USER FUNCTIONS` check) combined with + # `CREATE FUNCTION IF NOT EXISTS` in the SQL template meant that + # once UDFs were deployed for a project, they were never + # redeployed -- so a config change to `python_udf_runtime_version` + # (e.g. after Snowflake decommissions a runtime) would silently + # keep using the stale, already-deployed runtime version. See + # https://github.com/feast-dev/feast/pull/6608#discussion_r3602461959. click.echo( f"Deploying materialization functions for {Style.BRIGHT + Fore.GREEN}{project}{Style.RESET_ALL}" ) @@ -151,7 +178,11 @@ def update( sqlCommands = sqlFile.split(";") for command in sqlCommands: command = command.replace("STAGE_HOLDER", f"{stage_path}") - query = command.replace("PROJECT_NAME", f"{project}") + command = command.replace("PROJECT_NAME", f"{project}") + query = command.replace( + "RUNTIME_VERSION_HOLDER", + self.repo_config.batch_engine.python_udf_runtime_version, + ) execute_snowflake_statement(conn, query) return None diff --git a/sdk/python/feast/infra/utils/snowflake/snowpark/snowflake_python_udfs_creation.sql b/sdk/python/feast/infra/utils/snowflake/snowpark/snowflake_python_udfs_creation.sql index e39b12c1f79..b1a0d85f526 100644 --- a/sdk/python/feast/infra/utils/snowflake/snowpark/snowflake_python_udfs_creation.sql +++ b/sdk/python/feast/infra/utils/snowflake/snowpark/snowflake_python_udfs_creation.sql @@ -1,127 +1,127 @@ -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_binary_to_bytes_proto(df BINARY) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_binary_to_bytes_proto(df BINARY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_binary_to_bytes_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_varchar_to_string_proto(df VARCHAR) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_varchar_to_string_proto(df VARCHAR) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_varchar_to_string_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_array_bytes_to_list_bytes_proto(df ARRAY) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_array_bytes_to_list_bytes_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_bytes_to_list_bytes_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_array_varchar_to_list_string_proto(df ARRAY) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_array_varchar_to_list_string_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_varchar_to_list_string_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_array_number_to_list_int32_proto(df ARRAY) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_array_number_to_list_int32_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_number_to_list_int32_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_array_number_to_list_int64_proto(df ARRAY) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_array_number_to_list_int64_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_number_to_list_int64_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_array_float_to_list_double_proto(df ARRAY) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_array_float_to_list_double_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_float_to_list_double_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_array_boolean_to_list_bool_proto(df ARRAY) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_array_boolean_to_list_bool_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_boolean_to_list_bool_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_array_timestamp_to_list_unix_timestamp_proto(df ARRAY) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_array_timestamp_to_list_unix_timestamp_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_timestamp_to_list_unix_timestamp_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_number_to_int32_proto(df NUMBER) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_number_to_int32_proto(df NUMBER) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_number_to_int32_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_number_to_int64_proto(df NUMBER) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_number_to_int64_proto(df NUMBER) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_number_to_int64_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_float_to_double_proto(df DOUBLE) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_float_to_double_proto(df DOUBLE) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_float_to_double_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_boolean_to_bool_proto(df BOOLEAN) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_boolean_to_bool_proto(df BOOLEAN) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_boolean_to_bool_boolean_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_snowflake_timestamp_to_unix_timestamp_proto(df NUMBER) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_snowflake_timestamp_to_unix_timestamp_proto(df NUMBER) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_timestamp_to_unix_timestamp_proto' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_serialize_entity_keys(names ARRAY, data ARRAY, types ARRAY) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_serialize_entity_keys(names ARRAY, data ARRAY, types ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_serialize_entity_keys' IMPORTS = ('@STAGE_HOLDER/feast.zip'); -CREATE FUNCTION IF NOT EXISTS feast_PROJECT_NAME_entity_key_proto_to_string(names ARRAY, data ARRAY, types ARRAY) +CREATE OR REPLACE FUNCTION feast_PROJECT_NAME_entity_key_proto_to_string(names ARRAY, data ARRAY, types ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = 'RUNTIME_VERSION_HOLDER' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_entity_key_proto_to_string' IMPORTS = ('@STAGE_HOLDER/feast.zip') diff --git a/sdk/python/feast/infra/utils/snowflake/snowpark/snowflake_udfs.py b/sdk/python/feast/infra/utils/snowflake/snowpark/snowflake_udfs.py index 23026d79109..c9cbb87a953 100644 --- a/sdk/python/feast/infra/utils/snowflake/snowpark/snowflake_udfs.py +++ b/sdk/python/feast/infra/utils/snowflake/snowpark/snowflake_udfs.py @@ -14,11 +14,21 @@ ) from feast.value_type import ValueType +# NOTE: The `CREATE OR REPLACE FUNCTION ...` blocks below are illustrative reference +# documentation only; they are not executed as-is. The SQL Feast actually runs is +# templated from `snowflake_python_udfs_creation.sql`, which fills in placeholders +# (stage path, project name, and Python UDF RUNTIME_VERSION) at deploy time -- see +# `SnowflakeComputeEngine.update()` in `snowflake_engine.py`. The RUNTIME_VERSION +# shown here documents Feast's default (`SnowflakeComputeEngineConfig +# .python_udf_runtime_version`, currently "3.10"); if you override that config +# field, the runtime actually deployed to Snowflake will differ from what's shown +# below. + """ CREATE OR REPLACE FUNCTION feast_snowflake_binary_to_bytes_proto(df BINARY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_binary_to_bytes_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -43,7 +53,7 @@ def feast_snowflake_binary_to_bytes_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_varchar_to_string_proto(df VARCHAR) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_varchar_to_string_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -68,7 +78,7 @@ def feast_snowflake_varchar_to_string_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_array_bytes_to_list_bytes_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_bytes_to_list_bytes_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -96,7 +106,7 @@ def feast_snowflake_array_bytes_to_list_bytes_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_array_varchar_to_list_string_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_varchar_to_list_string_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -120,7 +130,7 @@ def feast_snowflake_array_varchar_to_list_string_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_array_number_to_list_int32_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_number_to_list_int32_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -144,7 +154,7 @@ def feast_snowflake_array_number_to_list_int32_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_array_number_to_list_int64_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_number_to_list_int64_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -168,7 +178,7 @@ def feast_snowflake_array_number_to_list_int64_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_array_float_to_list_double_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_float_to_list_double_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -194,7 +204,7 @@ def feast_snowflake_array_float_to_list_double_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_array_boolean_to_list_bool_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_boolean_to_list_bool_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -218,7 +228,7 @@ def feast_snowflake_array_boolean_to_list_bool_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_array_timestamp_to_list_unix_timestamp_proto(df ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_array_timestamp_to_list_unix_timestamp_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -244,7 +254,7 @@ def feast_snowflake_array_timestamp_to_list_unix_timestamp_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_number_to_int32_proto(df NUMBER) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_number_to_int32_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -269,7 +279,7 @@ def feast_snowflake_number_to_int32_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_number_to_int64_proto(df NUMBER) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_number_to_int64_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -296,7 +306,7 @@ def feast_snowflake_number_to_int64_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_float_to_double_proto(df DOUBLE) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_float_to_double_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -321,7 +331,7 @@ def feast_snowflake_float_to_double_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_boolean_to_bool_proto(df BOOLEAN) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_boolean_to_bool_boolean_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -346,7 +356,7 @@ def feast_snowflake_boolean_to_bool_boolean_proto(df): CREATE OR REPLACE FUNCTION feast_snowflake_timestamp_to_unix_timestamp_proto(df NUMBER) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_snowflake_timestamp_to_unix_timestamp_proto' IMPORTS = ('@feast_stage/feast.zip'); @@ -374,7 +384,7 @@ def feast_snowflake_timestamp_to_unix_timestamp_proto(df): CREATE OR REPLACE FUNCTION feast_serialize_entity_keys(names ARRAY, data ARRAY, types ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_serialize_entity_keys' IMPORTS = ('@feast_stage/feast.zip') @@ -423,7 +433,7 @@ def feast_serialize_entity_keys(df): CREATE OR REPLACE FUNCTION feast_entity_key_proto_to_string(names ARRAY, data ARRAY, types ARRAY) RETURNS BINARY LANGUAGE PYTHON - RUNTIME_VERSION = '3.9' + RUNTIME_VERSION = '3.10' PACKAGES = ('protobuf', 'pandas') HANDLER = 'feast.infra.utils.snowflake.snowpark.snowflake_udfs.feast_entity_key_proto_to_string' IMPORTS = ('@feast_stage/feast.zip') diff --git a/sdk/python/tests/unit/infra/compute_engines/snowflake/__init__.py b/sdk/python/tests/unit/infra/compute_engines/snowflake/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/sdk/python/tests/unit/infra/compute_engines/snowflake/test_snowflake_compute_engine.py b/sdk/python/tests/unit/infra/compute_engines/snowflake/test_snowflake_compute_engine.py new file mode 100644 index 00000000000..f450c124bed --- /dev/null +++ b/sdk/python/tests/unit/infra/compute_engines/snowflake/test_snowflake_compute_engine.py @@ -0,0 +1,117 @@ +"""Regression tests for SnowflakeComputeEngine.update(). + +See https://github.com/feast-dev/feast/pull/6608#discussion_r3602461959: a +maintainer pointed out that once materialization UDFs were deployed for a +project, a `SHOW USER FUNCTIONS` check made `update()` return early on every +subsequent call, and the SQL template used `CREATE FUNCTION IF NOT EXISTS`. +Together, this meant a user changing `python_udf_runtime_version` (e.g. after +Snowflake decommissions a runtime) would never actually get the new runtime +version deployed -- the stale, already-deployed UDFs were left untouched. + +The fix removes the early return and switches the SQL template to +`CREATE OR REPLACE FUNCTION`, so `update()` always (re)deploys the UDFs using +the currently configured runtime version. +""" + +import shutil +import tempfile +from unittest.mock import MagicMock, patch + +from feast.infra.compute_engines.snowflake.snowflake_engine import ( + SnowflakeComputeEngine, + SnowflakeComputeEngineConfig, +) + +MODULE = "feast.infra.compute_engines.snowflake.snowflake_engine" + + +def _make_engine(python_udf_runtime_version: str = "3.10") -> SnowflakeComputeEngine: + repo_config = MagicMock() + repo_config.offline_store.type = "snowflake.offline" + repo_config.batch_engine = SnowflakeComputeEngineConfig( + database="FEAST", + schema="PUBLIC", + python_udf_runtime_version=python_udf_runtime_version, + ) + + return SnowflakeComputeEngine( + repo_config=repo_config, + offline_store=MagicMock(), + online_store=MagicMock(), + ) + + +def _run_update(engine: SnowflakeComputeEngine, mock_execute: MagicMock) -> None: + mock_conn_cm = MagicMock() + mock_conn_cm.__enter__ = MagicMock(return_value=MagicMock()) + mock_conn_cm.__exit__ = MagicMock(return_value=False) + + def _fake_package_snowpark_zip(project_name): + copy_path = tempfile.mkdtemp() + return copy_path, f"{copy_path}/feast.zip" + + with ( + patch(f"{MODULE}.GetSnowflakeConnection", return_value=mock_conn_cm), + patch(f"{MODULE}.execute_snowflake_statement", mock_execute), + patch(f"{MODULE}.package_snowpark_zip", side_effect=_fake_package_snowpark_zip), + patch(f"{MODULE}.shutil.rmtree", side_effect=shutil.rmtree), + ): + engine.update( + project="myproject", + views_to_delete=[], + views_to_keep=[], + entities_to_delete=[], + entities_to_keep=[], + ) + + +def test_update_never_short_circuits_on_existing_functions(): + """update() must not query `SHOW USER FUNCTIONS` and skip deployment -- + that early-return path is what caused stale runtime versions to stick + around after a config change. + """ + engine = _make_engine() + mock_execute = MagicMock() + + _run_update(engine, mock_execute) + + executed_queries = [call.args[1] for call in mock_execute.call_args_list] + assert not any("SHOW USER FUNCTIONS" in q for q in executed_queries), ( + "update() should no longer perform a SHOW USER FUNCTIONS check that " + "causes it to skip (re)deploying the UDFs" + ) + assert any("CREATE OR REPLACE FUNCTION" in q for q in executed_queries), ( + "update() should deploy UDFs using CREATE OR REPLACE FUNCTION so " + "redeploys pick up a changed python_udf_runtime_version" + ) + + +def test_update_redeploys_with_current_runtime_version_on_every_call(): + """Calling update() twice (simulating two `feast apply` runs, e.g. before + and after a user bumps `python_udf_runtime_version`) must deploy the UDFs + both times, each time using whatever runtime version is currently + configured. + """ + engine = _make_engine(python_udf_runtime_version="3.11") + mock_execute = MagicMock() + + _run_update(engine, mock_execute) + first_call_count = mock_execute.call_count + assert first_call_count > 0 + + _run_update(engine, mock_execute) + second_call_count = mock_execute.call_count + assert second_call_count == 2 * first_call_count, ( + "the second update() call should re-run the full deployment (no " + "early return), just like the first" + ) + + executed_queries = [call.args[1] for call in mock_execute.call_args_list] + create_function_queries = [ + q for q in executed_queries if "CREATE OR REPLACE FUNCTION" in q + ] + assert create_function_queries, "expected CREATE OR REPLACE FUNCTION queries" + assert all("RUNTIME_VERSION = '3.11'" in q for q in create_function_queries), ( + "every deployed UDF should use the currently configured " + "python_udf_runtime_version ('3.11'), including on redeploy" + ) diff --git a/sdk/python/tests/unit/infra/utils/snowflake/test_snowflake_udfs.py b/sdk/python/tests/unit/infra/utils/snowflake/test_snowflake_udfs.py new file mode 100644 index 00000000000..5e8046bd83d --- /dev/null +++ b/sdk/python/tests/unit/infra/utils/snowflake/test_snowflake_udfs.py @@ -0,0 +1,182 @@ +import os +import re + +import pytest + +DECOMMISSIONED_RUNTIME_VERSION = "3.9" + +# Snowflake requires the Python UDF RUNTIME_VERSION to match a Python runtime +# that Snowflake still supports, and Feast should never request a runtime +# below the minimum Python version Feast itself supports +# (see `requires-python` in pyproject.toml). +MIN_SUPPORTED_RUNTIME_VERSION = "3.10" + +# Placeholder substituted into snowflake_python_udfs_creation.sql at deploy time by +# SnowflakeComputeEngine.update(), using +# SnowflakeComputeEngineConfig.python_udf_runtime_version. See +# https://github.com/feast-dev/feast/issues/6606 and the follow-up request to make +# the runtime version user-configurable on PR #6608. +RUNTIME_VERSION_PLACEHOLDER = "RUNTIME_VERSION_HOLDER" + + +def _snowpark_dir() -> str: + import feast + + return os.path.join( + feast.__path__[0], + "infra", + "utils", + "snowflake", + "snowpark", + ) + + +def _render_udf_creation_sql(runtime_version: str) -> str: + """Reproduce the substitution SnowflakeComputeEngine.update() performs on + snowflake_python_udfs_creation.sql, using a given runtime version, without + needing a live Snowflake connection. + """ + sql_path = os.path.join(_snowpark_dir(), "snowflake_python_udfs_creation.sql") + with open(sql_path) as f: + sql = f.read() + + sql = sql.replace("STAGE_HOLDER", "feast_stage") + sql = sql.replace("PROJECT_NAME", "myproject") + sql = sql.replace(RUNTIME_VERSION_PLACEHOLDER, runtime_version) + return sql + + +def test_udf_creation_sql_template_uses_runtime_version_placeholder(): + """The CREATE FUNCTION statements Feast executes against Snowflake must not + hardcode a specific (and eventually decommissioned) Python runtime version. + Instead, the SQL template should carry a placeholder that + SnowflakeComputeEngine.update() fills in from the user-configurable + `python_udf_runtime_version` setting (see + https://github.com/feast-dev/feast/issues/6606). + """ + sql_path = os.path.join(_snowpark_dir(), "snowflake_python_udfs_creation.sql") + with open(sql_path) as f: + sql = f.read() + + versions = re.findall(r"RUNTIME_VERSION\s*=\s*'([^']+)'", sql) + assert versions, ( + "expected to find RUNTIME_VERSION declarations in the UDF creation SQL" + ) + assert DECOMMISSIONED_RUNTIME_VERSION not in versions, ( + f"snowflake_python_udfs_creation.sql still requests the decommissioned " + f"Snowflake Python runtime {DECOMMISSIONED_RUNTIME_VERSION!r}" + ) + assert all(v == RUNTIME_VERSION_PLACEHOLDER for v in versions), ( + "expected all RUNTIME_VERSION declarations to use the " + f"{RUNTIME_VERSION_PLACEHOLDER!r} placeholder (filled in at deploy time " + f"from the configurable python_udf_runtime_version), got " + f"{sorted(set(versions))}" + ) + + +def test_udf_creation_sql_renders_default_runtime_version(): + """When rendered with the default `python_udf_runtime_version` + ("3.10"), the UDF creation SQL should request Feast's default runtime and + never the decommissioned 3.9 runtime. + """ + from feast.infra.compute_engines.snowflake.snowflake_engine import ( + SnowflakeComputeEngineConfig, + ) + + default_runtime_version = SnowflakeComputeEngineConfig( + database="FEAST" + ).python_udf_runtime_version + assert default_runtime_version == MIN_SUPPORTED_RUNTIME_VERSION + + rendered_sql = _render_udf_creation_sql(default_runtime_version) + versions = re.findall(r"RUNTIME_VERSION\s*=\s*'([^']+)'", rendered_sql) + assert versions, "expected to find rendered RUNTIME_VERSION declarations" + assert DECOMMISSIONED_RUNTIME_VERSION not in versions + assert all(v == MIN_SUPPORTED_RUNTIME_VERSION for v in versions) + + +def test_udf_creation_sql_renders_custom_runtime_version(): + """A custom `python_udf_runtime_version` should flow through to every + rendered RUNTIME_VERSION declaration, so users aren't blocked on a Feast + release the next time Snowflake decommissions a runtime. + """ + custom_runtime_version = "3.12" + rendered_sql = _render_udf_creation_sql(custom_runtime_version) + + versions = re.findall(r"RUNTIME_VERSION\s*=\s*'([^']+)'", rendered_sql) + assert versions, "expected to find rendered RUNTIME_VERSION declarations" + assert RUNTIME_VERSION_PLACEHOLDER not in versions + assert all(v == custom_runtime_version for v in versions), ( + f"expected all RUNTIME_VERSION declarations to be " + f"{custom_runtime_version!r}, got {sorted(set(versions))}" + ) + + +def test_udf_module_docstrings_do_not_reference_decommissioned_runtime_version(): + """The reference `CREATE OR REPLACE FUNCTION` docstrings embedded in + snowflake_udfs.py document the same UDFs and should stay consistent with + the SQL template that is actually executed. + + Note: we read the file as text rather than importing the module, since + `snowflake_udfs.py` imports `_snowflake`, which is only available inside + Snowflake's own Python UDF runtime. + """ + udfs_path = os.path.join(_snowpark_dir(), "snowflake_udfs.py") + with open(udfs_path) as f: + source = f.read() + + versions = re.findall(r"RUNTIME_VERSION\s*=\s*'([^']+)'", source) + assert versions, ( + "expected to find RUNTIME_VERSION declarations in snowflake_udfs.py" + ) + assert DECOMMISSIONED_RUNTIME_VERSION not in versions, ( + f"snowflake_udfs.py docstrings still reference the decommissioned " + f"Snowflake Python runtime {DECOMMISSIONED_RUNTIME_VERSION!r}" + ) + + +def test_python_udf_runtime_version_defaults_to_3_10(): + """The Snowflake compute engine config should default + `python_udf_runtime_version` to "3.10", matching Feast's own minimum + supported Python version, without requiring users to set it explicitly. + """ + from feast.infra.compute_engines.snowflake.snowflake_engine import ( + SnowflakeComputeEngineConfig, + ) + + config = SnowflakeComputeEngineConfig(database="FEAST") + assert config.python_udf_runtime_version == MIN_SUPPORTED_RUNTIME_VERSION + + +def test_python_udf_runtime_version_accepts_custom_value(): + """Per maintainer feedback on + https://github.com/feast-dev/feast/pull/6608 (@ntkathole: "is it possible + to make it user configurable?"), users must be able to override the + runtime version Feast requests, e.g. once Snowflake decommissions 3.10. + """ + from feast.infra.compute_engines.snowflake.snowflake_engine import ( + SnowflakeComputeEngineConfig, + ) + + config = SnowflakeComputeEngineConfig( + database="FEAST", python_udf_runtime_version="3.11" + ) + assert config.python_udf_runtime_version == "3.11" + + +@pytest.mark.parametrize("bad_version", ["", "three-ten", "3", "3.x", "3.10.", "v3.10"]) +def test_python_udf_runtime_version_rejects_invalid_values(bad_version): + """python_udf_runtime_version should validate that it looks like a Python + version string (e.g. "3.10" or "3.11.2"), so misconfiguration is caught + early rather than surfacing as an opaque Snowflake SQL compilation error. + """ + from pydantic import ValidationError + + from feast.infra.compute_engines.snowflake.snowflake_engine import ( + SnowflakeComputeEngineConfig, + ) + + with pytest.raises(ValidationError): + SnowflakeComputeEngineConfig( + database="FEAST", python_udf_runtime_version=bad_version + )