Harden Dataflow preview behavior

README.md

@@ -2,7 +2,7 @@
 
 GovOPlaN Dataflow defines and runs governed tabular transformation pipelines.
 Power users can work with the same pipeline as a graphical node graph or as a
-constrained SQL query. Every saved change produces an immutable revision, and
+constrained SQL query. Every saved definition change produces an immutable revision, and
 every preview records diagnostics and reproducibility metadata without storing
 the previewed row contents.
 

src/govoplan_dataflow/backend/executor.py

@@ -267,10 +267,10 @@ def _sort_rows(rows: list[dict[str, Any]], config: dict[str, Any]) -> list[dict[
     for field in reversed(config["fields"]):
         column = str(field["column"])
         reverse = field.get("direction", "asc") == "desc"
-        result.sort(
-            key=lambda row: (row.get(column) is None, _sortable_value(row.get(column))),
-            reverse=reverse,
-        )
+        concrete = [row for row in result if row.get(column) is not None]
+        nulls = [row for row in result if row.get(column) is None]
+        concrete.sort(key=lambda row: _sortable_value(row[column]), reverse=reverse)
+        result = [*concrete, *nulls]
     return result
 
 

tests/test_graph_and_sql.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import unittest
 
-from govoplan_dataflow.backend.executor import execute_preview
+from govoplan_dataflow.backend.executor import PipelineExecutionError, execute_preview
 from govoplan_dataflow.backend.schemas import (
     GraphEdge,
     GraphNode,
@@ -120,6 +120,89 @@ class DataflowGraphAndSqlTests(unittest.TestCase):
         self.assertEqual(3, result.total_rows)
         self.assertTrue(result.truncated)
 
+    def test_empty_aggregate_and_null_sort_are_deterministic(self) -> None:
+        graph, _, _ = compile_sql(
+            """
+            SELECT status, COUNT(*) AS records
+            FROM monthly_files
+            GROUP BY status
+            ORDER BY status DESC
+            """,
+            source_nodes=[
+                inline_source().model_copy(
+                    update={
+                        "config": {
+                            "source_name": "monthly_files",
+                            "rows": [
+                                {"status": None},
+                                {"status": "open"},
+                                {"status": "closed"},
+                            ],
+                        }
+                    }
+                )
+            ],
+        )
+
+        first = execute_preview(graph, row_limit=100)
+        second = execute_preview(graph, row_limit=100)
+
+        self.assertEqual(first.rows, second.rows)
+        self.assertEqual(
+            [
+                {"status": "open", "records": 1},
+                {"status": "closed", "records": 1},
+                {"status": None, "records": 1},
+            ],
+            first.rows,
+        )
+        self.assertEqual(
+            [
+                diagnostic.model_dump(exclude={"duration_ms"})
+                for diagnostic in first.node_diagnostics
+            ],
+            [
+                diagnostic.model_dump(exclude={"duration_ms"})
+                for diagnostic in second.node_diagnostics
+            ],
+        )
+
+    def test_empty_global_aggregate_returns_a_count_row(self) -> None:
+        graph, _, _ = compile_sql(
+            "SELECT COUNT(*) AS records FROM monthly_files",
+            source_nodes=[
+                inline_source().model_copy(
+                    update={"config": {"source_name": "monthly_files", "rows": []}}
+                )
+            ],
+        )
+
+        result = execute_preview(graph, row_limit=100)
+
+        self.assertEqual([{"records": 0}], result.rows)
+        self.assertEqual(0, result.input_row_count)
+
+    def test_invalid_runtime_comparison_is_scoped_to_the_node(self) -> None:
+        graph, _, _ = compile_sql(
+            "SELECT * FROM monthly_files WHERE amount > 10",
+            source_nodes=[
+                inline_source().model_copy(
+                    update={
+                        "config": {
+                            "source_name": "monthly_files",
+                            "rows": [{"amount": "not-a-number"}],
+                        }
+                    }
+                )
+            ],
+        )
+
+        with self.assertRaises(PipelineExecutionError) as raised:
+            execute_preview(graph, row_limit=100)
+
+        self.assertEqual("filter-1", raised.exception.node_id)
+        self.assertIn("Cannot apply", str(raised.exception))
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_service.py

@@ -24,7 +24,9 @@ from govoplan_dataflow.backend.service import (
     DataflowConflictError,
     DataflowNotFoundError,
     create_pipeline,
+    delete_pipeline,
     get_pipeline,
+    list_pipelines,
     preview_pipeline,
     update_pipeline,
 )
@@ -168,6 +170,37 @@ class DataflowServiceTests(unittest.TestCase):
                 pipeline_id=pipeline.id,
             )
 
+    def test_list_and_soft_delete_are_tenant_isolated(self) -> None:
+        first = self._create()
+        second = self._create(tenant_id="tenant-2")
+
+        self.assertEqual([first.id], [item.id for item in list_pipelines(self.session, tenant_id="tenant-1")])
+        self.assertEqual([second.id], [item.id for item in list_pipelines(self.session, tenant_id="tenant-2")])
+
+        delete_pipeline(
+            self.session,
+            tenant_id="tenant-1",
+            pipeline_id=first.id,
+            actor_id="user-2",
+        )
+        self.session.commit()
+
+        self.assertEqual([], list_pipelines(self.session, tenant_id="tenant-1"))
+        with self.assertRaises(DataflowNotFoundError):
+            get_pipeline(
+                self.session,
+                tenant_id="tenant-1",
+                pipeline_id=first.id,
+            )
+        self.assertEqual(
+            1,
+            self.session.scalar(
+                select(func.count())
+                .select_from(DataflowPipelineRevision)
+                .where(DataflowPipelineRevision.pipeline_id == first.id)
+            ),
+        )
+
     def test_saved_preview_records_lineage_but_not_result_rows(self) -> None:
         pipeline = self._create()