§17 AI Data Ecosystem

Status: Approved · 2026-09-25 (decisions D51–D56) · amended 2026-09-26 (§3.7, M1.2 as built). Items marked (verify) are resolved in the implementation plans.

This document uses the product name Loams and the package name loams (D33, D400). Scan pinning (M1.2) and to_arrow() / to_polars() (M1.6) landed before the rename; everything else here ships after it.

1. Goal and principleCopy Anchor Link

Frontier AI labs run their data pipelines on a small set of tools: Ray Data for curation, deduplication and batch inference; PySpark (on Apache Spark, and increasingly on Sail) for large filters and joins; PyTorch and JAX loaders for training; Polars and DuckDB for local inspection. A lab adopts a new store when those tools read and write it without a copy step. Loams's retrieval data (documents, vectors, the eval sets built from them) must therefore be a source and a sink for those tools, alongside the Qdrant, Elasticsearch and Flight SQL surfaces that agent frameworks use (D42).

Principle (D51): integrate over protocols and open formats, never by embedding other engines. The integration surfaces are:

Lockstep rationale. A Rust binary links one version of DataFusion and one of arrow-rs, and engines built on them upgrade together. Lance 12 pins DataFusion 54 / arrow 58, which fixes Loams's versions (§11 §1). Sail's main branch moved to DataFusion 55.1 / arrow 59.2 on 2026-09-17; Spice patches crates.io with its own forks of DataFusion 54 and arrow-rs; Polars carries a second Arrow implementation (polars-arrow, forked from arrow2). Linking any of them would tie Loams's upgrade cadence to the slowest of Lance, Sail and Spice, or force a fork (risk 21). Sail's maintainers declined a Rust-level Lance integration for the same reason. Protocols and file formats do not move in lockstep, so every integration in this document crosses a process or format boundary, is pinned, and is tested in CI against the released tool.

2. Integration mapCopy Anchor Link

Daft and Hugging Face datasets need nothing from Loams beyond scan pinning; they are listed so that the scan plan stays readable by any Lance reader that accepts a dataset URI and a version.

Log shippers and stream tools (Fluent Bit, the OpenTelemetry Collector, Vector, RisingWave, Flink, Kafka Connect) are listed in §02 §7.3, with OTLP logs ingest in M2 (D73) and the Kafka gateway in M5 (D74).

3. Scan pinning (D53, M1.2)Copy Anchor Link

Loams commits Lance as detached versions (D34): the dataset's mainline holds only the empty version 1, and every collection manifest names its own detached lance_version. A plain ray.data.read_lance(uri) or lance.dataset(uri) therefore opens an empty dataset. Typed fields exist only in Tantivy (D36), and writes newer than the manifest are in the stream tail, not in Lance. External readers therefore get a pinned, consistent Lance version only by asking Loams for a scan plan (§03 §3.3).

3.1 Request and planCopy Anchor Link

POST /v1/namespaces/{ns}/collections/{c}/scan-plans (and the SDK's collection.scan_plan(…)) resolves the collection at one of:

The request may also carry columns (Lance columns, vector and sparse-vector names, or _source field paths), a filter in the native filter IR (§3.3), tail (§3.2) and, from M2, credentials: true (§3.4). A plan on a multi-target alias (D57) returns one pinned sub-plan per target collection.

The plan:

The same plan is exposed through Flight: GetFlightInfo for a scan-plan command returns a FlightInfo whose endpoints are the fragments (plus one endpoint for the tail), so a Flight-native client parallelizes a scan without the REST call. Exact shapes are settled in the M1.2 plan.

Readers read each fragment directly from object storage with Lance (pylance, lance-ray), which applies the fragment's Lance deletion file, or through its Flight ticket when they have no object-store access. Both paths return identical rows. Lance deletion files are part of the pinned version (§03 §3.3 step 3), so a direct read of the pinned fragments sees exactly the live documents of the manifest.

3.2 The tailCopy Anchor Link

A manifest reflects the implicit stream up to its applied offsets; later acknowledged writes are the tail (§05 §5). The request's tail field chooses how a plan treats it:

A reader that merges the tail reads each fragment with _rowid excluded if it is in the superseded set, then appends the tail batch. The adapters do this themselves; a plain Lance reader given only lance.uri and lance.version must use a tail-free plan. A tag is always tail-free (D52), which is why training and eval jobs read tags.

3.3 Filters and projectionsCopy Anchor Link

Lance holds _source, system columns and vectors; typed fields exist only in Tantivy (D36). A direct fragment read therefore sees _source (JSON bytes), _pk, _ingest_partition, _ingest_offset, _rowid and the vector columns, and no typed columns.

Server-side filtering keeps one filter evaluator (Tantivy, R5 of the M1 overview) for search, SQL and scans, so a filter means the same thing on every surface.

3.4 CredentialsCopy Anchor Link

Direct reads bypass per-request authorization inside a collection, so credentials are vended only to principals whose read is unrestricted on that collection; a principal with field-level masking (Phase B) or any other restricted read gets Flight tickets only, and Loams enforces the restriction on the Flight path.

3.5 Opening a detached version (Q20)Copy Anchor Link

Answered (Q20, M1.2 Task 0): pylance 12.0.0 and the Rust lance crate 12.0.0 open a detached version by id: lance.dataset(uri, version=<detached id>) and DatasetBuilder::from_uri(uri).with_version(id).load(), with Lance's default session and registry, both accept the u64 id (bit 63 set). The fallbacks below are not needed; the plan keeps lance.manifest_path anyway. Opening from the manifest file's bytes does not work (serialized_manifest expects the bare protobuf). The original question: whether pylance (and the Rust lance crate behind lance-ray and Daft) opens a detached version by id, e.g. lance.dataset(uri, version=<detached id>). If it cannot, the fallbacks, in order: open the Lance manifest at lance.manifest_path through a Lance API that accepts a manifest location; have the Loams adapters build Lance fragment objects from the plan's fragment metadata; or, if Lance tags can reference detached versions, mirror each dataset tag (§4) as a Lance tag so that readers that know only Lance tags work unmodified (verify; Lance cleanup is never run, D35, so a Lance tag would not change GC). Daft and Hugging Face datasets work without a Loams adapter only under the first answer.

3.6 ValidityCopy Anchor Link

A plan pins a manifest without a metastore hold (D38): it stays readable while the manifest is retained, i.e. at least time_travel_retention (24 h) after the manifest is superseded, measured from the child manifest's creation. valid_until reports the earliest time the manifest may be released. A job that runs longer, or must be rerun later, pins a tag instead (§4). A read of a released manifest fails with a typed SnapshotExpired error, never with missing rows.

3.7 As built (M1.2)Copy Anchor Link

M1.2 ships the core of §3.1–§3.6; the rest (filters, projections, Flight tickets per fragment, tail: merge|wait|none, credentials, multi-target aliases) stays with the SDK adapters (M1.6, M2) and D52–D57's milestones.

• Route: POST /v1/namespaces/{ns}/collections/{c}/scan (alias ok), body {"at"?: …}, answering the plan and the header Loams-Consistency-Token: <pin.token>. at is "current" (the default), {"manifest_version": v} or {"token": "v1:…"}; {"tag": …} is refused ("tags arrive in M2 (D52)"). Any node serves it: it reads the metastore, the manifest chain and one Lance manifest, never the tail.
• The plan: namespace, collection, collection_id, manifest_version, schema_version; lance (null before the first Lance version, else {uri, version, manifest_path, storage_format, stable_row_ids}, version the detached id); fragments[] in Lance's order (id, physical_rows, deleted_rows, live_rows, files as data/… paths, deletion_file as {path, kind, deleted_rows} or null, and lance, Lance's own Fragment JSON for FragmentMetadata.from_json); live_rows (equal to the manifest's live document count, checked); columns (the Lance version's schema with roles pk, source, ingest_partition, ingest_offset, vector, sparse_vector, and each vector's name, dim, distance or modifier); pk_encoding loams_canonical_v1; tail, tail_records and per-partition offsets (applied against target); durable_token; pin; planned_at_ms and expires_at_ms. The plan carries no credentials (D54, M2).
• Lance deletion files are authoritative (M1.2 Ruling 22): M1.1's commits delete every removed and superseded row in Lance itself, so the Lance version holds exactly the manifest's live rows, and the plan lists each fragment's deletion file for readers that bypass Lance; split delete bitmaps are not in the plan. What Lance cannot hold is the tail: the plan reports it (tail, tail_records, offsets) and carries a pin ({manifest_version, token}) that reads exactly the requested state, tail included, through the native API ("consistency": {"pinned": …}) or Flight SQL (metadata loams-pin-manifest with loams-consistency-token).
• Token scans wait for durability, not for the tail (M1.2 Ruling 24): {"token": t} waits up to consistency_wait (10 s) for a live manifest whose applied offsets cover t, then plans it (its target is max(token, applied) per partition, M1.2 plan row 14.2), else Timeout; current never waits and reports the tail; a manifest version is planned as it is, with no tail.
• Expiry (D38, A21): expires_at_ms is planned_at_ms + time_travel_retention for the live manifest, the child manifest's created_at_ms + time_travel_retention for a superseded one, and null for manifest version 0. Both are on the metastore clock, which advances only with time-stamped metastore writes, so planned_at_ms is 0 on a cluster that has had none (M1.2 plan row 15.5). A reader compares expires_at_ms − planned_at_ms with its own clock.
• Verified with the Rust lance crate over file:// (the Lance version holds exactly the folded live documents); pylance, Daft, FragmentMetadata.from_json and s3:// URIs are checked by M1.6 and M1.7.

4. Retained dataset tags (D52, M2)Copy Anchor Link

A dataset tag is a named, immutable pointer from a collection to one manifest (from M4, from a table to one Iceberg snapshot) that is retained until the tag is deleted. Tags are the unit of reproducibility for training runs and eval sets, which outlive the 24 h time-travel window. The idea comes from TileDB's timestamped fragments and from Lance and Iceberg tags.

4.1 APICopy Anchor Link

Tag names are unique per collection and never move: retagging is delete and create. The SDK mirrors the API (collection.create_tag(name, at=token), list_tags(), delete_tag(name), and tag= on every read and adapter). Tags live in the catalog behind MetaStore as domain operations (create, delete, list), so they are linearizable with manifest retention. The Qdrant and Elasticsearch surfaces do not expose tags. A tag created with protect: true makes a drop of its collection fail until the tag is deleted; otherwise dropping a collection deletes its tags.

4.2 GC and trimmingCopy Anchor Link

• A tag is a GC root: CollectionGcRoots keeps the tagged manifest and every object a retained manifest keeps (the files of its Lance version, its splits and delete bitmaps, its PK delta and dead letters; §03 §7). Manifests are self-contained, so a tag retains neither its manifest's ancestors nor their objects.
• Creating a tag and releasing a manifest are decided in the metastore: a tag can be created only on a manifest that is still retained, and GC reads tags and retention from one metastore snapshot. A manifest GC saw as retained is not deleted in that pass, and a manifest GC saw as released can no longer be tagged.
• Tags never hold back implicit-stream trimming (D38): a tag is taken only on a manifest whose applied offsets cover the requested token, so reading a tag never needs the tail, and streams are trimmed below the oldest time-travel-retained manifest exactly as before.
• Compaction and split merges after a tag leave the tagged manifest's old fragments and splits in place: a tag costs storage in proportion to how much of the collection has been rewritten since it was taken.
• Expired tags (expires_at) are deleted by a worker task, which then leaves their objects to GC.

4.3 QuotasCopy Anchor Link

Tags are covered by the M2 tenant quotas: a per-namespace max_tags_per_collection (default 64) and the storage quota, which counts the bytes reachable only from tags. Each tag reports its exclusive bytes (objects no live or time-travel-retained manifest references) and the namespace metrics carry the total, so the storage cost of an old tag is visible.

4.4 Tags on tables (M4)Copy Anchor Link

A tag on a table creates an Iceberg tag ref on the snapshot through Lakekeeper, with no max-ref-age-ms, so Iceberg snapshot expiry retains it for as long as the Loams tag exists. External engines read it with Iceberg's own syntax (Spark VERSION AS OF '<tag>', Trino FOR VERSION AS OF '<tag>'; verify per engine), and Ray Data, Polars and loams.torch read it through the snapshot id the tag resolves to (M4 gate). Keyed tables read at a tag apply their deletion vectors as of that snapshot.

5. Python SDK extras (D54)Copy Anchor Link

5.1 PackagingCopy Anchor Link

The SDK's runtime dependencies stay minimal (the M1.6 plan's httpx and anyio); every integration is an optional extra, imported only by its module:

Versions are pinned in the M2 plan and each extra is tested in CI against the released tool (risk 21).

5.2 to_arrow() and to_polars() (M1.6)Copy Anchor Link

Query results, search results and fetched documents expose to_arrow() (a pyarrow.Table) and to_polars() (a polars.DataFrame), and implement __arrow_c_stream__, so any PyCapsule consumer (Polars since 1.3, pandas, DuckDB, pyarrow) takes them without a copy. Results that arrive over Flight SQL are Arrow already; REST results are converted once in the SDK. Typed fields come from _source with the collection's field schema.

5.3 Ray Data datasource and datasink (M2)Copy Anchor Link

import ray
from loams.ray import read_loams, LoamsDatasink

ds = read_loams(client, "web_docs", tag="dedup-2026-09-25",
               columns=["_pk", "text", "embedding"], filter={"term": {"lang": "en"}})
ds = ds.map_batches(Embedder, concurrency=16, num_gpus=1, batch_size=256)
sink = LoamsDatasink(client, "web_docs", write="patch", vectors=["embedding_v2"])
ds.write_datasink(sink)
token = sink.consistency_token

• Read. read_loams requests a scan plan (§3) and builds a ray.data.Datasource whose get_read_tasks(parallelism) returns one read task per fragment (small fragments packed together up to parallelism), each with its live row count and size estimate as block metadata. Tasks read fragments directly with lance-ray's fragment reader (Flight tickets when mode="flight"), apply their row_filter slice and the superseded-row set, and project typed fields from _source. Ray's projection and predicate pushdown hooks map onto columns and filter; the tail, when merged, is one extra read task.
• Write. LoamsDatasink implements on_write_start (checks or creates the collection), write (each task streams its blocks through Flight DoPut with a path descriptor, ["collections", c], and returns the merged PutAck token of its puts; M1.2 Task 13) and on_write_complete (merges every task's token into one consistency token, exposed as sink.consistency_token, which a following read_loams(…, at=token) or tag creation uses). on_write_failed has nothing to roll back: acknowledged batches are visible, and writes are upserts keyed by _id, so a retried Ray task or a rerun job converges to the same documents.
• Embedding backfills write only the new vector column: the sink's write="patch" sends a DoPut write mode that maps each row to a Patch of the listed vectors instead of an Upsert (a DoPut option added in M2; verify against M1.2 Task 13's mapping). Large backfills run on the lab's own Ray cluster with ray.data.llm; Loams's embed() links (§09) serve continuous, smaller streams. Both are documented with an example.
• Streams become a Ray source in M5, over the native streaming subscribe (which ships in M2, D72).

5.4 Polars scan_loams() (M2, experimental)Copy Anchor Link

loams.polars.scan_loams(client, collection, tag=…) returns a LazyFrame registered with polars.io.plugins.register_io_source, whose generator receives with_columns, predicate, n_rows and batch_size. It is marked experimental because Polars marks register_io_source unstable.

• with_columns becomes the plan's columns.
• The predicate is translated from Polars expressions to the native filter IR where it can be (comparisons, is_in, is_null, &, |, ~ over typed fields and system columns) and sent with the plan request (§3.3). The generator applies any untranslated remainder with DataFrame.filter before yielding each batch.
• n_rows stops the scan early; batch_size sizes the Lance reads.
• Tables (M4) use Polars' native scan_iceberg against Lakekeeper; scan_loams() is for collections. Polars' own scan_lance (a draft PR upstream) would still need a scan plan to find the version.

5.5 loams.torch (M2)Copy Anchor Link

A thin training adapter, not a data-loading framework. It reads with Lance's PyTorch reader (lance.torch.data.LanceDataset) and replaces Lance's ShardedFragmentSampler / ShardedBatchSampler, which are deterministic for a fixed world size only, with a sampler that is elastic:

• Pinned input. A dataset is opened on a tag (or an explicit manifest version for short runs); the loader records the tag's manifest version and refuses to resume against a different one.
• Deterministic order. The epoch's sample order is a function of (manifest, seed, epoch, P) only. Samples, addressed by their position among the live rows of the plan's fragments in fragment order, are divided into P canonical partitions (contiguous fragment ranges; P is fixed for the whole run and stored in the checkpoint, by default the first run's world size, as Mosaic Streaming's num_canonical_nodes). Each partition is shuffled independently with a seeded block shuffle (shuffle fixed-size blocks, then samples within a window of blocks), and the global order interleaves the P partitions round-robin.
• Elastic assignment. Rank r of W takes global positions consumed + r, consumed + r + W, …; DataLoader workers stripe their rank's positions further. Changing W changes only the striding, never the global order, so every sample is seen exactly once per epoch at any GPU count. When W divides P, each rank reads from P/W partitions, which keeps reads local to few fragments.
• Mid-epoch resume. state_dict() returns {format: 1, tag, manifest_version, seed, epoch, P, consumed}, where consumed counts samples handed to the trainer (prefetched batches are not counted), and load_state_dict() continues from it on any world size; it fits a stateful DataLoader's checkpoint hook (e.g. torchdata's StatefulDataLoader; verify).
• Uneven input. With "pad" and "drop", every rank runs the same number of steps per epoch, so a synchronous DDP collective never waits on a rank that has already finished. uneven= chooses how the epoch's tail is handled: "pad" (default) fills the short ranks' last batches with samples repeated from the start of the epoch order, flagged by a boolean _pad column so the trainer masks them out of the loss, and consumed counts only unpadded samples; "drop" drops the tail that does not fill one step on every rank (as drop_last across ranks), so those samples are not seen that epoch; "none" leaves step counts uneven, for callers that wrap training in DDP's Join context manager or an equivalent.
• Columns. _source fields named in columns are decoded into tensors or strings; vector columns are yielded as tensors; a user transform runs per batch.
• JAX. The sampler and batch iterator are framework-neutral; loams.torch also yields NumPy batches for JAX input pipelines.

5.6 PySpark data source format("loams") (M2)Copy Anchor Link

A PySpark Python data source (pyspark.sql.datasource.DataSource, Spark 4) that runs unchanged on Apache Spark 4 and on Sail, which serves PySpark through Spark Connect and runs Python data sources in-process (verify the Sail version that supports them):

from loams.spark import LoamsDataSource
spark.dataSource.register(LoamsDataSource)

df = (spark.read.format("loams")
      .option("endpoint", "https://loams.internal:8080")
      .option("namespace", "ml").option("collection", "web_docs")
      .option("tag", "dedup-2026-09-25")
      .load())
(df.write.format("loams").mode("append")
   .option("endpoint", "https://loams.internal:8080")
   .option("namespace", "ml").option("collection", "web_docs_clean")
   .save())

Tables (M4) are read by Spark and Sail through Iceberg REST (§6.1), not through format("loams").

6. Spark and Sail (D55), Spice (D56)Copy Anchor Link

6.1 Spark and SailCopy Anchor Link

• Sail is a named Iceberg reader in the M4 gate (D55). Sail 0.7.1 writes Iceberg copy-on-write only and lists position deletes, equality deletes and deletion vectors as under construction. Loams's keyed tables carry deletion vectors (§03 §2.3), so M4 budgets an upstream contribution of DV and delete-file reads to Sail; the reader mirrors the DV writer M4 builds for iceberg-rust anyway.
• A table has one writer class (§03 §2.3): Spark and Sail may write tables that are not Loams link targets, never the keyed tables Loams's links maintain.
• Loams serves no Spark Connect endpoint (D42): Sail is the Spark Connect server; Loams is its source and sink.
• Sail's stateless workers with blocking shuffle to object storage and checkpointing (0.7) are the reference design for M6's distributed shuffle (§11 §3).

6.2 SpiceCopy Anchor Link

Spice (Apache-2.0 runtime; Spice.ai Enterprise proprietary) federates and accelerates data for agent apps on its own DataFusion fork. It is both a distribution channel and a competitor (risk 11).

• Flight SQL source (M1 gate). Spice's Flight SQL connector (beta) is a client in the M1.7 Flight SQL gate: a spicepod declares Loams collections as Flight SQL datasets, and the gate compares Spice's results with Loams's own, federated (no acceleration) and accelerated (Spice's Arrow accelerator). Spice's accelerators copy the data, so accelerated reads have Spice's refresh staleness, not Loams's consistency tokens. The namespace travels in the loams-namespace gRPC metadata (M1.6 W14); if Spice's connector cannot set custom metadata, the Flight SQL frontend also accepts namespace-qualified table names (verify in M1.7).

• Function names (D56). Loams's SQL search functions use Spice's names where they overlap:

  Function	Loams	Spice
  vector_search	Table function: dense or sparse top-k over a collection's vector (M1.2)	Vector search over an accelerated dataset
  text_search	Table function: BM25 over Tantivy (M1.2)	BM25 over Tantivy
  rrf	Reciprocal-rank fusion of two or more ranked table-function results (M1.2)	Reciprocal-rank fusion
  rerank	Reserved for M3's rerank stage	Model reranking
  hybrid_search	Loams's shorthand for filter + retrievers + fusion in one call (§05 §4); no Spice equivalent	—

  Argument lists follow Spice's where the semantics match and extend them with named arguments (a vector literal, k, a filter); signatures are fixed in the M1.2 plan (verify against Spice's documentation).

• Whether Spice's federation pushes table-function calls down to a Flight SQL source is unverified; if it does not, Loams's search reaches Spice users through Loams views or direct Flight SQL queries.

• datafusion-federation (Apache-2.0, from Spice) is a candidate for M4, pinned at 0.5.5, the last release on DataFusion 54 (§11 §1.2). The Spice runtime itself is never embedded (§7).

7. What Loams does not buildCopy Anchor Link

• Distributed Python compute. Ray and Spark/Sail schedule the work; Loams hands out scan plans and accepts DoPut writes.
• Dataframes. Polars, pandas and DuckDB consume Arrow; Loams ships no dataframe API.
• Training loaders beyond the thin adapter. loams.torch is a sampler and resume state on top of Lance's reader; no prefetching framework, caching layer, sample-format codec or streaming shard format of its own.
• GPU scheduling and model serving. Embedding backfills and batch inference run on the lab's Ray or Spark cluster; embed() links and the M3 rerank stage call external model endpoints (§09).
• A Spark Connect server, Lance or Iceberg forks, or engine plugins in other engines' processes beyond the Python data source and IO plugin above.

Rejected dependencies:

8. Gates and open questionsCopy Anchor Link

Risks: 11 (Spice, among others, competes for the same slot) and 21 (DataFusion/Arrow lockstep, mitigated by D51) in §12.

Open questions (§13):

9. SourcesCopy Anchor Link

• Spice: github.com/spiceai/spiceai (v2.3.2/Cargo.toml) · spiceai.org/blog/announcing-1.0-stable · spiceai.org/docs/reference/distributions · spice.ai/pricing · spiceai.org/docs/components/data-connectors · spiceai.org/docs/components/data-accelerators · spiceai.org/docs/features/search
• Sail: github.com/lakehq/sail (issue 2573; v0.7.1/docs/guide/sources/iceberg/features.md) · lakesail.com/blog/sail-0-7-blocking-shuffle-checkpoint · lakesail.com/product/spark-connect
• Ray: pypi.org/pypi/ray/json · pytorch.org/blog/pytorch-foundation-welcomes-ray-to-deliver-a-unified-open-source-ai-compute-stack · hpcwire.com/bigdatawire/2023/02/10/anyscale-bolsters-ray-the-super-scalable-framework-used-to-train-chatgpt · docs.ray.io/en/latest/data/api/doc/ray.data.Datasource.html · docs.ray.io/en/latest/data/api/doc/ray.data.Datasink.html · docs.ray.io/en/latest/data/working-with-llms.html · github.com/lance-format/lance-ray
• Daft: docs.daft.ai/en/stable/connectors/lance
• Polars: pola.rs/posts/announcing-polars-2 · docs.pola.rs/user-guide/misc/arrow · docs.pola.rs/api/python/stable/reference/api/polars.io.plugins.register_io_source.html · github.com/pola-rs/polars/pull/29413 · github.com/jonasdedden/polars-pylance
• PyTorch loaders: lance.org/integrations/pytorch · docs.mosaicml.com/projects/streaming/en/latest/distributed_training/elastic_determinism.html · docs.databricks.com/aws/en/archive/machine-learning/petastorm
• PySpark Python data sources: spark.apache.org/docs/latest/api/python/tutorial/sql/python_data_source.html
• TileDB: github.com/TileDB-Inc/TileDB · github.com/TileDB-Inc/tiledb-rs