Add Table.add_update_timestamp_utc (#9468)

GitOrigin-RevId: 8085403dca86639b300b5004a3e3426322239d48

Author: lewymati

CHANGELOG.md

@@ -8,6 +8,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Added
 - JetStream extension is now supported in both NATS read and write connectors.
 - The Iceberg connectors now support Glue as a catalog backend.
+- New `Table.add_update_timestamp_utc` function for tracking update time of rows in the table
 
 ### Changed
 - **BREAKING** The API for the Iceberg connectors has changed. The `catalog` parameter is now required in both `pw.io.iceberg.read` and `pw.io.iceberg.write`. This parameter can be either of type `pw.io.iceberg.RestCatalog` or `pw.io.iceberg.GlueCatalog`, and it must contain the connection parameters.

python/pathway/__init__.py

@@ -211,3 +211,4 @@
 Table._repr_mimebundle_ = viz._repr_mimebundle_
 
 Table.inactivity_detection = temporal.inactivity_detection
+Table.add_update_timestamp_utc = temporal.add_update_timestamp_utc

python/pathway/internals/table.py

@@ -75,6 +75,7 @@ class Table(
         from pathway.stdlib.ordered import diff  # type: ignore[misc]
         from pathway.stdlib.statistical import interpolate  # type: ignore[misc]
         from pathway.stdlib.temporal import (  # type: ignore[misc]
+            add_update_timestamp_utc,
             asof_join,
             asof_join_left,
             asof_join_outer,

python/pathway/stdlib/temporal/__init__.py

@@ -40,7 +40,7 @@
     common_behavior,
     exactly_once_behavior,
 )
-from .time_utils import inactivity_detection, utc_now
+from .time_utils import add_update_timestamp_utc, inactivity_detection, utc_now
 
 __all__ = [
     "AsofJoinResult",
@@ -79,4 +79,5 @@
     "exactly_once_behavior",
     "utc_now",
     "inactivity_detection",
+    "add_update_timestamp_utc",
 ]

python/pathway/stdlib/temporal/time_utils.py

@@ -184,3 +184,46 @@ def get_now_timestamp_utc(for_test_only: pw.Pointer) -> pw.DateTimeUtc:
         inactivities = inactivities.without(pw.this.instance)
 
     return inactivities
+
+
+@check_arg_types
+@trace_user_frame
+def add_update_timestamp_utc(
+    self: pw.Table,
+    refresh_rate: pw.Duration = pw.Duration(seconds=1),
+    update_timestamp_column_name: str = "updated_timestamp_utc",
+) -> pw.Table:
+    """Adds a column with the UTC timestamp of the last row update
+
+    Args:
+        refresh_rate (pw.Duration, optional): The interval at which the UTC
+            timestamp is refreshed. Defaults to 1 second.
+        update_timestamp_column_name (str, optional): The name of the column to
+            store the update timestamp. Defaults to "updated_timestamp_utc".
+
+    Returns:
+        pw.Table: A new table with an additional column containing the UTC
+            timestamp of the last update for each row. The id column is preserved.
+    """
+    utc_now_single_row = utc_now(refresh_rate=refresh_rate).reduce(
+        timestamp_utc=pw.reducers.latest(pw.this.timestamp_utc)
+    )
+
+    stream = self.with_columns(_id=pw.this.id).to_stream()
+
+    stream_joined = stream.asof_now_join_left(utc_now_single_row)
+    new_cols = {
+        update_timestamp_column_name: pw.coalesce(
+            pw.right.timestamp_utc,
+            pw.DateTimeUtc(datetime.datetime.now(tz=datetime.timezone.utc)),
+        )
+    }
+    stream_with_update_time = stream_joined.select(*pw.left, **new_cols)
+
+    result = (
+        stream_with_update_time.stream_to_table(pw.this.is_upsert)
+        .without(pw.this.is_upsert)
+        .with_id(pw.this._id)
+        .without(pw.this._id)
+    )
+    return result

python/pathway/tests/temporal/test_time_utils.py

@@ -9,7 +9,11 @@
 import pandas as pd
 
 import pathway as pw
-from pathway.tests.utils import T, assert_stream_equality_wo_index
+from pathway.tests.utils import (
+    T,
+    assert_stream_equality,
+    assert_stream_equality_wo_index,
+)
 
 
 def fake_utc_now(total_duration_ms: int) -> Callable[[pw.Duration], pw.Table]:
@@ -233,3 +237,37 @@ def test_inactivity_detection_empty_instance(utc_now_mock, get_now_timestamp_utc
     )
 
     assert_stream_equality_wo_index(inactivities, expected_inactivities)
+
+
+@patch("pathway.stdlib.temporal.time_utils.utc_now")
+def test_add_update_timestamp_utc(utc_now_mock):
+    utc_now_mock.side_effect = fake_utc_now(500)
+    t = T(
+        """
+           |  value | __time__ | __diff__
+        1  |     10 |        0 |        1
+        2  |     20 |        0 |        1
+        3  |     30 |        0 |        1
+        1  |     10 |      130 |       -1
+        1  |    110 |      130 |        1
+        3  |     30 |      230 |       -1
+    """
+    )
+    result = t.add_update_timestamp_utc(refresh_rate=pw.Duration(milliseconds=100))
+
+    expected = T(
+        """
+           |  value | updated_timestamp_utc | __time__ | __diff__
+        1  |     10 |                     0 |        0 |        1
+        2  |     20 |                     0 |        0 |        1
+        3  |     30 |                     0 |        0 |        1
+        1  |     10 |                     0 |      130 |       -1
+        1  |    110 |                   100 |      130 |        1
+        3  |     30 |                     0 |      230 |       -1
+    """
+    ).with_columns(
+        updated_timestamp_utc=pw.this.updated_timestamp_utc.dt.utc_from_timestamp(
+            unit="ms"
+        )
+    )
+    assert_stream_equality(result, expected)