SNOW-3473261: Add Iceberg branch / tag time travel via version_ref

Add ``version_ref`` time-travel parameter that emits Snowflake's
``AT(VERSION_REF => '<name>')`` clause — the unified branch / tag
syntax that subsumes Spark Iceberg's ``VERSION AS OF '<x>'``
(branch + tag) and the 4-part ``branch_<x>`` / ``tag_<x>``
identifier forms.

Mirrors the merged ``version=`` (snapshot-id) PR (#4231):
* Hidden behind ``**kwargs`` on ``Session.table()``,
  ``DataFrameReader.table()``, and ``Table.__init__()`` — consumed
  by Snowpark Connect, not part of the advertised public surface
  yet.
* ``DataFrameReader`` reader option aliases ``version_ref`` and
  ``version-ref`` (matching the ``snapshot-id`` / ``snapshot_id``
  style); both auto-set ``time_travel_mode='at'`` since branch /
  tag reads are positional, not range-of-time.
* Validation: ``time_travel_mode='before'`` is rejected;
  ``version_ref`` is mutually exclusive with the other time-travel
  parameters (``statement``, ``offset``, ``timestamp``, ``stream``,
  ``version``); non-string and empty refs are rejected.
* No AST proto changes — the field travels through ``**kwargs``
  the same way ``version=`` does.
* No CHANGELOG entry — internal kwargs surface, not customer-
  visible (matches PR #4231).

Unit tests cover SQL emission, validation, conflicts, and
``_extract_time_travel_from_options`` extraction for both alias
forms. Two integ tests under ``tests/integ/test_dataframe.py`` are
added but ``pytest.mark.skip``'d pending a CI account with a
CLD-linked unmanaged Iceberg table that has named branches / tags
and ``FEATURE_ICEBERG_TIME_TRAVEL`` enabled (manual reproducer
lives in the snowflake-eng/sas repo).

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: sfc-gh-igarish

src/snowflake/snowpark/_internal/utils.py

@@ -1956,6 +1956,7 @@ class TimeTravelConfig(NamedTuple):
     timestamp_type: Optional[str] = None
     stream: Optional[str] = None
     version: Optional[int] = None
+    version_ref: Optional[str] = None
 
     @staticmethod
     def validate_and_normalize_params(
@@ -1966,6 +1967,7 @@ def validate_and_normalize_params(
         timestamp_type: Optional[Union[str, "TimestampTimeZone"]] = None,
         stream: Optional[str] = None,
         version: Optional[int] = None,
+        version_ref: Optional[str] = None,
     ) -> Optional["TimeTravelConfig"]:
         """
         Validates and normalizes time travel parameters.
@@ -1988,7 +1990,8 @@ def validate_and_normalize_params(
             ValueError: If parameters are invalid.
         """
         time_travel_arg_count = sum(
-            arg is not None for arg in (statement, offset, timestamp, stream, version)
+            arg is not None
+            for arg in (statement, offset, timestamp, stream, version, version_ref)
         )
 
         # Validate mode
@@ -2023,10 +2026,34 @@ def validate_and_normalize_params(
                 f"'version' must be an int Iceberg snapshot id, got {type(version).__name__}."
             )
 
+        # version_ref (Iceberg branch / tag name, unified under Snowflake's
+        # ``AT(VERSION_REF => '<name>')`` grammar) only works with 'at' mode —
+        # Iceberg branch / tag reads are positional, not range-of-time, so
+        # ``BEFORE`` has no meaning.
+        if version_ref is not None and time_travel_mode.lower() != "at":
+            raise ValueError(
+                "Iceberg version_ref time travel can only be used with "
+                "time_travel_mode='at', not 'before'."
+            )
+
+        # Validate version_ref type — Iceberg ref names are strings (branch /
+        # tag identifiers). Empty strings are invalid.
+        if version_ref is not None:
+            if not isinstance(version_ref, str):
+                raise ValueError(
+                    f"'version_ref' must be a string Iceberg branch / tag name, "
+                    f"got {type(version_ref).__name__}."
+                )
+            if not version_ref:
+                raise ValueError(
+                    "'version_ref' must be a non-empty Iceberg branch / tag name."
+                )
+
         # Validate exactly one parameter is provided
         if time_travel_arg_count != 1:
             raise ValueError(
-                "Exactly one of 'statement', 'offset', 'timestamp', 'stream', or 'version' must be provided."
+                "Exactly one of 'statement', 'offset', 'timestamp', 'stream', "
+                "'version', or 'version_ref' must be provided."
             )
 
         # Normalize timestamp
@@ -2061,6 +2088,7 @@ def validate_and_normalize_params(
             timestamp_type=timestamp_type,
             stream=stream,
             version=version,
+            version_ref=version_ref,
         )
 
     def generate_sql_clause(self) -> str:
@@ -2069,8 +2097,10 @@ def generate_sql_clause(self) -> str:
         Args:
             config: Time travel configuration.
         Returns:
-            SQL clause like " AT (TIMESTAMP => TO_TIMESTAMP_NTZ('...'))" or
-            " AT (VERSION => 1234567890)" for Iceberg snapshot id time travel.
+            SQL clause like " AT (TIMESTAMP => TO_TIMESTAMP_NTZ('...'))",
+            " AT (VERSION => 1234567890)" for Iceberg snapshot id time travel,
+            or " AT (VERSION_REF => 'release_v1')" for Iceberg branch / tag
+            time travel.
         """
         clause = f" {self.time_travel_mode.upper()} "
 
@@ -2082,6 +2112,8 @@ def generate_sql_clause(self) -> str:
             clause += f"(STREAM => '{self.stream}')"
         elif self.version is not None:
             clause += f"(VERSION => {self.version})"
+        elif self.version_ref is not None:
+            clause += f"(VERSION_REF => '{self.version_ref}')"
         elif self.timestamp is not None:
             if self.timestamp_type is not None:
                 timestamp_type = self.timestamp_type.upper()

src/snowflake/snowpark/dataframe_reader.py

@@ -161,6 +161,14 @@ def _extract_time_travel_from_options(options: dict) -> dict:
     - Automatically set time_travel_mode to 'at'
       (Iceberg snapshot ids only support ``AT(VERSION => N)``, not ``BEFORE``)
     - Cannot be used with time_travel_mode='before' (raises error)
+
+    Special handling for 'VERSION_REF' / 'VERSION-REF' (Iceberg branch / tag
+    name) — both aliases map to the internal ``version_ref`` time travel
+    parameter and emit ``AT(VERSION_REF => '<name>')`` on the Snowflake side
+    (unified branch / tag clause, see Spark Iceberg's ``VERSION AS OF '<x>'``):
+    - Automatically set time_travel_mode to 'at'
+      (branch / tag reads are positional, not range-of-time)
+    - Cannot be used with time_travel_mode='before' (raises error)
     """
     result = {}
     excluded_keys = set()
@@ -211,6 +219,27 @@ def _extract_time_travel_from_options(options: dict) -> dict:
             )
         result["time_travel_mode"] = "at"
 
+    # Handle Iceberg branch / tag ref (``version_ref`` / ``version-ref``).
+    # Both aliases route to the internal ``version_ref`` parameter and emit
+    # ``AT(VERSION_REF => '<name>')`` server-side. Auto-sets mode='at'.
+    version_ref_value = options.get("VERSION_REF")
+    version_ref_source = "version_ref"
+    if version_ref_value is None:
+        version_ref_value = options.get("VERSION-REF")
+        version_ref_source = "version-ref"
+    if version_ref_value is not None:
+        if (
+            "TIME_TRAVEL_MODE" in options
+            and options["TIME_TRAVEL_MODE"].lower() == "before"
+        ):
+            raise ValueError(
+                f"Cannot use '{version_ref_source}' option with "
+                "time_travel_mode='before'. Iceberg branch / tag time travel "
+                "only supports time_travel_mode='at'."
+            )
+        result["version_ref"] = str(version_ref_value)
+        result["time_travel_mode"] = "at"
+
     for option_key, param_name in _TIME_TRAVEL_OPTIONS_PARAMS_MAP.items():
         if option_key in options and option_key not in excluded_keys:
             result[param_name] = options[option_key]
@@ -634,11 +663,13 @@ def table(
             ...     .option("offset", -60)                 # This will be IGNORED
             ...     .table("my_table", time_travel_mode="at", offset=-3600))  # Only this is used
         """
-        # ``version`` (Iceberg snapshot id) is intentionally not in the public
-        # signature — it's consumed by Snowpark Connect and may be removed
-        # once a first-class API lands. Accept it through **kwargs so direct
-        # callers can still pass it without us advertising it.
+        # ``version`` (Iceberg snapshot id) and ``version_ref`` (Iceberg
+        # branch / tag name) are intentionally not in the public signature —
+        # they are consumed by Snowpark Connect and may be removed once a
+        # first-class API lands. Accept them through **kwargs so direct
+        # callers can still pass them without us advertising the surface.
         version = kwargs.pop("version", None)
+        version_ref = kwargs.pop("version_ref", None)
         if kwargs:
             raise TypeError(
                 f"table() got unexpected keyword arguments: {sorted(kwargs)}"
@@ -664,13 +695,20 @@ def table(
             if stream is not None:
                 ast.stream.value = stream
 
-        if time_travel_mode is not None or version is not None:
-            # If version is provided without mode, default to 'at' (snapshot ids
-            # only make sense with AT — symmetric with iceberg_tag handling).
+        if (
+            time_travel_mode is not None
+            or version is not None
+            or version_ref is not None
+        ):
+            # If version / version_ref is provided without mode, default to
+            # 'at' — snapshot ids and branch / tag reads only make sense with
+            # AT (symmetric with the as-of-timestamp option handling).
             effective_mode = (
                 time_travel_mode
                 if time_travel_mode
-                else ("at" if version is not None else None)
+                else (
+                    "at" if (version is not None or version_ref is not None) else None
+                )
             )
             time_travel_params = {
                 "time_travel_mode": effective_mode,
@@ -680,6 +718,7 @@ def table(
                 "timestamp_type": timestamp_type,
                 "stream": stream,
                 "version": version,
+                "version_ref": version_ref,
             }
         else:
             # if time_travel_mode is not provided, extract time travel config from options

src/snowflake/snowpark/session.py

@@ -2773,11 +2773,13 @@ def table(
             # timestamp_type remains "NTZ" (user's explicit choice respected)
             >>> table2 = session.read.table("my_table", time_travel_mode="at", timestamp=tz_aware, timestamp_type="NTZ")  # doctest: +SKIP
         """
-        # ``version`` (Iceberg snapshot id) is intentionally not in the public
-        # signature — it's consumed by Snowpark Connect and may be removed
-        # once a first-class API lands. Accept it through **kwargs so direct
-        # callers can still pass it without us advertising it.
+        # ``version`` (Iceberg snapshot id) and ``version_ref`` (Iceberg
+        # branch / tag name) are intentionally not in the public signature —
+        # they are consumed by Snowpark Connect and may be removed once a
+        # first-class API lands. Accept them through **kwargs so direct
+        # callers can still pass them without us advertising the surface.
         version = kwargs.pop("version", None)
+        version_ref = kwargs.pop("version_ref", None)
         if kwargs:
             raise TypeError(
                 f"table() got unexpected keyword arguments: {sorted(kwargs)}"
@@ -2820,6 +2822,7 @@ def table(
             timestamp_type=timestamp_type,
             stream=stream,
             version=version,
+            version_ref=version_ref,
         )
         # Replace API call origin for table
         set_api_call_source(t, "Session.table")

src/snowflake/snowpark/table.py

@@ -296,11 +296,13 @@ def __init__(
         stream: Optional[str] = None,
         **kwargs,
     ) -> None:
-        # ``version`` (Iceberg snapshot id) is intentionally not in the public
-        # signature — it's consumed by Snowpark Connect and may be removed
-        # once a first-class API lands. Accept it through **kwargs so direct
-        # callers can still pass it without us advertising it.
+        # ``version`` (Iceberg snapshot id) and ``version_ref`` (Iceberg
+        # branch / tag name) are intentionally not in the public signature —
+        # they are consumed by Snowpark Connect and may be removed once a
+        # first-class API lands. Accept them through **kwargs so direct
+        # callers can still pass them without us advertising the surface.
         version = kwargs.pop("version", None)
+        version_ref = kwargs.pop("version_ref", None)
         if kwargs:
             raise TypeError(
                 f"Table() got unexpected keyword arguments: {sorted(kwargs)}"
@@ -333,6 +335,7 @@ def __init__(
             timestamp_type=timestamp_type,
             stream=stream,
             version=version,
+            version_ref=version_ref,
         )
 
         snowflake_table_plan = SnowflakeTable(

tests/integ/test_dataframe.py

@@ -8423,3 +8423,72 @@ def test_iceberg_snapshot_id_time_travel_dataframe_reader_option(session):
         session.read.option("snapshot-id", snapshot_id).table(table_fqn).collect()
     )
     assert via_kwarg == via_option
+
+
+# ----------------------------------------------------------------------
+# Iceberg branch / tag (``version_ref=``) time travel.
+#
+# TODO(SNOW-3473261): Wire these up to a CI test account that has:
+#   * a Catalog-Linked Database (CLD) such as cldUnity / cldglue, AND
+#   * an unmanaged Iceberg table inside it that exposes a named branch or
+#     tag (e.g. created via ``ALTER TABLE ... CREATE TAG <name>`` on the
+#     OSS Iceberg side, or via the catalog's branch / tag API).
+#
+# Snowflake's ``AT(VERSION_REF => '<name>')`` syntax is the unified
+# branch / tag clause that subsumes Spark Iceberg's
+# ``VERSION AS OF '<x>'`` (which accepts both branch and tag names) and
+# the 4-part ``prod.db.t.`branch_<x>`` / ``tag_<x>`` identifier forms.
+# Like the snapshot-id surface, it currently requires
+# ``FEATURE_ICEBERG_TIME_TRAVEL`` on the account and is scoped to
+# unmanaged Iceberg tables in CLDs, so these tests are skipped by default
+# and exercised manually against ``sfctest0`` (see
+# ``tests/sas_tests/test_iceberg_version_ref_sample.py`` in
+# snowflake-eng/sas for the manual reproducer).
+# ----------------------------------------------------------------------
+@pytest.mark.skip(
+    reason=(
+        "Requires a CLD-linked unmanaged Iceberg table with at least one "
+        "named branch / tag and FEATURE_ICEBERG_TIME_TRAVEL enabled on the "
+        "account. Tested manually; see TODO above."
+    )
+)
+def test_iceberg_version_ref_time_travel_session_table(session):
+    """End-to-end: ``Session.table(..., version_ref='<name>')`` returns the
+    table state at the requested Iceberg branch / tag ref."""
+    table_fqn = "CLDUNITY.scosschema.snapshot_demo"
+    # Demo table is set up with a tag ``first_load`` pointing at the
+    # earliest snapshot (see sas-side reproducer for the setup script).
+    tag_name = "first_load"
+
+    tagged = session.table(
+        table_fqn, time_travel_mode="at", version_ref=tag_name
+    ).collect()
+    latest = session.table(table_fqn).collect()
+    # Tag reads use the snapshot's schema as it existed at the tag —
+    # row count at an earlier tag should be ≤ current row count, since the
+    # tag is bound to a specific (earlier) snapshot.
+    assert len(tagged) <= len(latest)
+
+
+@pytest.mark.skip(
+    reason=(
+        "Requires a CLD-linked unmanaged Iceberg table with at least one "
+        "named branch / tag and FEATURE_ICEBERG_TIME_TRAVEL enabled on the "
+        "account. Tested manually; see TODO above."
+    )
+)
+def test_iceberg_version_ref_time_travel_dataframe_reader_option(session):
+    """End-to-end: ``session.read.option('version_ref', 'name').table(...)``
+    routes through the Iceberg-compat option alias and produces the same
+    result as the explicit ``version_ref=`` kwarg."""
+    table_fqn = "CLDUNITY.scosschema.snapshot_demo"
+    tag_name = "first_load"
+
+    via_kwarg = session.read.table(
+        table_fqn, time_travel_mode="at", version_ref=tag_name
+    ).collect()
+    via_option = session.read.option("version_ref", tag_name).table(table_fqn).collect()
+    via_hyphen_option = (
+        session.read.option("version-ref", tag_name).table(table_fqn).collect()
+    )
+    assert via_kwarg == via_option == via_hyphen_option