Hold MLOps artifacts on Unity Catalog volumes through the Files API

packages/kumo-databricks/README.md

@@ -67,6 +67,35 @@ events = Table.from_source_table(
 snapshot = Graph(tables=[events]).snapshot()
 ```
 
+## Artifacts on a volume
+
+A `/Volumes/<catalog>/<schema>/<volume>/...` path is a file destination of a
+`DatabricksConnector` from any machine, not only on Databricks compute: the
+sink reads and writes it through the Files API with the connector's identity.
+
+```python
+from kumo.engine import ModelOutputConfig, PredictionOutputConfig, Trainer
+
+root = "/Volumes/main/kumo/artifacts"
+trainer.fit(training_table, snapshot=snapshot, export_config=ModelOutputConfig(
+    databricks, path=f"{root}/models/churn"))
+trainer = Trainer.load_from_connector(databricks, path=f"{root}/models/churn")
+trainer.predict(prediction_table, snapshot=snapshot, output_config=PredictionOutputConfig(
+    databricks, output_paths={"predictions": f"{root}/scores/"}))
+```
+
+For an MLOps artifact root on a volume, see `packages/kumo-engine/docs/mlops.md`.
+
+A file written with a known size (bundle files, exported Parquet parts) is one
+object up to 4 GiB, readable directly by Spark and other tools. A larger one
+streams into `<name>.chunks/<generation>/00000`, `00001`, ... (4 GiB each,
+under the Files API's 5 GiB upload cap) with an empty `_SUCCESS` written last;
+Kumo reads the newest committed generation back as one file. A file written
+without a known size is staged on local disk under the engine scratch
+directory and stored the same way. The identity needs `USE CATALOG`,
+`USE SCHEMA`, and `READ VOLUME` (loads and exports) and `WRITE VOLUME`
+(exports) on the volume; no warehouse is used.
+
 ## Model registry
 
 The sink publishes Kumo bundles directly to Unity Catalog using the connector's

packages/kumo-databricks/src/kumo_databricks/registry.py

@@ -34,7 +34,8 @@ def _address(connector: SourceConnector, path: Optional[str],
              location: Optional[ModelRegistryLocation]) -> str:
     if path is not None:
         raise ValueError('Databricks models take a ModelRegistryLocation, '
-                         'rather than a path.')
+                         'or a /Volumes/<catalog>/<schema>/<volume>/... '
+                         'path.')
     if location is None:
         raise ValueError('Pass a ModelRegistryLocation naming the model.')
     config = _target_config(connector)

packages/kumo-databricks/src/kumo_databricks/sink.py

@@ -26,64 +26,18 @@
 
 import pandas as pd
 from databricks.sql.client import Connection
+from kumo_core.object_store_sink import FileStoreRouting
 from kumo_databricks.registry import DatabricksModelRegistryMixin
 from kumo_databricks.session import _target_config
+from kumo_databricks.volume import is_volume_path, volume_filesystem
 from kumo_databricks.write import RunVolume
 from kumoapi.data_source import DataSourceType
 from kumoapi.jobs import WriteMode
 from kumoapi.plugin import SourceConnector
+from pyarrow.fs import FileSystem
 
 
-class DatabricksSinkBackend(DatabricksModelRegistryMixin):
-    r"""Writes a produced artifact to a Databricks Delta table.
-
-    Registered by default beside
-    :class:`~kumo_databricks.backend.DatabricksSourceBackend`, so a
-    run that read its graph from a catalog exports its predictions back to one
-    with no further setup. Installed as
-    ``pip install 'kumoai-studio[databricks]'``.
-
-    A destination names its schema and its table in one string, joined by
-    ``__kumo__``, because that is the only field a Databricks output has --
-    :func:`kumoai_studio.databricks.to_db_table_name` builds it from the
-    ``(schema, table)`` tuple an :class:`~kumoai_studio.artifact_export.OutputConfig`
-    takes. The schema must be the connector's own, which
-    :func:`kumo_databricks.output.within_schema` checks and the
-    hosted connector checks through as well.
-
-    Every write stages through the connector's managed volume,
-    ``<catalog>.<schema>._kumo_staging``, created on first use. That is not a
-    local choice: ``COPY INTO`` is how a Delta table is loaded, and hosted
-    states the alternatives -- S3, ADLS, an arbitrary volume, inline cloud
-    credentials -- are unsupported on purpose. The volume is emptied of what
-    this export put there once the table has landed.
-
-    Args:
-        identity: A fallback identity, used only for connectors that name none
-            themselves. As with the reader, leaving it unset is the common
-            case, since a bound
-            :class:`~kumoai_studio.connector.DatabricksConnector` records the
-            identity it was created with. That identity needs ``CREATE TABLE``
-            on the connector's schema and ``READ``/``WRITE VOLUME`` on its
-            staging volume, which a read-only role used for ingest will not
-            have.
-
-    Example:
-        >>> import kumoai_studio as kumo  # doctest: +SKIP
-        >>> from kumo_databricks import (  # doctest: +SKIP
-        ...     DatabricksSinkBackend)
-        >>> kumo.set_backend(kumo.LocalBackend(  # doctest: +SKIP
-        ...     storage=paths,
-        ...     sinks={kumo.DataSourceType.DATABRICKS:
-        ...            DatabricksSinkBackend(identity='prod-dbx')},
-        ... ))
-    """
-
-    source_type = DataSourceType.DATABRICKS
-
-    _default_hint = ('LocalBackend(sinks={DataSourceType.DATABRICKS: '
-                     'DatabricksSinkBackend(identity=identity)})')
-
+class _DatabricksTables(DatabricksModelRegistryMixin):
     def write_table(
         self,
         connector: SourceConnector,
@@ -153,7 +107,7 @@ def write_table(
     def write_files(self, connector, table_name, files,
                     write_mode=WriteMode.OVERWRITE, project_id=None, *,
                     export_id=None, extra_columns=None):
-        """Pass selected Parquet directly to the existing hosted COPY owner."""
+        """COPY selected Parquet into a table."""
         del project_id, export_id
         from functools import partial
 
@@ -187,6 +141,30 @@ def write_files(self, connector, table_name, files,
                     prepare_file=prepare)
         return sql.qualified_name(config.catalog, schema, table)
 
+
+class DatabricksSinkBackend(FileStoreRouting, _DatabricksTables):
+    """Writes artifacts to Delta tables, staged through the schema's volume.
+
+    Args:
+        identity: Fallback identity for connectors that name none.
+    """
+
+    source_type = DataSourceType.DATABRICKS
+
+    _default_hint = ('LocalBackend(sinks={DataSourceType.DATABRICKS: '
+                     'DatabricksSinkBackend(identity=identity)})')
+
+    def holds_files(self, connector: SourceConnector, path: str) -> bool:
+        return is_volume_path(path)
+
+    def filesystem(self, connector: SourceConnector,
+                   path: str) -> tuple[FileSystem, str]:
+        if not self.holds_files(connector, path):
+            raise ValueError(
+                f"{path!r} is not a Unity Catalog volume path; Databricks "
+                f"artifacts go under /Volumes/<catalog>/<schema>/<volume>/.")
+        return volume_filesystem(self._client(connector)), path.rstrip('/')
+
     def __repr__(self) -> str:
         return f'DatabricksSinkBackend(identity={self._identity!r})'