SNOW-213702, SNOW-2014127: Add support for DataFrame/Series.create_or_replace_dynamic_table (#3209)

sfc-gh-helmeleegy · web-flow · commit 2c31fd690cfc · 2025-04-01T17:30:48.000-07:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -53,6 +53,7 @@
 - Added support for `pd.Grouper` objects in group by operations. When `freq` is specified, the default values of the `sort`, `closed`, `label`, and `convention` arguments are supported; `origin` is supported when it is `start` or `start_day`.
 - Added support for relaxed consistency and ordering guarantees in `pd.read_snowflake` for both named data sources (e.g., tables and views) and query data sources by introducing the new parameter `relaxed_ordering`.
 - Added support for `DataFrame.create_or_replace_view` and `Series.create_or_replace_view`.
+- Added support for `DataFrame.create_or_replace_dynamic_table` and `Series.create_or_replace_dynamic_table`.
 
 #### Improvements
 
diff --git a/docs/source/modin/dataframe.rst b/docs/source/modin/dataframe.rst
@@ -39,6 +39,7 @@ DataFrame
     DataFrame.to_snowpark
     DataFrame.cache_result
     DataFrame.create_or_replace_view
+    DataFrame.create_or_replace_dynamic_table
 
 .. rubric:: Conversion
 
diff --git a/docs/source/modin/series.rst b/docs/source/modin/series.rst
@@ -45,6 +45,7 @@ Series
     Series.to_snowpark
     Series.cache_result
     Series.create_or_replace_view
+    Series.create_or_replace_dynamic_table
 
 .. rubric:: Conversion
 
diff --git a/src/snowflake/snowpark/modin/plugin/extensions/dataframe_extensions.py b/src/snowflake/snowpark/modin/plugin/extensions/dataframe_extensions.py
@@ -15,6 +15,7 @@
 from modin.pandas.api.extensions import register_dataframe_accessor
 from pandas._typing import IndexLabel
 
+from snowflake.snowpark._internal.type_utils import ColumnOrName
 from snowflake.snowpark.dataframe import DataFrame as SnowparkDataFrame
 from snowflake.snowpark.modin.plugin.extensions.utils import add_cache_result_docstring
 from snowflake.snowpark.modin.plugin.utils.warning_message import (
@@ -284,3 +285,87 @@ def create_or_replace_view(
         comment=comment,
         statement_params=statement_params,
     )
+
+
+@register_dataframe_accessor("create_or_replace_dynamic_table")
+def create_or_replace_dynamic_table(
+    self,
+    name: Union[str, Iterable[str]],
+    *,
+    warehouse: str,
+    lag: str,
+    comment: Optional[str] = None,
+    mode: str = "overwrite",
+    refresh_mode: Optional[str] = None,
+    initialize: Optional[str] = None,
+    clustering_keys: Optional[Iterable[ColumnOrName]] = None,
+    is_transient: bool = False,
+    data_retention_time: Optional[int] = None,
+    max_data_extension_time: Optional[int] = None,
+    statement_params: Optional[Dict[str, str]] = None,
+    iceberg_config: Optional[dict] = None,
+    _emit_ast: bool = True,
+) -> List[Row]:
+    """
+    Creates a dynamic table that captures the computation expressed by this DataFrame.
+
+    For ``name``, you can include the database and schema name (i.e. specify a
+    fully-qualified name). If no database name or schema name are specified, the
+    dynamic table will be created in the current database or schema.
+
+    ``name`` must be a valid `Snowflake identifier <https://docs.snowflake.com/en/sql-reference/identifiers-syntax.html>`_.
+
+    Args:
+        name: The name of the dynamic table to create or replace. Can be a list of strings
+            that specifies the database name, schema name, and view name.
+        warehouse: The name of the warehouse used to refresh the dynamic table.
+        lag: specifies the target data freshness
+        comment: Adds a comment for the created table. See
+            `COMMENT <https://docs.snowflake.com/en/sql-reference/sql/comment>`_.
+        mode: Specifies the behavior of create dynamic table. Allowed values are:
+            - "overwrite" (default): Overwrite the table by dropping the old table.
+            - "errorifexists": Throw and exception if the table already exists.
+            - "ignore": Ignore the operation if table already exists.
+        refresh_mode: Specifies the refresh mode of the dynamic table. The value can be "AUTO",
+            "FULL", or "INCREMENTAL".
+        initialize: Specifies the behavior of initial refresh. The value can be "ON_CREATE" or
+            "ON_SCHEDULE".
+        clustering_keys: Specifies one or more columns or column expressions in the table as the clustering key.
+            See `Clustering Keys & Clustered Tables <https://docs.snowflake.com/en/user-guide/tables-clustering-keys>`_
+            for more details.
+        is_transient: A boolean value that specifies whether the dynamic table is transient.
+        data_retention_time: Specifies the retention period for the dynamic table in days so that
+            Time Travel actions can be performed on historical data in the dynamic table.
+        max_data_extension_time: Specifies the maximum number of days for which Snowflake can extend
+            the data retention period of the dynamic table to prevent streams on the dynamic table
+            from becoming stale.
+        statement_params: Dictionary of statement level parameters to be set while executing this action.
+        iceberg_config: A dictionary that can contain the following iceberg configuration values:
+
+            - external_volume: specifies the identifier for the external volume where
+                the Iceberg table stores its metadata files and data in Parquet format.
+            - catalog: specifies either Snowflake or a catalog integration to use for this table.
+            - base_location: the base directory that snowflake can write iceberg metadata and files to.
+            - catalog_sync: optionally sets the catalog integration configured for Polaris Catalog.
+            - storage_serialization_policy: specifies the storage serialization policy for the table.
+
+
+    Note:
+        See `understanding dynamic table refresh <https://docs.snowflake.com/en/user-guide/dynamic-tables-refresh>`_.
+        for more details on refresh mode.
+    """
+    return self.to_snowpark().create_or_replace_dynamic_table(
+        name=name,
+        warehouse=warehouse,
+        lag=lag,
+        comment=comment,
+        mode=mode,
+        refresh_mode=refresh_mode,
+        initialize=initialize,
+        clustering_keys=clustering_keys,
+        is_transient=is_transient,
+        data_retention_time=data_retention_time,
+        max_data_extension_time=max_data_extension_time,
+        statement_params=statement_params,
+        iceberg_config=iceberg_config,
+    )
diff --git a/src/snowflake/snowpark/modin/plugin/extensions/series_extensions.py b/src/snowflake/snowpark/modin/plugin/extensions/series_extensions.py
@@ -15,6 +15,7 @@
 from modin.pandas.api.extensions import register_series_accessor
 from pandas._typing import Axis, IndexLabel
 
+from snowflake.snowpark._internal.type_utils import ColumnOrName
 from snowflake.snowpark.dataframe import DataFrame as SnowparkDataFrame
 from snowflake.snowpark.modin.plugin.extensions.utils import add_cache_result_docstring
 from snowflake.snowpark.modin.plugin.utils.warning_message import (
@@ -274,3 +275,87 @@ def create_or_replace_view(
         comment=comment,
         statement_params=statement_params,
     )
+
+
+@register_series_accessor("create_or_replace_dynamic_table")
+def create_or_replace_dynamic_table(
+    self,
+    name: Union[str, Iterable[str]],
+    *,
+    warehouse: str,
+    lag: str,
+    comment: Optional[str] = None,
+    mode: str = "overwrite",
+    refresh_mode: Optional[str] = None,
+    initialize: Optional[str] = None,
+    clustering_keys: Optional[Iterable[ColumnOrName]] = None,
+    is_transient: bool = False,
+    data_retention_time: Optional[int] = None,
+    max_data_extension_time: Optional[int] = None,
+    statement_params: Optional[Dict[str, str]] = None,
+    iceberg_config: Optional[dict] = None,
+    _emit_ast: bool = True,
+) -> List[Row]:
+    """
+    Creates a dynamic table that captures the computation expressed by this Series.
+
+    For ``name``, you can include the database and schema name (i.e. specify a
+    fully-qualified name). If no database name or schema name are specified, the
+    dynamic table will be created in the current database or schema.
+
+    ``name`` must be a valid `Snowflake identifier <https://docs.snowflake.com/en/sql-reference/identifiers-syntax.html>`_.
+
+    Args:
+        name: The name of the dynamic table to create or replace. Can be a list of strings
+            that specifies the database name, schema name, and view name.
+        warehouse: The name of the warehouse used to refresh the dynamic table.
+        lag: specifies the target data freshness
+        comment: Adds a comment for the created table. See
+            `COMMENT <https://docs.snowflake.com/en/sql-reference/sql/comment>`_.
+        mode: Specifies the behavior of create dynamic table. Allowed values are:
+            - "overwrite" (default): Overwrite the table by dropping the old table.
+            - "errorifexists": Throw and exception if the table already exists.
+            - "ignore": Ignore the operation if table already exists.
+        refresh_mode: Specifies the refresh mode of the dynamic table. The value can be "AUTO",
+            "FULL", or "INCREMENTAL".
+        initialize: Specifies the behavior of initial refresh. The value can be "ON_CREATE" or
+            "ON_SCHEDULE".
+        clustering_keys: Specifies one or more columns or column expressions in the table as the clustering key.
+            See `Clustering Keys & Clustered Tables <https://docs.snowflake.com/en/user-guide/tables-clustering-keys>`_
+            for more details.
+        is_transient: A boolean value that specifies whether the dynamic table is transient.
+        data_retention_time: Specifies the retention period for the dynamic table in days so that
+            Time Travel actions can be performed on historical data in the dynamic table.
+        max_data_extension_time: Specifies the maximum number of days for which Snowflake can extend
+            the data retention period of the dynamic table to prevent streams on the dynamic table
+            from becoming stale.
+        statement_params: Dictionary of statement level parameters to be set while executing this action.
+        iceberg_config: A dictionary that can contain the following iceberg configuration values:
+
+            - external_volume: specifies the identifier for the external volume where
+                the Iceberg table stores its metadata files and data in Parquet format.
+            - catalog: specifies either Snowflake or a catalog integration to use for this table.
+            - base_location: the base directory that snowflake can write iceberg metadata and files to.
+            - catalog_sync: optionally sets the catalog integration configured for Polaris Catalog.
+            - storage_serialization_policy: specifies the storage serialization policy for the table.
+
+
+    Note:
+        See `understanding dynamic table refresh <https://docs.snowflake.com/en/user-guide/dynamic-tables-refresh>`_.
+        for more details on refresh mode.
+    """
+    return self.to_snowpark().create_or_replace_dynamic_table(
+        name=name,
+        warehouse=warehouse,
+        lag=lag,
+        comment=comment,
+        mode=mode,
+        refresh_mode=refresh_mode,
+        initialize=initialize,
+        clustering_keys=clustering_keys,
+        is_transient=is_transient,
+        data_retention_time=data_retention_time,
+        max_data_extension_time=max_data_extension_time,
+        statement_params=statement_params,
+        iceberg_config=iceberg_config,
+    )
diff --git a/tests/integ/modin/frame/test_create_or_replace_dynamic_table.py b/tests/integ/modin/frame/test_create_or_replace_dynamic_table.py
@@ -0,0 +1,129 @@
+#
+# Copyright (c) 2012-2025 Snowflake Computing Inc. All rights reserved.
+#
+
+import modin.pandas as pd
+import pytest
+
+import snowflake.snowpark.modin.plugin  # noqa: F401
+from snowflake.snowpark._internal.utils import TempObjectType
+from snowflake.snowpark.exceptions import SnowparkSQLException
+from snowflake.snowpark.session import Session
+from tests.integ.modin.utils import BASIC_TYPE_DATA1, BASIC_TYPE_DATA2
+from tests.integ.utils.sql_counter import sql_count_checker
+from tests.utils import Utils
+
+
+@sql_count_checker(query_count=5)
+def test_create_or_replace_dynamic_table_no_relaxed_ordering_raises(session) -> None:
+    try:
+        # create table
+        table_name = Utils.random_table_name()
+        session.create_dataframe(
+            [BASIC_TYPE_DATA1, BASIC_TYPE_DATA2]
+        ).write.save_as_table(table_name)
+
+        # create dataframe with relaxed_ordering disabled
+        snow_dataframe = pd.read_snowflake(
+            f"(((SELECT * FROM {table_name})))", relaxed_ordering=False
+        )
+
+        # creating dynamic_table fails when relaxed_ordering is disabled
+        # because it cannot depend on a temp table
+        dynamic_table_name = Utils.random_name_for_temp_object(
+            TempObjectType.DYNAMIC_TABLE
+        )
+        with pytest.raises(
+            SnowparkSQLException,
+            match="Dynamic Tables cannot depend on a temporary object",
+        ):
+            snow_dataframe.create_or_replace_dynamic_table(
+                name=dynamic_table_name,
+                warehouse=session.get_current_warehouse(),
+                lag="1000 minutes",
+            )
+    finally:
+        # cleanup
+        Utils.drop_dynamic_table(session, dynamic_table_name)
+        Utils.drop_table(session, table_name)
+
+
+@sql_count_checker(query_count=5)
+def test_create_or_replace_dynamic_table_relaxed_ordering(session) -> None:
+    try:
+        # create table
+        table_name = Utils.random_name_for_temp_object(TempObjectType.DYNAMIC_TABLE)
+        session.create_dataframe(
+            [BASIC_TYPE_DATA1, BASIC_TYPE_DATA2]
+        ).write.save_as_table(table_name)
+
+        # create dataframe with relaxed_ordering enabled
+        snow_dataframe = pd.read_snowflake(
+            f"(((SELECT * FROM {table_name})))", relaxed_ordering=True
+        )
+
+        # creating dynamic_table succeeds when relaxed_ordering is enabled
+        dynamic_table_name = Utils.random_name_for_temp_object(
+            TempObjectType.DYNAMIC_TABLE
+        )
+        assert (
+            "successfully created"
+            in snow_dataframe.create_or_replace_dynamic_table(
+                name=dynamic_table_name,
+                warehouse=session.get_current_warehouse(),
+                lag="1000 minutes",
+            )[0]["status"]
+        )
+
+        # accessing the created dynamic_table in the same session also succeeds
+        res = session.sql(f"select * from {dynamic_table_name}").collect()
+        assert len(res) == 2
+    finally:
+        # cleanup
+        Utils.drop_dynamic_table(session, dynamic_table_name)
+        Utils.drop_table(session, table_name)
+
+
+@sql_count_checker(query_count=4)
+def test_create_or_replace_dynamic_table_multiple_sessions_relaxed_ordering(
+    session,
+    db_parameters,
+) -> None:
+    try:
+        # create table
+        table_name = Utils.random_name_for_temp_object(TempObjectType.DYNAMIC_TABLE)
+        session.create_dataframe(
+            [BASIC_TYPE_DATA1, BASIC_TYPE_DATA2]
+        ).write.save_as_table(table_name)
+
+        # create dataframe with relaxed_ordering enabled
+        snow_dataframe = pd.read_snowflake(
+            f"(((SELECT * FROM {table_name})))", relaxed_ordering=True
+        )
+
+        # creating dynamic_table succeeds when relaxed_ordering is enabled
+        dynamic_table_name = Utils.random_name_for_temp_object(
+            TempObjectType.DYNAMIC_TABLE
+        )
+        assert (
+            "successfully created"
+            in snow_dataframe.create_or_replace_dynamic_table(
+                name=dynamic_table_name,
+                warehouse=session.get_current_warehouse(),
+                lag="1000 minutes",
+            )[0]["status"]
+        )
+
+        # another session
+        new_session = Session.builder.configs(db_parameters).create()
+        pd.session = new_session
+
+        # accessing the created dynamic_table in another session also succeeds
+        res = new_session.sql(f"select * from {dynamic_table_name}").collect()
+        assert len(res) == 2
+        new_session.close()
+    finally:
+        # cleanup
+        Utils.drop_dynamic_table(session, dynamic_table_name)
+        Utils.drop_table(session, table_name)
+        pd.session = session
diff --git a/tests/integ/modin/series/test_create_or_replace_dynamic_table.py b/tests/integ/modin/series/test_create_or_replace_dynamic_table.py
