replace pandas usage with pyarrow

Author: zilto

dlt/helpers/dashboard/dlt_dashboard.py

@@ -13,14 +13,10 @@
 
     import marimo as mo
 
-    import pandas as pd
-
     import dlt
-    from dlt.common.json import json
+    import pyarrow
     from dlt.helpers.dashboard import strings, utils, ui_elements as ui
     from dlt.helpers.dashboard.config import DashboardConfiguration
-    from dlt.destinations.dataset.dataset import ReadableDBAPIDataset
-    from dlt.destinations.dataset.relation import ReadableDBAPIRelation
 
 
 @app.cell(hide_code=True)
@@ -848,7 +844,7 @@ def utils_caches_and_state(
     """
 
     # some state variables
-    dlt_get_last_query_result, dlt_set_last_query_result = mo.state(pd.DataFrame())
+    dlt_get_last_query_result, dlt_set_last_query_result = mo.state(pyarrow.table({}))
     # a cache of query results in the form of {query: row_count}
     dlt_get_query_cache, dlt_set_query_cache = mo.state(cast(Dict[str, int], {}))
 

dlt/helpers/dashboard/runner.py

@@ -20,7 +20,7 @@ def _detect_dashboard_command() -> str:
 # keep this, will raise if user tries to run dashboard without dependencies
 try:
     import marimo
-    import pandas
+    import pyarrow
     import ibis
 except ModuleNotFoundError:
     raise MissingDependencyException(

dlt/helpers/dashboard/utils.py

@@ -10,8 +10,7 @@
 
 import dlt
 import marimo as mo
-import pandas as pd
-import yaml
+import pyarrow
 import traceback
 
 
@@ -381,7 +380,7 @@ def get_example_query_for_dataset(pipeline: dlt.Pipeline, schema_name: str) -> T
     return "", "Schema does not contain any tables.", None
 
 
-def get_query_result(pipeline: dlt.Pipeline, query: str) -> Tuple[pd.DataFrame, str, str]:
+def get_query_result(pipeline: dlt.Pipeline, query: str) -> Tuple[pyarrow.Table, str, str]:
     """
     Get the result of a query. Parses the query to ensure it is a valid SQL query before sending it to the destination.
     """
@@ -392,12 +391,12 @@ def get_query_result(pipeline: dlt.Pipeline, query: str) -> Tuple[pd.DataFrame,
         )
         return get_query_result_cached(pipeline, query), None, None
     except Exception as exc:
-        return pd.DataFrame(), _exception_to_string(exc), traceback.format_exc()
+        return pyarrow.table({}), _exception_to_string(exc), traceback.format_exc()
 
 
 @functools.cache
-def get_query_result_cached(pipeline: dlt.Pipeline, query: str) -> pd.DataFrame:
-    return pipeline.dataset()(query, _execute_raw_query=True).df()
+def get_query_result_cached(pipeline: dlt.Pipeline, query: str) -> pyarrow.Table:
+    return pipeline.dataset()(query, _execute_raw_query=True).arrow()
 
 
 def get_row_counts(
@@ -415,8 +414,8 @@ def get_row_counts(
             i["table_name"]: i["row_count"]
             for i in pipeline.dataset(schema=selected_schema_name)
             .row_counts(dlt_tables=True, load_id=load_id)
-            .df()
-            .to_dict(orient="records")
+            .arrow()
+            .to_pylist()
         }
     except (
         DatabaseUndefinedRelation,
@@ -451,7 +450,7 @@ def get_loads(
         if limit:
             loads = loads.limit(limit)
         loads = loads.order_by("inserted_at", "desc")
-        loads_list = loads.df().to_dict(orient="records")
+        loads_list = loads.arrow().to_pylist()
         loads_list = [_humanize_datetime_values(c, load) for load in loads_list]
         return loads_list, None, None
     except Exception as exc:

pyproject.toml

@@ -179,9 +179,9 @@ postgis = [
 ]
 workspace = [
     "duckdb>=0.9",
-    "ibis-framework[duckdb]>=10.5.0 ; python_version >= '3.10'",
+    "ibis-framework>=10.5.0 ; python_version >= '3.10'",
+    "duckdb",
     "pyarrow>=16.0.0",
-    "pandas>=2.1.4",
     "marimo>=0.14.5",
 ]
 dbml = [

tests/helpers/dashboard/test_utils.py

@@ -2,20 +2,13 @@
 import tempfile
 from datetime import datetime
 from pathlib import Path
-from this import d
 
-from ibis import pi
 import marimo as mo
+import pyarrow
 import pytest
-import pandas as pd
-
-from dlt.sources._single_file_templates.fruitshop_pipeline import (
-    fruitshop as fruitshop_source,
-)
-import pendulum
-
 
 import dlt
+from dlt.common import pendulum
 from dlt.helpers.dashboard.config import DashboardConfiguration
 from dlt.helpers.dashboard.utils import (
     PICKLE_TRACE_FILE,
@@ -346,10 +339,10 @@ def test_get_query_result(pipeline: dlt.Pipeline):
     )
 
     if pipeline.pipeline_name in PIPELINES_WITH_LOAD:
-        assert isinstance(result, pd.DataFrame)
+        assert isinstance(result, pyarrow.Table)
         assert len(result) == 1
         assert (
-            result.iloc[0]["count"] == 100
+            result[0][0].as_py() == 100
             if pipeline.pipeline_name == SUCCESS_PIPELINE_DUCKDB
             else 103
         )  #  merge does not work on filesystem
@@ -697,7 +690,7 @@ def test_integration_pipeline_workflow(pipeline, temp_pipelines_dir):
     )
     if pipeline.pipeline_name in PIPELINES_WITH_LOAD:
         assert len(query_result) == 13
-        assert query_result.iloc[0]["name"] == "simon"
+        assert query_result[0][0].as_py() == "simon"
         assert not error_message
         assert not traceback_string
     else: