Avoid recomputing Opas in Pipe (#54)

* Avoid recomputing Opas in Pipe

* Minor demo cleanup

* Remove private name mangling variables in Pipe

Author: mare5x

examples/demo.py

@@ -34,6 +34,8 @@
 thread = session.thread()
 thread.ask("count cancelled shows by directors")
 print(thread.text())
+print(thread.code)
 print(thread.df())
+
 plot = thread.plot()
-print(thread.code)
+print(plot.text)

portus/agents/lighthouse/agent.py

@@ -53,6 +53,8 @@ def _get_graph_and_compiled(self, session: Session) -> tuple[Any, ExecuteSubmit,
     def execute(
         self, session: Session, opa: Opa, *, rows_limit: int = 100, cache_scope: str = "common_cache"
     ) -> ExecutionResult:
+        # TODO rows_limit is ignored
+
         # Get or create graph (cached after first use)
         data_connection, graph, compiled_graph = self._get_graph_and_compiled(session)
 

portus/core/pipe.py

@@ -1,3 +1,4 @@
+import uuid
 from typing import TYPE_CHECKING, Any
 
 from pandas import DataFrame
@@ -12,64 +13,68 @@
 
 class Pipe:
     def __init__(self, session: "Session", *, default_rows_limit: int = 1000):
-        self.__session = session
-        self.__default_rows_limit = default_rows_limit
+        self._session = session
+        self._default_rows_limit = default_rows_limit
 
-        self._data_materialized = False
         self._data_materialized_rows: int | None = None
         self._data_result: ExecutionResult | None = None
+
         self._visualization_materialized = False
         self._visualization_result: VisualisationResult | None = None
         self._visualization_request: str | None = None
 
+        # N.B. Pipes/Threads are currently append-only and cannot be "forked".
+        self._opas_processed_count = 0
         self._opas: list[Opa] = []
         self._meta: dict[str, Any] = {}
 
-    def __materialize_data(self, rows_limit: int | None) -> "ExecutionResult":
-        rows_limit = rows_limit if rows_limit else self.__default_rows_limit
-        if not self._data_materialized or rows_limit != self._data_materialized_rows:
-            # Execute each opa individually, keeping the last result
-            for opa in self._opas:
-                self._data_result = self.__session.executor.execute(
-                    self.__session, opa, rows_limit=rows_limit, cache_scope=str(id(self))
+        self._cache_scope = f"{self._session.name}/{uuid.uuid4()}"
+
+    def _materialize_data(self, rows_limit: int | None) -> "ExecutionResult":
+        # TODO Recompute on rows_limit change without recomputing the last Opa
+        rows_limit = rows_limit if rows_limit else self._default_rows_limit
+        new_opas = self._opas[self._opas_processed_count :]
+        if len(new_opas) > 0 or rows_limit != self._data_materialized_rows:
+            for opa in new_opas:
+                self._data_result = self._session.executor.execute(
+                    self._session, opa, rows_limit=rows_limit, cache_scope=self._cache_scope
                 )
                 self._meta.update(self._data_result.meta)
-            self._data_materialized = True
+            self._opas_processed_count += len(new_opas)
             self._data_materialized_rows = rows_limit
         if self._data_result is None:
-            raise RuntimeError("__data_result is None after materialization")
+            raise RuntimeError("_data_result is None after materialization")
         return self._data_result
 
-    def __materialize_visualization(self, request: str | None, rows_limit: int | None) -> "VisualisationResult":
-        data = self.__materialize_data(rows_limit)
+    def _materialize_visualization(self, request: str | None, rows_limit: int | None) -> "VisualisationResult":
+        data = self._materialize_data(rows_limit)
         if not self._visualization_materialized or request != self._visualization_request:
             # TODO Cache visualization results as in Executor.execute()?
-            self._visualization_result = self.__session.visualizer.visualize(request, data)
+            self._visualization_result = self._session.visualizer.visualize(request, data)
             self._visualization_materialized = True
             self._visualization_request = request
             self._meta.update(self._visualization_result.meta)
             self._meta["plot_code"] = self._visualization_result.code  # maybe worth to expand as a property later
         if self._visualization_result is None:
-            raise RuntimeError("__visualization_result is None after materialization")
+            raise RuntimeError("_visualization_result is None after materialization")
         return self._visualization_result
 
     def df(self, *, rows_limit: int | None = None) -> DataFrame | None:
-        return self.__materialize_data(rows_limit if rows_limit else self._data_materialized_rows).df
+        return self._materialize_data(rows_limit if rows_limit else self._data_materialized_rows).df
 
     def plot(self, request: str | None = None, *, rows_limit: int | None = None) -> "VisualisationResult":
         # TODO Currently, we can't chain calls or maintain a "plot history": pipe.plot("red").plot("blue").
         #  We have to do pipe.plot("red"), but then pipe.plot("blue") is independent of the first call.
-        return self.__materialize_visualization(request, rows_limit if rows_limit else self._data_materialized_rows)
+        return self._materialize_visualization(request, rows_limit if rows_limit else self._data_materialized_rows)
 
     def text(self) -> str:
-        return self.__materialize_data(self._data_materialized_rows).text
+        return self._materialize_data(self._data_materialized_rows).text
 
     def __str__(self) -> str:
         return self.text()
 
     def ask(self, query: str) -> "Pipe":
         self._opas.append(Opa(query=query))
-        self._data_materialized = False
         self._visualization_materialized = False
         return self
 
@@ -79,4 +84,4 @@ def meta(self) -> dict[str, Any]:
 
     @property
     def code(self) -> str | None:
-        return self.__materialize_data(self._data_materialized_rows).code
+        return self._materialize_data(self._data_materialized_rows).code