documentation fixes and write mode removal

Author: jcatt-sf

CHANGELOG.md

@@ -9,16 +9,16 @@
   New methods let an entry point process a Data Lake Object's Change Data Feed continuously instead of reading a bounded snapshot:
 
   - `read_dlo_deltas(name)` / `read_dmo_deltas(name)` – return a streaming DataFrame over the object's change feed.
-  - `write_dlo_deltas(name, dataframe, write_mode)` – start a streaming query that writes each micro-batch to the target DLO and return the `StreamingQuery` handle.
+  - `write_dlo_deltas(name, dataframe)` – start a streaming query that writes each micro-batch to the target DLO and return the `StreamingQuery` handle.
 
   ```python
   deltas = client.read_dlo_deltas("Input__dll")
   transformed = deltas.withColumn("description__c", upper(col("description__c")))
-  query = client.write_dlo_deltas("Output__dll", transformed, WriteMode.APPEND)
+  query = client.write_dlo_deltas("Output__dll", transformed)
   query.awaitTermination()
   ```
 
-  Supported streaming write modes are `WriteMode.APPEND`, `WriteMode.OVERWRITE`, and `WriteMode.MERGE_UPSERT_DELETE`. These methods run only inside the Data Cloud streaming (`DELTA_SYNC`) runtime; locally they raise `NotImplementedError`. See the `examples/streaming_deltas/entrypoint.py` example and the "Streaming (delta) transforms" section of the README.
+  These methods run only inside the Data Cloud streaming (`DELTA_SYNC`) runtime; locally they raise `NotImplementedError`. See the `examples/streaming_deltas/entrypoint.py` example and the "Streaming (delta) transforms" section of the README.
 
 ## 6.0.0
 

README.md

@@ -157,7 +157,7 @@ You should only need the following methods:
 For streaming (delta) transforms, the streaming counterparts are:
 * `read_dlo_deltas(name)` – Read the streaming change feed (deltas) of a Data Lake Object as a streaming DataFrame
 * `read_dmo_deltas(name)` – Read the streaming change feed (deltas) of a Data Model Object as a streaming DataFrame
-* `write_dlo_deltas(name, spark_dataframe, write_mode)` – Write a streaming DataFrame of deltas to a Data Lake Object; returns the started `StreamingQuery`
+* `write_dlo_deltas(name, spark_dataframe)` – Write a streaming DataFrame of deltas to a Data Lake Object; returns the started `StreamingQuery`
 
 For example:
 ```python
@@ -182,28 +182,24 @@ Streaming BYOC transforms process a Data Lake Object's Change Data Feed continuo
 from pyspark.sql.functions import col, upper
 
 from datacustomcode import Client
-from datacustomcode.io.writer.base import WriteMode
 
 client = Client()
 
 # read_dlo_deltas returns a *streaming* DataFrame over the change feed.
 deltas = client.read_dlo_deltas("Input__dll")
 
-# Ordinary PySpark transform. Keep the change-feed metadata columns
-# (those starting with "_") — the streaming sink needs them to apply
-# inserts, updates, and deletes to the target DLO.
+# Ordinary PySpark transform.
 transformed = deltas.withColumn("description__c", upper(col("description__c")))
 
 # write_dlo_deltas starts a streaming query and returns the StreamingQuery.
-# The runtime owns the trigger and checkpoint location; you choose only the
-# target table and write mode.
-query = client.write_dlo_deltas("Output__dll", transformed, WriteMode.APPEND)
+# The runtime owns the trigger and checkpoint location; you
+# choose only the target table.
+query = client.write_dlo_deltas("Output__dll", transformed)
 query.awaitTermination()
 ```
 
 Notes:
 
-- Supported streaming write modes are `WriteMode.APPEND`, `WriteMode.OVERWRITE`, and `WriteMode.MERGE_UPSERT_DELETE`.
 - These methods only run inside the Data Cloud streaming (`DELTA_SYNC`) runtime. Locally (`datacustomcode run`) they raise `NotImplementedError`, since there is no change feed to stream.
 - A complete runnable entry point is provided in [`examples/streaming_deltas/entrypoint.py`](src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py).
 

src/datacustomcode/client.py

@@ -396,7 +396,7 @@ def write_to_dmo(
         return self._writer.write_to_dmo(name, dataframe, write_mode, **kwargs)  # type: ignore[no-any-return]
 
     def write_dlo_deltas(
-        self, name: str, dataframe: PySparkDataFrame, write_mode: WriteMode, **kwargs
+        self, name: str, dataframe: PySparkDataFrame, **kwargs
     ) -> StreamingQuery:
         """Write a streaming DataFrame of deltas to a DLO in Data Cloud.
 
@@ -409,15 +409,12 @@ def write_dlo_deltas(
         Args:
             name: The name of the DLO to write to.
             dataframe: The streaming PySpark DataFrame to write.
-            write_mode: The write mode to use. Supported streaming modes are
-                ``WriteMode.APPEND``, ``WriteMode.OVERWRITE``, and
-                ``WriteMode.MERGE_UPSERT_DELETE``.
 
         Returns:
             The started ``StreamingQuery``.
         """
         self._validate_data_layer_history_does_not_contain(DataCloudObjectType.DMO)
-        return self._writer.write_dlo_deltas(name, dataframe, write_mode, **kwargs)  # type: ignore[no-any-return]
+        return self._writer.write_dlo_deltas(name, dataframe, **kwargs)  # type: ignore[no-any-return]
 
     def find_file_path(self, file_name: str) -> Path:
         """Resolve a bundled file shipped in the package to an absolute path.

src/datacustomcode/io/writer/base.py

@@ -60,25 +60,22 @@ def write_to_dmo(
     ) -> None: ...
 
     def write_dlo_deltas(
-        self, name: str, dataframe: PySparkDataFrame, write_mode: WriteMode
+        self, name: str, dataframe: PySparkDataFrame
     ) -> StreamingQuery:
         """Write a streaming DataFrame of deltas to a Data Lake Object.
 
         Streaming counterpart to :meth:`write_to_dlo`. Starts a streaming query
         that writes each micro-batch to the target DLO via the Data Cloud
         streaming sink and returns the resulting ``StreamingQuery`` handle. The
         runtime owns the trigger and checkpoint location; callers pass only the
-        table name and write mode. Concrete streaming behavior is provided by
-        the deployed Data Cloud runtime; the base implementation raises
+        table name. Concrete streaming behavior is provided by the deployed
+        Data Cloud runtime; the base implementation raises
         :class:`NotImplementedError`.
 
         Args:
             name: Target Data Lake Object name.
             dataframe: Streaming PySpark DataFrame produced from a
                 ``read_dlo_deltas`` / ``read_dmo_deltas`` source.
-            write_mode: Write mode for the streaming sink. Supported modes are
-                ``WriteMode.APPEND``, ``WriteMode.OVERWRITE``, and
-                ``WriteMode.MERGE_UPSERT_DELETE``.
 
         Returns:
             The started ``StreamingQuery``; the caller drives its lifecycle

src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py

@@ -7,10 +7,10 @@
 * ``client.read_dlo_deltas(name)`` returns a *streaming* DataFrame over the
   Change Data Feed of the source DLO. Each row carries the source columns plus
   change-feed metadata columns (``_record_type``, ``_commit_*``).
-* ``client.write_dlo_deltas(name, df, write_mode)`` starts a streaming query
-  that writes each micro-batch to the target DLO and returns the
-  ``StreamingQuery`` handle. The runtime owns the trigger and checkpoint
-  location — the caller only chooses the table and write mode.
+* ``client.write_dlo_deltas(name, df)`` starts a streaming query that writes
+  each micro-batch to the target DLO and returns the ``StreamingQuery`` handle.
+  The runtime owns the trigger, and checkpoint location — the caller only
+  chooses the table.
 
 The transform in between is ordinary PySpark. Because the source is a change
 feed, keep the metadata columns on the DataFrame you hand to
@@ -24,7 +24,6 @@
 from pyspark.sql.functions import col, upper
 
 from datacustomcode.client import Client
-from datacustomcode.io.writer.base import WriteMode
 
 
 def main():
@@ -33,16 +32,12 @@ def main():
     # Streaming DataFrame over the source DLO's change feed.
     deltas = client.read_dlo_deltas("Account_std__dll")
 
-    # Ordinary PySpark transform. Note we do NOT drop the change-feed metadata
-    # columns (those starting with "_") — the streaming sink needs them to apply
-    # inserts, updates, and deletes to the target DLO.
+    # Ordinary PySpark transform.
     transformed = deltas.withColumn("description__c", upper(col("description__c")))
 
     # Start the streaming write. write_dlo_deltas returns the StreamingQuery;
     # the trigger and checkpoint location are provided by the runtime.
-    query = client.write_dlo_deltas(
-        "Account_std_copy__dll", transformed, WriteMode.APPEND
-    )
+    query = client.write_dlo_deltas("Account_std_copy__dll", transformed)
 
     # Drive the query's lifecycle. In the streaming runtime this blocks until
     # the job is stopped by the platform.

tests/io/writer/test_print.py

@@ -62,7 +62,7 @@ def test_write_to_dmo(self, print_writer, mock_dataframe):
     def test_write_dlo_deltas_not_supported_locally(self, print_writer, mock_dataframe):
         """Streaming delta writes are not supported by the local writer."""
         with pytest.raises(NotImplementedError) as exc_info:
-            print_writer.write_dlo_deltas("test_dll", mock_dataframe, WriteMode.APPEND)
+            print_writer.write_dlo_deltas("test_dll", mock_dataframe)
         assert "write_dlo_deltas" in str(exc_info.value)
 
     def test_config_name(self):

tests/test_client.py

@@ -229,12 +229,10 @@ def test_write_dlo_deltas(self, reset_client, mock_spark):
         client = Client(reader=reader, writer=writer)
         client._record_dlo_access("some_dlo")
 
-        result = client.write_dlo_deltas(
-            "test_dlo", mock_df, WriteMode.APPEND, extra_param=True
-        )
+        result = client.write_dlo_deltas("test_dlo", mock_df, extra_param=True)
 
         writer.write_dlo_deltas.assert_called_once_with(
-            "test_dlo", mock_df, WriteMode.APPEND, extra_param=True
+            "test_dlo", mock_df, extra_param=True
         )
         assert result is mock_query
 
@@ -250,7 +248,7 @@ def test_write_dlo_deltas_after_dmo_read_raises_exception(
         client._record_dmo_access("test_dmo")
 
         with pytest.raises(DataCloudAccessLayerException) as exc_info:
-            client.write_dlo_deltas("test_dlo", mock_df, WriteMode.APPEND)
+            client.write_dlo_deltas("test_dlo", mock_df)
 
         assert "test_dmo" in str(exc_info.value)
         writer.write_dlo_deltas.assert_not_called()
@@ -265,12 +263,10 @@ def test_streaming_read_write_flow(self, reset_client, mock_spark):
         client = Client(reader=reader, writer=writer)
 
         df = client.read_dlo_deltas("source_dll")
-        client.write_dlo_deltas("target_dll", df, WriteMode.MERGE_UPSERT_DELETE)
+        client.write_dlo_deltas("target_dll", df)
 
         reader.read_dlo_deltas.assert_called_once_with("source_dll")
-        writer.write_dlo_deltas.assert_called_once_with(
-            "target_dll", stream_df, WriteMode.MERGE_UPSERT_DELETE
-        )
+        writer.write_dlo_deltas.assert_called_once_with("target_dll", stream_df)
         assert "source_dll" in client._data_layer_history[DataCloudObjectType.DLO]
 
     def test_mixed_dlo_dmo_raises_exception(self, reset_client, mock_spark):