Mark streaming mode as unstable

pola-rs · Jan 24, 2024 · 384ca90 · 384ca90
1 parent b03b5b7
commit 384ca90
Show file tree

Hide file tree

Showing 2 changed files with 75 additions and 14 deletions.
diff --git a/py-polars/polars/functions/lazy.py b/py-polars/polars/functions/lazy.py
@@ -17,7 +17,7 @@
     deprecate_renamed_function,
     issue_deprecation_warning,
 )
-from polars.utils.unstable import unstable
+from polars.utils.unstable import issue_unstable_warning, unstable
 
 with contextlib.suppress(ImportError):  # Module not available when building docs
     import polars.polars as plr
@@ -1636,7 +1636,17 @@ def collect_all(
     comm_subexpr_elim
         Common subexpressions will be cached and reused.
     streaming
-        Run parts of the query in a streaming fashion (this is in an alpha state)
+        Process the query in batches to handle larger-than-memory data.
+        If set to `False` (default), the entire query is processed in a single
+        batch.
+
+        .. warning::
+            Streaming mode is considered **unstable**. It may be changed
+            at any point without it being considered a breaking change.
+
+        .. note::
+            Use :func:`explain` to see if Polars can process the query in streaming
+            mode.
 
     Returns
     -------
@@ -1650,6 +1660,10 @@ def collect_all(
         comm_subplan_elim = False
         comm_subexpr_elim = False
 
+    if streaming:
+        issue_unstable_warning("Streaming mode is considered unstable.")
+        comm_subplan_elim = False
+
     prepared = []
 
     for lf in lazy_frames:
@@ -1762,7 +1776,17 @@ def collect_all_async(
     comm_subexpr_elim
         Common subexpressions will be cached and reused.
     streaming
-        Run parts of the query in a streaming fashion (this is in an alpha state)
+        Process the query in batches to handle larger-than-memory data.
+        If set to `False` (default), the entire query is processed in a single
+        batch.
+
+        .. warning::
+            Streaming mode is considered **unstable**. It may be changed
+            at any point without it being considered a breaking change.
+
+        .. note::
+            Use :func:`explain` to see if Polars can process the query in streaming
+            mode.
 
     See Also
     --------
@@ -1788,6 +1812,10 @@ def collect_all_async(
         comm_subplan_elim = False
         comm_subexpr_elim = False
 
+    if streaming:
+        issue_unstable_warning("Streaming mode is considered unstable.")
+        comm_subplan_elim = False
+
     prepared = []
 
     for lf in lazy_frames:

diff --git a/py-polars/polars/lazyframe/frame.py b/py-polars/polars/lazyframe/frame.py
@@ -70,7 +70,7 @@
     deprecate_saturating,
     issue_deprecation_warning,
 )
-from polars.utils.unstable import unstable
+from polars.utils.unstable import issue_unstable_warning, unstable
 from polars.utils.various import (
     _in_notebook,
     _prepare_row_index_args,
@@ -1646,7 +1646,8 @@ def collect(
             batch.
 
             .. warning::
-                This functionality is currently in an alpha state.
+                Streaming mode is considered **unstable**. It may be changed
+                at any point without it being considered a breaking change.
 
             .. note::
                 Use :func:`explain` to see if Polars can process the query in streaming
@@ -1712,6 +1713,7 @@ def collect(
             comm_subexpr_elim = False
 
         if streaming:
+            issue_unstable_warning("Streaming mode is considered unstable.")
             comm_subplan_elim = False
 
         ldf = self._ldf.optimization_toggle(
@@ -1813,17 +1815,17 @@ def collect_async(
         comm_subexpr_elim
             Common subexpressions will be cached and reused.
         streaming
-            Run parts of the query in a streaming fashion (this is in an alpha state)
+            Process the query in batches to handle larger-than-memory data.
+            If set to `False` (default), the entire query is processed in a single
+            batch.
 
-        Notes
-        -----
-        In case of error `set_exception` is used on
-        `asyncio.Future`/`gevent.event.AsyncResult` and will be reraised by them.
+            .. warning::
+                Streaming mode is considered **unstable**. It may be changed
+                at any point without it being considered a breaking change.
 
-        See Also
-        --------
-        polars.collect_all : Collect multiple LazyFrames at the same time.
-        polars.collect_all_async: Collect multiple LazyFrames at the same time lazily.
+            .. note::
+                Use :func:`explain` to see if Polars can process the query in streaming
+                mode.
 
         Returns
         -------
@@ -1832,6 +1834,16 @@ def collect_async(
         If `gevent=True` then returns wrapper that has
         `.get(block=True, timeout=None)` method.
 
+        See Also
+        --------
+        polars.collect_all : Collect multiple LazyFrames at the same time.
+        polars.collect_all_async: Collect multiple LazyFrames at the same time lazily.
+
+        Notes
+        -----
+        In case of error `set_exception` is used on
+        `asyncio.Future`/`gevent.event.AsyncResult` and will be reraised by them.
+
         Examples
         --------
         >>> import asyncio
@@ -1868,6 +1880,7 @@ def collect_async(
             comm_subexpr_elim = False
 
         if streaming:
+            issue_unstable_warning("Streaming mode is considered unstable.")
             comm_subplan_elim = False
 
         ldf = self._ldf.optimization_toggle(
@@ -1886,6 +1899,7 @@ def collect_async(
         ldf.collect_with_callback(result._callback)  # type: ignore[attr-defined]
         return result  # type: ignore[return-value]
 
+    @unstable()
     def sink_parquet(
         self,
         path: str | Path,
@@ -1906,6 +1920,10 @@ def sink_parquet(
         """
         Evaluate the query in streaming mode and write to a Parquet file.
 
+        .. warning::
+            Streaming mode is considered **unstable**. It may be changed
+            at any point without it being considered a breaking change.
+
         This allows streaming results that are larger than RAM to be written to disk.
 
         Parameters
@@ -1978,6 +1996,7 @@ def sink_parquet(
             maintain_order=maintain_order,
         )
 
+    @unstable()
     def sink_ipc(
         self,
         path: str | Path,
@@ -1994,6 +2013,10 @@ def sink_ipc(
         """
         Evaluate the query in streaming mode and write to an IPC file.
 
+        .. warning::
+            Streaming mode is considered **unstable**. It may be changed
+            at any point without it being considered a breaking change.
+
         This allows streaming results that are larger than RAM to be written to disk.
 
         Parameters
@@ -2045,6 +2068,7 @@ def sink_ipc(
 
     @deprecate_renamed_parameter("quote", "quote_char", version="0.19.8")
     @deprecate_renamed_parameter("has_header", "include_header", version="0.19.13")
+    @unstable()
     def sink_csv(
         self,
         path: str | Path,
@@ -2072,6 +2096,10 @@ def sink_csv(
         """
         Evaluate the query in streaming mode and write to a CSV file.
 
+        .. warning::
+            Streaming mode is considered **unstable**. It may be changed
+            at any point without it being considered a breaking change.
+
         This allows streaming results that are larger than RAM to be written to disk.
 
         Parameters
@@ -2182,6 +2210,7 @@ def sink_csv(
             maintain_order=maintain_order,
         )
 
+    @unstable()
     def sink_ndjson(
         self,
         path: str | Path,
@@ -2197,6 +2226,10 @@ def sink_ndjson(
         """
         Evaluate the query in streaming mode and write to an NDJSON file.
 
+        .. warning::
+            Streaming mode is considered **unstable**. It may be changed
+            at any point without it being considered a breaking change.
+
         This allows streaming results that are larger than RAM to be written to disk.
 
         Parameters