Add dagster-pixeltable integration

libraries/dagster-pixeltable/.gitignore

@@ -0,0 +1,2 @@
+# temporary Dagster instances that `dagster asset materialize` creates without DAGSTER_HOME
+.tmp_dagster_home_*/

libraries/dagster-pixeltable/CHANGELOG.md

@@ -0,0 +1,11 @@
+# Changelog
+
+## [Unreleased]
+
+### Added
+
+- `PixeltableResource`, a `ConfigurableResource` that checks the Pixeltable home of the Dagster process against
+  `expected_home`, initializes the SDK during resource setup, and opens existing local tables with `get_table()`.
+- `examples/basic`: a table deployed with `pxt schema update`, upserted by primary key, with materialization
+  metadata and an asset check.
+- `examples/video`: a video table and frame view, appended with `insert`.

libraries/dagster-pixeltable/Makefile

@@ -0,0 +1,15 @@
+install:
+	uv sync
+
+build:
+	uv build
+
+test:
+	uv run pytest
+
+ruff:
+	uv run ruff check --fix
+	uv run ruff format
+
+check:
+	uv run ty check .

libraries/dagster-pixeltable/README.md

@@ -0,0 +1,132 @@
+# dagster-pixeltable
+
+Run Dagster jobs against existing [Pixeltable](https://github.com/pixeltable/pixeltable) tables. Dagster schedules
+runs, orders work across systems, and records run history. Pixeltable stores the data and maintains computed
+columns, views, and indexes.
+
+The package provides one resource, `PixeltableResource`. During resource setup it checks that the Dagster process
+uses the Pixeltable home you expect, then initializes the SDK. Assets open tables by full path and write to them
+with the Pixeltable SDK.
+
+## Installation
+
+```sh
+pip install dagster-pixeltable
+```
+
+Requires Python 3.11 or later, `dagster>=1.9.4`, and `pixeltable>=0.7.15`.
+
+## Usage
+
+```python
+import dagster as dg
+from dagster_pixeltable import PixeltableResource
+
+
+@dg.asset
+def ingest_documents(pixeltable: PixeltableResource) -> dg.MaterializeResult:
+    table = pixeltable.get_table("dagster_demo.documents")
+    status = table.batch_update(
+        [{"source_id": "doc-1", "body": "Hello", "page_count": 1}],
+        if_not_exists="insert",
+    )
+    return dg.MaterializeResult(
+        metadata={
+            "dagster/row_count": table.count(),
+            "pixeltable/num_excs": status.num_excs,
+        }
+    )
+
+
+defs = dg.Definitions(
+    assets=[ingest_documents],
+    resources={"pixeltable": PixeltableResource(expected_home="/srv/pixeltable")},
+    executor=dg.in_process_executor,
+)
+```
+
+### Environment
+
+Before starting the Dagster process, set `PIXELTABLE_HOME` to an absolute path, along with any database settings and
+provider API keys. Pixeltable fixes its home the first time the process uses the SDK, so a later change to
+`PIXELTABLE_HOME` has no effect. Run `pxt schema update` with the same environment.
+
+| Field | Meaning |
+| --- | --- |
+| `expected_home` | Absolute path of the Pixeltable home the job uses. Setup fails unless the process uses this home. |
+
+### Execution mode
+
+This release supports the in-process executor on one host with persistent local storage. Set
+`executor=dg.in_process_executor` on `Definitions`: Dagster's default executor runs each step in its own process,
+and this release has not been tested that way.
+
+### Table access
+
+`get_table(table_path)` opens an existing local table, view, or snapshot by its full catalog path, such as
+`dagster_demo.documents`.
+
+- It never creates a table, and never derives a path from an asset key.
+- A version suffix, such as `dagster_demo.documents:7`, opens a read-only handle to that version.
+- Hosted paths (`pxt://...`, `https://pixeltable.com/t/...`) are rejected.
+- It works only during a run, after resource setup.
+
+## Examples
+
+Each example declares its tables in an `app.py`. `pxt schema update` creates the tables, and Dagster writes rows
+into them. Run the commands from `examples/`, which holds the Pixeltable project file:
+
+```sh
+cd examples
+export PIXELTABLE_HOME="$HOME/.pixeltable"
+pxt schema update basic/app.py dagster_demo
+dagster asset materialize -f basic/definitions.py -a defs --select ingest_documents
+```
+
+On Dagster 1.13 and later, `dagster asset materialize` prints a supersession warning and still runs. For the UI,
+install `dagster-webserver` and start `dagster dev -f basic/definitions.py -a defs` from the same shell, so it
+inherits `PIXELTABLE_HOME`.
+
+### `examples/basic`: upsert by source id
+
+`dagster_demo.documents` has a `source_id` primary key, two input columns, and two computed columns. The
+`ingest_documents` asset reads `basic/documents.json` and writes it with
+`batch_update(records, if_not_exists="insert")`:
+
+- **Replay.** A record whose `source_id` already exists updates that row, so running the asset twice keeps one row
+  per id. Each run can still advance the table version and recompute values. UDF side effects are not
+  exactly-once.
+- **Rejected input.** A record without a `source_id`, or two records with the same one, fail the run before the
+  write. An empty file skips the write and succeeds.
+- **SDK errors.** An unknown column or a value of the wrong type raises, and the table is unchanged.
+- **Computed errors.** A record with `page_count` 0 commits with a division error stored in `chars_per_page`. The
+  asset reports the count in `pixeltable/num_excs`, and the `documents_computed` check fails. A failed check does
+  not undo the write.
+
+The asset returns a `MaterializeResult` with the table path, row count, column schema, table id, and version. The
+version is the one observed after the write; downstream reads see the current table, not that version.
+
+### `examples/video`: append media
+
+`batch_update` cannot write media columns, so `ingest_videos` uses `insert`. Each run appends `clip.mp4` again, and
+the `frames` view extracts its frames. Declare media columns explicitly, as `pxt.Video` here: a URL inserted into
+an inferred schema becomes a `String` column, which `frame_iterator` rejects.
+
+```sh
+pxt schema update video/app.py dagster_video_demo
+dagster asset materialize -f video/definitions.py -a defs --select ingest_videos
+```
+
+## Development
+
+```sh
+make install
+make test
+make ruff
+make check
+```
+
+The tests create a temporary Pixeltable home with its own embedded Postgres server and pxt daemon, and stop both
+when the session ends.
+
+Tested on macOS and Linux.

libraries/dagster-pixeltable/dagster_pixeltable/__init__.py

@@ -0,0 +1,11 @@
+from dagster._core.libraries import DagsterLibraryRegistry
+
+from dagster_pixeltable.resource import PixeltableResource
+
+__version__ = "0.1.0"
+
+__all__ = ["PixeltableResource"]
+
+DagsterLibraryRegistry.register(
+    "dagster-pixeltable", __version__, is_dagster_package=False
+)

libraries/dagster-pixeltable/dagster_pixeltable/resource.py

@@ -0,0 +1,116 @@
+import os
+from pathlib import Path
+
+import pixeltable as pxt
+from dagster import ConfigurableResource, InitResourceContext
+from pydantic import Field, PrivateAttr
+
+# pxt.get_table() opens these outside the catalog of this process; the web forms normalize to pxt:// URIs
+_REMOTE_PREFIXES = (
+    "pxt://",
+    "https://www.pixeltable.com/t/",
+    "http://www.pixeltable.com/t/",
+    "https://pixeltable.com/t/",
+    "http://pixeltable.com/t/",
+)
+
+
+class PixeltableResource(ConfigurableResource):
+    """Opens existing tables in the Pixeltable instance of the Dagster process.
+
+    Pixeltable keeps one instance per process, chosen by the ``PIXELTABLE_HOME`` environment variable when the
+    SDK first initializes. ``expected_home`` does not select that instance: setup fails unless the process
+    already uses it.
+
+    Examples:
+        .. code-block:: python
+
+            import dagster as dg
+            from dagster_pixeltable import PixeltableResource
+
+            @dg.asset
+            def documents(pixeltable: PixeltableResource) -> dg.MaterializeResult:
+                table = pixeltable.get_table("my_app.documents")
+                status = table.insert([{"source_id": "a", "body": "hello"}])
+                return dg.MaterializeResult(metadata={"num_rows": status.num_rows})
+
+            defs = dg.Definitions(
+                assets=[documents],
+                resources={"pixeltable": PixeltableResource(expected_home="/srv/pixeltable")},
+                executor=dg.in_process_executor,
+            )
+    """
+
+    expected_home: str = Field(
+        description=(
+            "Absolute path of the Pixeltable home this job uses. Setup fails unless it matches the home of the "
+            "process, which PIXELTABLE_HOME sets before the SDK first initializes."
+        ),
+    )
+
+    _ready: bool = PrivateAttr(default=False)
+
+    @classmethod
+    def _is_dagster_maintained(cls) -> bool:
+        return False
+
+    def setup_for_execution(self, context: InitResourceContext) -> None:
+        self._ready = False
+        expected = Path(self.expected_home).expanduser()
+        if not expected.is_absolute():
+            raise ValueError(
+                f"expected_home must be an absolute path, got {self.expected_home!r}"
+            )
+        expected = expected.resolve()
+
+        # checked before pxt.home(), which creates the home directory and fixes the process configuration
+        env_home = os.environ.get("PIXELTABLE_HOME")
+        if env_home:
+            if not Path(env_home).is_absolute():
+                raise ValueError(
+                    f"PIXELTABLE_HOME must be an absolute path with no '~', got {env_home!r}. Pixeltable does not "
+                    "expand it, and a relative path depends on the working directory."
+                )
+            _check_home(expected, Path(env_home).resolve())
+
+        # an earlier initialization in this process can hold a different home than the environment names now
+        effective = pxt.home()
+        if not effective.is_absolute():
+            raise ValueError(
+                f"The Pixeltable home of this process is not an absolute path: {str(effective)!r}"
+            )
+        _check_home(expected, effective.resolve())
+
+        pxt.init()
+        self._ready = True
+
+    def teardown_after_execution(self, context: InitResourceContext) -> None:
+        self._ready = False
+
+    def get_table(self, table_path: str) -> pxt.Table:
+        """Open an existing local table, view, or snapshot.
+
+        Args:
+            table_path: The full catalog path, such as ``my_app.documents``. A version suffix, such as
+                ``my_app.documents:7``, opens a read-only handle to that version.
+        """
+        if not self._ready:
+            raise RuntimeError(
+                "PixeltableResource.get_table() runs only inside a Dagster execution, after resource setup."
+            )
+        if table_path.startswith(_REMOTE_PREFIXES):
+            raise ValueError(
+                f"PixeltableResource opens only local tables, got {table_path!r}"
+            )
+        table = pxt.get_table(table_path)
+        # with the default if_not_exists="error", a missing table raises instead
+        assert table is not None
+        return table
+
+
+def _check_home(expected: Path, effective: Path) -> None:
+    if effective != expected:
+        raise ValueError(
+            f"This process uses the Pixeltable home {str(effective)!r}, but expected_home is {str(expected)!r}. "
+            "Set PIXELTABLE_HOME to the expected home before starting the Dagster process."
+        )

libraries/dagster-pixeltable/dagster_pixeltable_tests/conftest.py

@@ -0,0 +1,45 @@
+import socket
+import uuid
+from collections.abc import Iterator
+from pathlib import Path
+
+import pixeltable as pxt
+import pixeltable_pgserver
+import pytest
+
+from dagster_pixeltable_tests.utils import run_command
+
+
+def _free_port() -> int:
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+@pytest.fixture(scope="session", autouse=True)
+def pixeltable_home(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Path]:
+    """One Pixeltable home for the session, set before the SDK first reads it."""
+    home = tmp_path_factory.mktemp("pixeltable_home").resolve()
+    with pytest.MonkeyPatch.context() as mp:
+        mp.setenv("PIXELTABLE_HOME", str(home))
+        # the pxt daemon of another home may hold the default port
+        mp.setenv("PXT_PORT", str(_free_port()))
+        try:
+            yield home
+        finally:
+            run_command("pxt", "daemon", "stop", cwd=home, check=False)
+            # the SDK leaves its embedded Postgres server running after the process exits
+            if (home / "pgdata" / "postmaster.pid").exists():
+                # pixeltable_pgserver generates pg_ctl() at import time
+                pixeltable_pgserver.pg_ctl(  # ty: ignore[unresolved-attribute]
+                    ["-w", "-m", "fast", "stop"], pgdata=home / "pgdata"
+                )
+
+
+@pytest.fixture
+def catalog_dir() -> Iterator[str]:
+    """A catalog directory owned by the test."""
+    name = f"test_{uuid.uuid4().hex[:8]}"
+    pxt.create_dir(name)
+    yield name
+    pxt.drop_dir(name, force=True)