Add intermediate previews and graph reconnection

tests/test_graph_and_sql.py

@@ -446,6 +446,43 @@ class DataflowGraphAndSqlTests(unittest.TestCase):
         self.assertEqual(3, result.total_rows)
         self.assertTrue(result.truncated)
 
+    def test_preview_can_return_a_bounded_intermediate_node_result(self) -> None:
+        source = inline_source()
+        filter_node = GraphNode(
+            id="filter",
+            type="filter",
+            label="Open cases",
+            position=GraphPosition(x=240, y=160),
+            config={"column": "status", "operator": "eq", "value": "open"},
+        )
+        output = GraphNode(
+            id="output",
+            type="output",
+            label="Output",
+            position=GraphPosition(x=440, y=160),
+            config={},
+        )
+        graph = PipelineGraph(
+            nodes=[source, filter_node, output],
+            edges=[
+                GraphEdge(id="source-filter", source="source", target="filter"),
+                GraphEdge(id="filter-output", source="filter", target="output"),
+            ],
+        )
+
+        result = execute_preview(
+            graph,
+            row_limit=1,
+            preview_node_id="source",
+        )
+
+        self.assertEqual(2, result.total_rows)
+        self.assertIsNotNone(result.node_preview)
+        self.assertEqual("source", result.node_preview.node_id)
+        self.assertEqual(3, result.node_preview.total_rows)
+        self.assertEqual(1, len(result.node_preview.rows))
+        self.assertTrue(result.node_preview.truncated)
+
     def test_empty_aggregate_and_null_sort_are_deterministic(self) -> None:
         graph, _, _ = compile_sql(
             """
@@ -524,7 +561,7 @@ class DataflowGraphAndSqlTests(unittest.TestCase):
         )
 
         with self.assertRaises(PipelineExecutionError) as raised:
-            execute_preview(graph, row_limit=100)
+            execute_preview(graph, row_limit=100, preview_node_id="source")
 
         self.assertEqual("filter-1", raised.exception.node_id)
         self.assertIn("Cannot apply", str(raised.exception))
@@ -536,6 +573,12 @@ class DataflowGraphAndSqlTests(unittest.TestCase):
             ],
         )
         self.assertEqual(1, len(raised.exception.source_fingerprints))
+        self.assertIsNotNone(raised.exception.node_preview)
+        self.assertEqual("source", raised.exception.node_preview.node_id)
+        self.assertEqual(
+            [{"amount": "not-a-number"}],
+            raised.exception.node_preview.rows,
+        )
 
 
 if __name__ == "__main__":

tests/test_service.py

@@ -277,7 +277,11 @@ class DataflowServiceTests(unittest.TestCase):
             self.session,
             tenant_id="tenant-1",
             actor_id="user-1",
-            payload=PipelinePreviewRequest(pipeline_id=pipeline.id, row_limit=1),
+            payload=PipelinePreviewRequest(
+                pipeline_id=pipeline.id,
+                preview_node_id="source",
+                row_limit=1,
+            ),
             principal=principal(),
         )
         self.session.commit()
@@ -287,6 +291,11 @@ class DataflowServiceTests(unittest.TestCase):
         self.assertEqual([{"id": 2, "amount": 15}], response.rows)
         self.assertEqual(2, response.total_rows)
         self.assertTrue(response.truncated)
+        self.assertIsNotNone(response.node_preview)
+        self.assertEqual("source", response.node_preview.node_id)
+        self.assertEqual([{"id": 1, "amount": 5}], response.node_preview.rows)
+        self.assertEqual(3, response.node_preview.total_rows)
+        self.assertTrue(response.node_preview.truncated)
         self.assertEqual(2, run.output_row_count)
         self.assertEqual(3, run.input_row_count)
         self.assertEqual(1, len(run.source_fingerprints))