feat: metadata propagation (system/user) in pynumaflow-lite

packages/pynumaflow-lite/pynumaflow_lite/mapper.pyi

@@ -7,6 +7,62 @@ import datetime as _dt
 from ._map_dtypes import Mapper as Mapper
 
 
+class SystemMetadata:
+    """System-generated metadata groups per message (read-only)."""
+
+    def __init__(self) -> None: ...
+
+    def groups(self) -> List[str]:
+        """Returns the groups of the system metadata."""
+        ...
+
+    def keys(self, group: str) -> List[str]:
+        """Returns the keys of the system metadata for the given group."""
+        ...
+
+    def value(self, group: str, key: str) -> bytes:
+        """Returns the value of the system metadata for the given group and key."""
+        ...
+
+    def __repr__(self) -> str: ...
+
+
+class UserMetadata:
+    """User-defined metadata groups per message (read-write)."""
+
+    def __init__(self) -> None: ...
+
+    def groups(self) -> List[str]:
+        """Returns the groups of the user metadata."""
+        ...
+
+    def keys(self, group: str) -> List[str]:
+        """Returns the keys of the user metadata for the given group."""
+        ...
+
+    def value(self, group: str, key: str) -> bytes:
+        """Returns the value of the user metadata for the given group and key."""
+        ...
+
+    def create_group(self, group: str) -> None:
+        """Creates a new group in the user metadata."""
+        ...
+
+    def add_kv(self, group: str, key: str, value: bytes) -> None:
+        """Adds a key-value pair to the user metadata."""
+        ...
+
+    def remove_key(self, group: str, key: str) -> None:
+        """Removes a key from a group in the user metadata."""
+        ...
+
+    def remove_group(self, group: str) -> None:
+        """Removes a group from the user metadata."""
+        ...
+
+    def __repr__(self) -> str: ...
+
+
 class Messages:
     def __init__(self) -> None: ...
 
@@ -21,12 +77,14 @@ class Message:
     keys: Optional[List[str]]
     value: bytes
     tags: Optional[List[str]]
+    user_metadata: Optional[UserMetadata]
 
     def __init__(
             self,
             value: bytes,
             keys: Optional[List[str]] = ...,
             tags: Optional[List[str]] = ...,
+            user_metadata: Optional[UserMetadata] = ...,
     ) -> None: ...
 
     @staticmethod
@@ -40,6 +98,8 @@ class Datum:
     watermark: _dt.datetime
     eventtime: _dt.datetime
     headers: Dict[str, str]
+    user_metadata: UserMetadata
+    system_metadata: SystemMetadata
 
     def __repr__(self) -> str: ...
 
@@ -62,6 +122,8 @@ class MapAsyncServer:
 
 
 __all__ = [
+    "SystemMetadata",
+    "UserMetadata",
     "Messages",
     "Message",
     "Datum",

packages/pynumaflow-lite/pynumaflow_lite/sinker.pyi

@@ -4,6 +4,46 @@ from typing import Optional, List, Dict, Callable, Awaitable, Any, AsyncIterator
 import datetime as _dt
 
 
+class SystemMetadata:
+    """System-generated metadata groups per message (read-only for sink)."""
+
+    def __init__(self) -> None: ...
+
+    def groups(self) -> List[str]:
+        """Returns the groups of the system metadata."""
+        ...
+
+    def keys(self, group: str) -> List[str]:
+        """Returns the keys of the system metadata for the given group."""
+        ...
+
+    def value(self, group: str, key: str) -> bytes:
+        """Returns the value of the system metadata for the given group and key."""
+        ...
+
+    def __repr__(self) -> str: ...
+
+
+class UserMetadata:
+    """User-defined metadata groups per message (read-only for sink)."""
+
+    def __init__(self) -> None: ...
+
+    def groups(self) -> List[str]:
+        """Returns the groups of the user metadata."""
+        ...
+
+    def keys(self, group: str) -> List[str]:
+        """Returns the keys of the user metadata for the given group."""
+        ...
+
+    def value(self, group: str, key: str) -> bytes:
+        """Returns the value of the user metadata for the given group and key."""
+        ...
+
+    def __repr__(self) -> str: ...
+
+
 class KeyValueGroup:
     key_value: Dict[str, bytes]
 
@@ -58,6 +98,8 @@ class Datum:
     eventtime: _dt.datetime
     id: str
     headers: Dict[str, str]
+    user_metadata: UserMetadata
+    system_metadata: SystemMetadata
 
     def __repr__(self) -> str: ...
 
@@ -81,6 +123,8 @@ class Sinker:
 
 
 __all__ = [
+    "SystemMetadata",
+    "UserMetadata",
     "KeyValueGroup",
     "Message",
     "Response",

packages/pynumaflow-lite/pynumaflow_lite/sourcer.pyi

@@ -7,13 +7,50 @@ import datetime as _dt
 from ._source_dtypes import Sourcer as Sourcer
 
 
+class UserMetadata:
+    """User-defined metadata groups per message (read-write for source)."""
+
+    def __init__(self) -> None: ...
+
+    def groups(self) -> List[str]:
+        """Returns the groups of the user metadata."""
+        ...
+
+    def keys(self, group: str) -> List[str]:
+        """Returns the keys of the user metadata for the given group."""
+        ...
+
+    def value(self, group: str, key: str) -> bytes:
+        """Returns the value of the user metadata for the given group and key."""
+        ...
+
+    def create_group(self, group: str) -> None:
+        """Creates a new group in the user metadata."""
+        ...
+
+    def add_kv(self, group: str, key: str, value: bytes) -> None:
+        """Adds a key-value pair to the user metadata."""
+        ...
+
+    def remove_key(self, group: str, key: str) -> None:
+        """Removes a key from a group in the user metadata."""
+        ...
+
+    def remove_group(self, group: str) -> None:
+        """Removes a group from the user metadata."""
+        ...
+
+    def __repr__(self) -> str: ...
+
+
 class Message:
     """A message to be sent from the source."""
     payload: bytes
     offset: Offset
     event_time: _dt.datetime
     keys: List[str]
     headers: Dict[str, str]
+    user_metadata: Optional[UserMetadata]
 
     def __init__(
         self,
@@ -22,6 +59,7 @@ class Message:
         event_time: _dt.datetime,
         keys: Optional[List[str]] = ...,
         headers: Optional[Dict[str, str]] = ...,
+        user_metadata: Optional[UserMetadata] = ...,
     ) -> None: ...
 
     def __repr__(self) -> str: ...
@@ -122,6 +160,7 @@ class SourceAsyncServer:
 
 
 __all__ = [
+    "UserMetadata",
     "Message",
     "Offset",
     "ReadRequest",

packages/pynumaflow-lite/pynumaflow_lite/sourcetransformer.pyi

@@ -7,6 +7,62 @@ import datetime as _dt
 from ._sourcetransformer_dtypes import SourceTransformer as SourceTransformer
 
 
+class SystemMetadata:
+    """System-generated metadata groups per message (read-only)."""
+
+    def __init__(self) -> None: ...
+
+    def groups(self) -> List[str]:
+        """Returns the groups of the system metadata."""
+        ...
+
+    def keys(self, group: str) -> List[str]:
+        """Returns the keys of the system metadata for the given group."""
+        ...
+
+    def value(self, group: str, key: str) -> bytes:
+        """Returns the value of the system metadata for the given group and key."""
+        ...
+
+    def __repr__(self) -> str: ...
+
+
+class UserMetadata:
+    """User-defined metadata groups per message (read-write)."""
+
+    def __init__(self) -> None: ...
+
+    def groups(self) -> List[str]:
+        """Returns the groups of the user metadata."""
+        ...
+
+    def keys(self, group: str) -> List[str]:
+        """Returns the keys of the user metadata for the given group."""
+        ...
+
+    def value(self, group: str, key: str) -> bytes:
+        """Returns the value of the user metadata for the given group and key."""
+        ...
+
+    def create_group(self, group: str) -> None:
+        """Creates a new group in the user metadata."""
+        ...
+
+    def add_kv(self, group: str, key: str, value: bytes) -> None:
+        """Adds a key-value pair to the user metadata."""
+        ...
+
+    def remove_key(self, group: str, key: str) -> None:
+        """Removes a key from a group in the user metadata."""
+        ...
+
+    def remove_group(self, group: str) -> None:
+        """Removes a group from the user metadata."""
+        ...
+
+    def __repr__(self) -> str: ...
+
+
 class Messages:
     def __init__(self) -> None: ...
 
@@ -22,13 +78,15 @@ class Message:
     value: bytes
     event_time: _dt.datetime
     tags: Optional[List[str]]
+    user_metadata: Optional[UserMetadata]
 
     def __init__(
             self,
             value: bytes,
             event_time: _dt.datetime,
             keys: Optional[List[str]] = ...,
             tags: Optional[List[str]] = ...,
+            user_metadata: Optional[UserMetadata] = ...,
     ) -> None: ...
 
     @staticmethod
@@ -42,6 +100,8 @@ class Datum:
     watermark: _dt.datetime
     event_time: _dt.datetime
     headers: Dict[str, str]
+    user_metadata: UserMetadata
+    system_metadata: SystemMetadata
 
     def __repr__(self) -> str: ...
 
@@ -61,6 +121,8 @@ class SourceTransformAsyncServer:
 
 
 __all__ = [
+    "SystemMetadata",
+    "UserMetadata",
     "Messages",
     "Message",
     "Datum",