Merge branch 'develop'

Author: dolfinus

.env.docker

@@ -37,6 +37,7 @@ DATA_RENTGEN__KAFKA__COMPRESSION=zstd
 DATA_RENTGEN__UI__API_BROWSER_URL=http://localhost:8000
 
 # Session
+DATA_RENTGEN__SERVER__SESSION__ENABLED=True
 DATA_RENTGEN__SERVER__SESSION__SECRET_KEY=session_secret_key
 
 # Keycloak Auth
@@ -54,8 +55,8 @@ DATA_RENTGEN__AUTH__PROVIDER=data_rentgen.server.providers.auth.dummy_provider.D
 DATA_RENTGEN__AUTH__ACCESS_TOKEN__SECRET_KEY=access_key_secret
 
 # Personal Tokens
-export DATA_RENTGEN__AUTH__PERSONAL_TOKENS__ENABLED=True
-export DATA_RENTGEN__AUTH__PERSONAL_TOKENS__SECRET_KEY=pat_secret
+DATA_RENTGEN__AUTH__PERSONAL_TOKENS__ENABLED=True
+DATA_RENTGEN__AUTH__PERSONAL_TOKENS__SECRET_KEY=pat_secret
 
 # Cors
 DATA_RENTGEN__SERVER__CORS__ENABLED=True

.env.local

@@ -16,6 +16,7 @@ export DATA_RENTGEN__SERVER__DEBUG=True
 export DATA_RENTGEN__UI__API_BROWSER_URL=http://localhost:8000
 
 # Session
+export DATA_RENTGEN__SERVER__SESSION__ENABLED=True
 export DATA_RENTGEN__SERVER__SESSION__SECRET_KEY=session_secret_key
 
 # Keycloak Auth

.pre-commit-config.yaml

@@ -51,7 +51,7 @@ repos:
           - --no-extra-eol
 
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.21.0
+    rev: v3.21.1
     hooks:
       - id: pyupgrade
         args: [--py37-plus, --keep-runtime-typing]
@@ -62,7 +62,7 @@ repos:
       - id: add-trailing-comma
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.14.2
+    rev: v0.14.5
     hooks:
       - id: ruff-check
         args: [--fix]
@@ -88,7 +88,7 @@ repos:
           - tomli
 
   - repo: https://github.com/astral-sh/uv-pre-commit
-    rev: 0.9.5
+    rev: 0.9.9
     hooks:
       - id: uv-lock
 

README.rst

@@ -48,15 +48,15 @@ Goals
 -----
 
 * Collect lineage events produced by OpenLineage clients & integrations.
-* Store operation-grained events for better detalization (instead of job grained `Marquez <https://marquezproject.ai/>`_).
+* Store operation-grained events for better detalization.
 * Provide API for fetching both job/run ↔ dataset lineage and dataset ↔ dataset lineage.
 
 Features
 --------
 
 * Support consuming large amounts of lineage events, use Apache Kafka as event buffer.
 * Store data in tables partitioned by event timestamp, to speed up lineage graph resolution.
-* Lineage graph is build with user-specified time boundaries (unlike Marquez where lineage is build only for last job run).
+* Lineage graph is build with user-specified time boundaries.
 * Lineage graph can be build with different granularity. e.g. merge all individual Spark commands into Spark applicationId or Spark applicationName.
 * Column-level lineage support.
 * Authentication support.
@@ -71,7 +71,7 @@ Limitations
 -----------
 
 * OpenLineage have integrations with Trino, Debezium and some other lineage sources. DataRentgen support may be added later.
-* Unlike Marquez, DataRentgen parses only limited set of facets send by OpenLineage, and doesn't store custom facets. This can be changed in future.
+* DataRentgen parses only limited set of OpenLineage facets, and doesn't store custom facets. This can be changed in future.
 
 .. documentation
 

data_rentgen/VERSION

@@ -1 +1 @@
-0.4.2
+0.4.3

data_rentgen/consumer/extractors/generic/operation.py

@@ -2,6 +2,8 @@
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
+import codecs
+import re
 from abc import ABC, abstractmethod
 from textwrap import dedent
 
@@ -17,6 +19,13 @@
     OpenLineageRunEventType,
 )
 
+# https://www.ascii-code.com/, but left \n intact
+ASCII_UNPRINTABLE = re.compile(r"[\x00-\x09\x0b-\x1f\x7f]", re.UNICODE)
+
+
+def encode_char(char: re.Match[str]) -> str:
+    return codecs.encode(char.group(0), "unicode-escape").decode("utf-8")
+
 
 class OperationExtractorMixin(ABC):
     @abstractmethod
@@ -74,5 +83,7 @@ def _enrich_operation_status(self, operation: OperationDTO, event: OpenLineageRu
     def _extract_sql_query(self, event: OpenLineageRunEvent) -> SQLQueryDTO | None:
         if event.job.facets.sql:
             query = dedent(event.job.facets.sql.query).strip()
+            # https://stackoverflow.com/questions/56237415/removing-encoding-utf8-0x00-chars-from-pandas-dataframe-for-psycopg2-cursor
+            query = ASCII_UNPRINTABLE.sub(encode_char, query)
             return SQLQueryDTO(query=query)
         return None

data_rentgen/consumer/subscribers.py

@@ -102,7 +102,6 @@ async def report_malformed(
         await publisher.publish(
             message.value,
             key=message.key,
-            partition=message.partition,
             timestamp_ms=message.timestamp,
             headers=headers or None,
             reply_to=message_id,

data_rentgen/db/migrations/versions/2025-11-21_8e8891273099_truncate_inputs_with_too_much_size_in_.py

@@ -0,0 +1,26 @@
+# SPDX-FileCopyrightText: 2024-2025 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+"""Truncate inputs with too much size in bytes
+
+Revision ID: 8e8891273099
+Revises: 102502e85b2d
+Create Date: 2025-11-21 18:28:52.279644
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "8e8891273099"
+down_revision = "102502e85b2d"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(sa.text("UPDATE input SET num_bytes = NULL WHERE num_bytes >= 9223372036854775807"))
+
+
+def downgrade() -> None:
+    pass

data_rentgen/openlineage/dataset_facets/input_statistics.py

@@ -1,18 +1,29 @@
 # SPDX-FileCopyrightText: 2024-2025 MTS PJSC
 # SPDX-License-Identifier: Apache-2.0
 
-from pydantic import Field
+from pydantic import Field, PositiveInt, field_validator
 
 from data_rentgen.openlineage.dataset_facets.base import (
     OpenLineageInputDatasetFacet,
 )
 
+MAX_LONG = 2**63 - 1
+
 
 class OpenLineageInputStatisticsInputDatasetFacet(OpenLineageInputDatasetFacet):
     """Dataset facet describing Input statistics.
     See [InputStatisticsInputDatasetFacet](https://github.com/OpenLineage/OpenLineage/blob/main/spec/facets/InputStatisticsInputDatasetFacet.json).
     """
 
-    rows: int | None = Field(default=None, alias="rowCount", examples=[1_000_000])
-    bytes: int | None = Field(default=None, alias="size", examples=[2**30])
-    files: int | None = Field(default=None, alias="fileCount", examples=[0])
+    rows: PositiveInt | None = Field(default=None, alias="rowCount", examples=[1_000_000])
+    bytes: PositiveInt | None = Field(default=None, alias="size", examples=[2**30])
+    files: PositiveInt | None = Field(default=None, alias="fileCount", examples=[0])
+
+    @field_validator("bytes", "rows", "files", mode="after")
+    @classmethod
+    def value_must_be_sane(cls, value: int | None):
+        if value and value >= MAX_LONG:
+            # https://github.com/apache/spark/blob/v3.5.7/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L2565
+            # https://github.com/apache/spark/blob/v3.5.7/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala#L209
+            return None
+        return value

data_rentgen/server/middlewares/session.py

@@ -8,9 +8,11 @@
 
 def apply_session_middleware(app: FastAPI, settings: SessionSettings) -> FastAPI:
     """Add SessionMiddleware middleware to the application."""
+    if not settings.enabled:
+        return app
 
-    settings_dict = settings.model_dump(exclude={"secret_key"})
-    settings_dict["secret_key"] = settings.secret_key.get_secret_value()
+    settings_dict = settings.model_dump(exclude={"secret_key", "enabled"})
+    settings_dict["secret_key"] = settings.secret_key.get_secret_value()  # type: ignore[union-attr]
 
     app.add_middleware(SessionMiddleware, **settings_dict)
     return app