Move

Warehouses

A Warehouse declares an upstream source system — the system of record pipelines extract rows from, plus the verified reachability and credential shape needed to reach it. Data in Layer is derived from a warehouse and reconstructible from it. The serving side is the VectorStore; the two sit on opposite sides of the gateway.

Supported Warehouses

Kind	Status	Source
`snowflake`	Shipped	Snowflake databases and tables, key-pair credential
`huggingface`	Shipped	Hugging Face Hub datasets, public or token-gated
`rss`	Shipped	Public RSS and Atom feeds
`youtube`	Shipped	Public channels, optional YouTube Data API key
`clouddrive`	Shipped	Microsoft 365 through Microsoft Graph: SharePoint and OneDrive
`rest`	Shipped	Any paginated JSON HTTP API
`databricks`	Reserved	Schema-reserved; rejected by the operator until implemented
`iceberg`	Reserved	Schema-reserved; rejected by the operator until implemented

Snowflake warehouses hold a key-pair credential. Hugging Face dataset warehouses can be public and credentialless, or can point at a read-token Secret for gated/private datasets. REST warehouses declare a JSON HTTP API — a base URL, optional auth, and a pagination rule — so any paginated JSON API is a source without a bespoke kind.

apiVersion: hevlayer.com/v1alpha1
kind: Warehouse
metadata:
  name: prod-snowflake
  namespace: layer
spec:
  kind: snowflake
  snowflake:
    account: acme-xy12345
    user: SVC_LAYER
    role: SVC_LAYER_ROLE
    warehouse: EXTRACT_WH
    keyPairSecretRef:
      name: snowflake-rsa
    pool:
      size: 5
      timeout: 30s
  verifyInterval: 1h

apiVersion: hevlayer.com/v1alpha1
kind: Warehouse
metadata:
  name: huggingface-hub
  namespace: hev-shop
spec:
  kind: huggingface
  huggingface:
    endpoint: https://huggingface.co
    # tokenSecretRef is optional for public datasets.
    # tokenSecretRef:
    #   name: hf-read          # key: token
  verifyInterval: 1h

apiVersion: hevlayer.com/v1alpha1
kind: Warehouse
metadata:
  name: openfda
  namespace: layer
spec:
  kind: rest
  rest:
    baseUrl: https://api.fda.gov
    userAgent: hevlayer/rest-source (+https://hevlayer.com/contact)
    # auth is optional — omit for open APIs.
    # auth:
    #   in: query              # query | header
    #   name: api_key          # the param/header carrying the credential
    #   secretRef:
    #     name: openfda-key    # key: token
    rateLimit:
      requestsPerSecond: 4
    verify:
      path: /drug/label.json   # a cheap GET the operator probes
      query:
        limit: "1"
  verifyInterval: 24h

Connection

spec.kind selects the source system. snowflake, huggingface, and rest select a source; databricks and iceberg are reserved by the schema but rejected by the operator until implemented. Exactly one kind block must be present and match spec.kind. verifyInterval (default 1h) sets the probe cadence for any kind.

A warehouse is identity and credential — not a catalog. What to read — a Snowflake database and table, a Hugging Face dataset and split — belongs to the pipeline source; one credential reaches many.

Snowflake

Field	Purpose
`kind`	`snowflake`, `huggingface`, `rss`, `youtube`, `rest`, or `clouddrive`. `databricks` and `iceberg` are reserved by the schema but rejected by the operator until implemented.
`snowflake.account`	Snowflake account identifier.
`snowflake.user`	Service user the key pair authenticates.
`snowflake.role`	Optional role assumed on connect.
`snowflake.warehouse`	Snowflake compute warehouse extraction queries run on.
`snowflake.keyPairSecretRef`	Secret in the same namespace holding `private-key.pem` and optional `passphrase`. The credential is never stored in the CRD.
`snowflake.pool`	Connection pool tuning: `size`, `timeout`.
`huggingface.endpoint`	Hub endpoint. Defaults to `https://huggingface.co`; override for Enterprise Hub or a mirror.
`huggingface.tokenSecretRef`	Optional Secret in the same namespace holding `token` for gated/private datasets. Omit for public datasets.
`verifyInterval`	Probe cadence. Defaults to `1h`.

A warehouse is source identity and credential — not a catalog. Which database, schema, table, dataset repo, config, or split to read belongs to the pipeline source; one credential can reach many tables or repos.

REST / HTTP JSON API

A rest warehouse declares a JSON HTTP API. The credential is optional — many public APIs need none, or take a key that only raises a rate limit (omit auth for the anonymous case, the same Verified-with-no-Secret shape as a public Hugging Face dataset).

Field	Purpose
`rest.baseUrl`	Required. API origin. A source’s `request.path` resolves against it.
`rest.userAgent`	User-Agent sent by the stock REST source worker. Defaults to `hevlayer/rest-source (+https://hevlayer.com/contact)`; set a project-specific contact URL for public crawls.
`rest.auth`	Optional. `in` (`query` or `header`), `name` (the param or header carrying the credential), and either `secretRef` (Secret holding a static `token`) or `login` (renewable token). Omit for open APIs.
`rest.rateLimit`	Optional. `requestsPerSecond` is a client-side request cap, to stay under the API’s fair-access ceiling. Omit to leave pacing to the source worker’s defaults.
`rest.verify`	Required. The reachability probe: a `path` and optional `query` the operator `GET`s to verify the API — and the credential, when `auth` is set.

rest holds access only; which endpoint, query, pagination, and field mapping to read belongs to the pipeline source — one warehouse serves many endpoints on the same API.

For APIs that issue short-lived tokens, use rest.auth.login instead of rest.auth.secretRef. The operator logs in before the verification GET; the stock REST source logs in at startup and renews before a request when the token is within 60 seconds of expiry. A 401 triggers one renewal and one retry. Refresh failure, or no refresh configuration, falls back to a new login.

rest:
  baseUrl: https://api.example.com
  auth:
    in: header
    name: Authorization
    prefix: "Bearer "
    login:
      path: /api/account/login
      body:
        email: "{{username}}"
        password: "{{password}}"
      secretRef:
        name: api-login
      tokenPath: $.accessToken
      expiresInPath: $.expiresIn
      refresh:
        path: /api/account/refresh
        body:
          refreshToken: "{{refreshToken}}"
  verify:
    path: /api/items

Create api-login in the Warehouse’s namespace with username and password keys. Login and refresh send POST requests with JSON bodies. Login body string values substitute {{key}} from Secret keys, including nested objects and arrays; JSON escaping is preserved. tokenPath is a JSONPath selecting one nonempty string. Optional expiresInPath selects a positive number of seconds. Without it, renewal happens on a 401. Refresh body templates use the login response’s top-level refreshToken; a refresh response may rotate that token.

Login and refresh paths resolve under baseUrl. Authenticated requests do not follow redirects. Authentication errors contain neither response bodies nor credentials; a failed operator login sets the Warehouse to Failed. Tokens are held in worker memory, while Secret keys remain projected files. Existing static auth is unchanged: auth.secretRef supplies the complete header or query value from its token key. prefix applies only to login tokens. Set exactly one of auth.login and auth.secretRef.

RSS and YouTube

An rss warehouse declares rss: {} with an optional userAgent (default hevlayer/rss-source (+https://hevlayer.com)). A youtube warehouse declares youtube: {} and optionally apiKeySecretRef: {name: youtube-key} inside that block; the Secret holds api-key. Public channel RSS and captions need no key. Feed URLs and channel selections belong to the Pipeline source.

These kinds verify their declaration and any required credential locally; workers check individual feeds and channels. The YouTube key, when present, is mounted and passed as apiKeyPath, never as credential text.

Cloud drives / Microsoft Graph

A clouddrive warehouse declares app-only Microsoft 365 access. The only provider is msgraph; drive selection belongs to the Pipeline source.

apiVersion: hevlayer.com/v1alpha1
kind: Warehouse
metadata:
  name: m365
spec:
  kind: clouddrive
  clouddrive:
    provider: msgraph
    msgraph:
      tenantId: 00000000-0000-0000-0000-000000000000
      clientId: 11111111-1111-1111-1111-111111111111
      credentialSecretRef:
        name: graph-credential
    rateLimit:
      requestsPerSecond: 8
      concurrentDownloads: 4
  verifyInterval: 1h

Use a same-namespace Secret containing certificate.pem and private-key.pem (the default), or client-secret. Upload the public certificate to the app registration and grant admin consent for the Graph application permission Sites.Read.All. If either certificate key is present, both are required; the operator never falls back to a client secret after a certificate error.

The operator exchanges client credentials for a token, signing an RS256 client assertion for certificate authentication, then calls Graph GET /v1.0/sites/root. A successful probe reports Verified and the token’s application permissions in status.roles. Failed conditions use AccessDenied for rejected credentials or missing consent/access grants; status.failureReason distinguishes these cases. Throttled includes the upstream Retry-After, when provided. Transport errors and other upstream failures use ProbeFailed.

The whole Secret is mounted at /var/run/hevlayer/warehouse/. Worker connection JSON in HEVLAYER_WAREHOUSE includes kind, provider, tenantId, clientId, optional rateLimit, and either certificatePath plus privateKeyPath, or clientSecretPath. It contains no credential material. Rotating the Secret re-verifies the warehouse, including changes between credential types.

Verification

The operator probes the warehouse on apply, whenever the referenced Secret’s content changes, and every verifyInterval.

For snowflake, the probe opens a key-pair session, runs SELECT 1 on the declared compute warehouse, and closes.

For huggingface, the probe calls the Hub dataset API with the optional token. If pipelines reference the warehouse, the operator resolves each declared repo@revision and records the resolved commit SHA in status.sourceRevisions. With no consumers yet, it verifies that the Hub dataset API is reachable.

For rest, the probe issues a single GET of rest.verify.path (with the optional query, and auth applied when set). Any 2xx is Verified, confirming both reachability and — when a credential is configured — that it is accepted.

Phase	Meaning
`Pending`	Not yet probed.
`Verified`	Last probe succeeded; `status.verifiedAt` is the probe time.
`Failed`	Last probe failed; `status.failureReason` says why.

Failed is a loud signal, not an outage: in-flight pipeline runs keep their connections, new runs refuse to start, and the condition surfaces in kubectl get warehouse and the dashboard. Pipelines start only against a Verified warehouse.

Rotation

Swap the referenced Secret’s content. The operator re-verifies and status.verifiedAt advances; consumers resolve credentials through the warehouse at connection-build time, so new connections pick up the new key with no redeploy. Pointing keyPairSecretRef or tokenSecretRef at a different Secret name is a spec edit with the same flow.

For huggingface, swapping the token Secret’s content re-verifies the same way. Adding or removing tokenSecretRef — moving a warehouse between anonymous and authenticated access — is a spec edit the operator re-verifies.

Pipeline source

A pipeline extracting from a warehouse names it in spec.sourceRef. The source block owns the what — for Snowflake a database, query, and cursor; for Hugging Face a dataset, split, and field mapping; for a cloud drive the sites, libraries, and folders — and the warehouse owns the where and who. The operator requires warehouseRef to name a Verified warehouse of the matching kind in the same namespace, and carries the source block verbatim to the worker as HEVLAYER_SOURCE_REF, as for any other source.

Snowflake

spec:
  sourceRef:
    kind: snowflake
    warehouseRef: prod-snowflake
    database: ANALYTICS
    query: >-
      SELECT ID, TITLE, BODY, REFRESH_ID FROM PUBLIC.NOTES
      WHERE REFRESH_ID > :cursor
    cursor:
      column: REFRESH_ID

The operator mounts the warehouse’s key-pair Secret into the worker pod at /var/run/hevlayer/warehouse/ and injects HEVLAYER_WAREHOUSE — connection JSON resolved from the warehouse spec (account, user, role, compute warehouse, pool), no credential material. The worker builds its own connection from the two.

Hugging Face

spec:
  target:
    namespace: squad
  sourceRef:
    kind: huggingface
    warehouseRef: hf-public
    dataset: rajpurkar/squad         # Hub repo id
    config: plain_text               # dataset config; omit for the default
    split: train                     # train | validation | test | …
    revision: ~                      # omit → the operator resolves and records the parquet-ref commit
    mapping:
      id: id                         # column → document id
      text: context                  # column to index or embed
      attributes: [title, question]  # columns to carry as attributes

`mapping` field	Purpose
`text`	Required. The column indexed, and embedded by a following stage.
`id`	Optional. Column used as the document id. Omitted, the worker synthesizes `{config}/{split}#{offset}` — stable within a revision; name a natural key for anything long-lived.
`attributes`	Optional. Columns carried as attributes. Omit or `[]` for every remaining scalar column; binary feature columns (image, audio) are skipped.

Set spec.worker.image to the stock Hugging Face source image from the mesh-account ECR registry, so no custom code is needed, or to your own image. The operator mounts the warehouse’s token Secret (when present) into the worker and injects HEVLAYER_WAREHOUSE with the endpoint only, no token. The worker streams rows from the dataset’s Parquet conversion at the pinned revision; the row offset is the cursor.

Omitting revision pins to the dataset’s current parquet-ref commit, which the operator resolves and records, so a long run never drifts onto a newer version mid-flight and re-enumeration is exact. Reading a newer version is a deliberate revision edit.

A dataset’s text column is often a whole document that must be split into chunks before it is embedded. Add a chunk block to the source to declare how — it is a pipeline-source feature, not specific to Hugging Face. See Chunking on the Pipelines page.

spec:
  sourceRef:
    kind: huggingface
    warehouseRef: huggingface-hub
    repo: McAuley-Lab/Amazon-Reviews-2023
    config: raw_meta_Electronics
    split: train
    revision: main
    cursor:
      field: parent_asin

When sourceRef.kind is huggingface, the operator requires warehouseRef to name a Verified Hugging Face warehouse in the same namespace and requires repo on the source block. It injects HEVLAYER_WAREHOUSE with the Hub endpoint and, when tokenSecretRef is set, mounts the Secret at /var/run/hevlayer/warehouse/token and adds tokenPath to the connection JSON. HEVLAYER_SOURCE_REF remains the verbatim source block and owns the dataset repo, config, split, revision, and cursor.

REST

spec:
  target:
    namespace: drug-labels
  sourceRef:
    kind: rest
    warehouseRef: openfda
    request:
      path: /drug/label.json
      query:                           # static query params
        search: 'openfda.product_type:"HUMAN PRESCRIPTION DRUG"'
    pagination:
      kind: offset
      pageSizeParam: limit             # required for offset
      offsetParam: skip                # required for offset
      pageSize: 1000                   # required for offset; must be > 0
    response:
      items: $.results                 # JSONPath to the record array on each page
    cursor:
      field: effective_time            # incremental + re-enumeration key (JSONPath per item)
    mapping:
      id: openfda.spl_set_id           # JSONPath into each item
      text: ~                          # omit when a chunk strategy supplies the section text
      attributes: [openfda.generic_name, openfda.brand_name, dea_schedule]

Source field	Purpose
`request.path` / `request.query`	The endpoint (resolved against `baseUrl`) and any static query params — the what.
`pagination`	How to walk pages. `offset` requires `pageSizeParam`, `offsetParam`, and a positive `pageSize`. `page` requires `pageParam`, `pageSizeParam`, and a positive `pageSize`; `startPage` defaults to 1 (0 is allowed). `searchAfter` requires `cursorParam`, `cursorPath`, and a positive `pageSize`; `pageSizeParam` is optional. `hasMorePath` optionally stops `searchAfter` enumeration when it resolves to `false`.
`response.items`	JSONPath to the record array on each page; `$` selects a bare top-level array.
`cursor.field`	JSONPath into each item, used as the incremental refresh key and the re-enumeration cursor.
`mapping`	JSONPath expressions for `id`, `text`, and `attributes` — the same mapping shape as Hugging Face, over JSON instead of Parquet columns.

Both offset and page can omit request.forEach to enumerate one partition. For example, this source reads a bare JSON array using page numbers:

sourceRef:
  kind: rest
  warehouseRef: orders
  request:
    path: /api/orders
  pagination:
    kind: page
    pageParam: PageNumber
    pageSizeParam: PageSize
    pageSize: 1000
    startPage: 1                       # optional; use 0 for zero-based APIs
    allowEmptySnapshot: false          # optional; fail closed by default
  response:
    items: "$"
  cursor:
    field: updated_at
  mapping:
    id: "{{id}}"

Page numbers increment by one; offsets increment by pageSize from zero. Both stop on a short or empty page. An empty first page fails unless pagination.allowEmptySnapshot: true explicitly permits an empty snapshot. Enable it only when an empty response reliably means the source has no rows: a successful empty full snapshot applies snapshot.absentRow.set to all previously present rows in that partition. A failed snapshot leaves rows and the manifest intact. Without forEach, snapshot.scope: forEachValue covers the single implicit partition.

For one origin with an editorial set of paths, add request.forEach. The default value source is a ConfigMap in the Pipeline’s namespace. select runs against the JSON stored at ref.key; each selected string is bound to as, substituted in request.path and request.query, and remains available while mapping rows.

spec:
  sourceRef:
    kind: rest
    warehouseRef: greenhouse
    request:
      forEach:
        values:
          from: configMap
          ref: { name: roles-companies, key: companies.json }
          select: "$[[email protected] == 'greenhouse'].token"
        as: token
      path: /v1/boards/{{token}}/jobs
      query: { content: "true" }
    pagination:
      kind: offset
      pageSizeParam: per_page
      offsetParam: page
      pageSize: 500
    response: { items: $.jobs }
    cursor: { field: updated_at }
    snapshot:
      mode: full
      scope: forEachValue
      absentRow:
        set: { open: false, closed_at: "{{runStartedAt}}" }
    mapping:
      id: "{{token}}:{{id}}"
      text: "{{title}}\n{{content}}"
      attributes:
        title: title
        company: "{{token}}"
        ats: "{{warehouseRef}}"
        department: departments[*].name
        posted_at: first_published
        open: true

To take the values from the API itself instead of a ConfigMap, set values.from: api. The worker pages a list endpoint of the same Warehouse (same auth, rate limit and robots rules) and runs select, a JSONPath applied to each list item (default $), to get one id per item. The endpoint is read again at the start of every sweep, so ids that appear upstream become partitions on the next sweep with no ConfigMap edit.

request:
  forEach:
    values:
      from: api
      path: /api/contracts
      query: { status: active }        # optional
      items: $.data                    # where the list is in each response
      select: $.contract.id            # id within each item
      pagination:                      # optional; defaults to the source's own
        kind: page
        pageParam: PageNumber
        pageSizeParam: PageSize
        pageSize: 100
        totalsHeader: X-Pagination
    as: contract
  path: /api/contracts/{{contract}}/jobs

For two-level fan-out, nest a forEach. The inner values.path and query may use the outer binding. Every (outer, inner) pair is one partition, with both bindings available to path, query, body and mapping. A nested partition is named by a JSON array of its values, such as ["C1","A"], in snapshot manifests and sweep reports.

request:
  forEach:
    values: { from: api, path: /api/contracts, items: $.data, select: $.id }
    as: contract
    forEach:
      values:
        from: api
        path: /api/contracts/{{contract}}/codes
        items: $.data
        select: $.code
        pagination: { kind: page, pageParam: p, pageSizeParam: s, pageSize: 100,
                      allowEmptySnapshot: true }   # a contract may have no codes
      as: code
  path: /api/contracts/{{contract}}/codes/{{code}}/jobs

Id list endpoints fail closed. If any list request errors, returns a non-2xx status, or reads short of its advertised total, the sweep stops before reading any partition: no row is written or tombstoned and the sweep report records the error. An empty list at any level is also a failure unless that level’s pagination.allowEmptySnapshot is true. Only the outermost level may use from: configMap. A partition that stops being listed is no longer swept; its existing rows are left as they were.

mapping.attributes accepts the existing list form or a name-to-expression object. Object values may be JSONPath expressions, template strings, or non-string JSON literals. {{token}} is the active forEach value; {{warehouseRef}} and {{runStartedAt}} are worker-provided values.

To index a number returned as a string, use an explicit cast in the attribute object. The cast resolves path as a JSONPath or template, then converts a numeric string or JSON number to a finite float. Missing and null values omit the attribute. Invalid strings, booleans, arrays, objects and non-finite numbers fail the partition before any row or snapshot is written; cast errors exclude the source value. Objects without cast remain literal attributes.

mapping:
  id: "{{id}}"
  attributes:
    ki_estimate_total: { path: $.total, cast: float }
schema:
  ki_estimate_total: { type: float, filterable: true }

float is the supported cast. A cast object requires exactly path and cast. Numeric MCP filters use {"ki_estimate_total":{"min":100,"max":200}}, which becomes inclusive Gte and Lte predicates in both search and aggregate. Declare the numeric field in the target schema; if the MCP namespace has an explicit filter allowlist, add the field there too.

List/detail APIs can use nested API fan-out: enumerate projects, enumerate estimate ids per project, then fetch /api/estimates/{{estimate}} as the partition request. Use response.items: $ for a detail object (or the path to its wrapped object) and a page size greater than one with page/offset pagination to finish after that single object. Both bindings remain available in mappings, so the row can preserve its lead id alongside the estimate id. All calls share the Warehouse auth and rate limit. This shape fetches every listed detail on every sweep; unchanged mapped rows avoid replacement writes. A failed detail request leaves that partition’s previous rows and snapshot intact. Removed fan-out partitions retain their rows; this is not a global absence reconciliation.

For an ordered incremental projection, omit request.forEach and the full-snapshot block. This Travels with Charlie source follows {data, next_cursor, has_more} pages and sends the stored cursor back as since:

spec:
  target:
    namespace: family-messages
  sourceRef:
    kind: rest
    warehouseRef: travels-with-charlie
    request:
      path: /api/layer/messages
    pagination:
      kind: searchAfter
      cursorParam: since
      cursorPath: $.next_cursor
      hasMorePath: $.has_more
      pageSizeParam: limit
      pageSize: 100
    response:
      items: $.data
    cursor:
      field: $.received_at
    mapping:
      id: $.id
      text: $.text
      attributes: [received_at, sender, subject]

The first run starts without since and follows echoed cursors until next_cursor is absent or null, or has_more is false. After successful row writes, the last cursor becomes the stream’s S3-backed high-water mark. The next refresh resumes from it, so previously committed rows are not emitted again. A non-advancing cursor while has_more remains true fails the stream; the existing 10,000-page guard also remains in force.

snapshot.mode: full means each selected value is a complete diff partition. After a partition has been read through its final page, rows from its prior S3 manifest that are absent now receive absentRow.set; the new manifest commits only after every row write succeeds. A transport error, non-2xx response, or an empty first page (unless pagination.allowEmptySnapshot: true) leaves that partition’s manifest and rows intact. This failure rule is deliberate: an unavailable or renamed endpoint is not evidence that all of its rows disappeared.

Totals in a header

Some APIs return a bare array and report totals in a response header. KI’s X-Pagination carries JSON such as {"TotalCount":2301,"PageSize":500,"TotalPages":5,"HasNext":true}. Name the header with pagination.totalsHeader on an offset or page source:

pagination:
  kind: page
  pageParam: PageNumber
  pageSizeParam: PageSize
  pageSize: 500
  totalsHeader: X-Pagination
response:
  items: "$"

With the header set, the walk ends on HasNext: false, on TotalPages, or when TotalCount is reached, instead of probing for a short page. Keys are matched case-insensitively and any one of the three is enough. The header also makes the snapshot checkable: if the rows read do not add up to TotalCount, the partition fails and nothing is tombstoned. A missing or malformed header fails the partition too. TotalCount: 0 counts as an empty snapshot without allowEmptySnapshot.

Staying in sync

A REST source reads once per pod start by default. Set sync.interval to keep it running:

sourceRef:
  kind: rest
  warehouseRef: orders
  sync:
    interval: 15m          # 60s minimum; units s, m, h, d

The worker then re-enumerates on a loop, with the interval measured from the end of the previous sweep, so a slow sweep never overlaps the next. Each sweep is a full enumeration of every partition:

Adds are rows with an id not in the previous manifest.
Edits are detected by content hash of the mapped row, so an endpoint without a modified-since cursor still notices changes. An unchanged row is not rewritten, which keeps attributes that Functions have added to it.
Deletes are rows the API no longer returns. They receive snapshot.absentRow.set, as with any full snapshot, and are restored if they come back.
Failures leave rows and manifests alone and the loop continues. A login token that expires mid-sweep is renewed once and the request retried.

The worker is read-only toward the API: it sends only GET requests. It writes only to the target namespace through the gateway.

Do not combine sync with spec.schedule; the worker exits with an error. Use scaling.mode: fixed with one replica so a single worker owns the manifests.

Each sweep overwrites pipelines/{pipelineId}/source-snapshots/rest/report.json in the snapshot bucket, like the cloud drive report:

Field	Contents
`startedAt`, `finishedAt`, `elapsedMs`	When the last sweep ran and how long it took.
`status`	`succeeded`, `partial` (some partitions failed), or `failed`.
`intervalSeconds`	The configured `sync.interval`.
`rows`, `upserted`, `refreshed`, `tombstoned`	Rows present, written, touched on volatile fields only, and tombstoned by the sweep.
`partitions`	One entry per partition with the same counts, `status` and `error`.
`errors`	Every error message from the sweep.
`lastSuccessAt`, `consecutiveFailures`	The last fully successful sweep, and failed sweeps since.

Set spec.worker.image to the stock REST source image from the mesh-account ECR registry, which pages any JSON API by these rules, so there is no per-API worker. The operator injects HEVLAYER_WAREHOUSE with the baseUrl (and mounts the auth Secret when set); HEVLAYER_SOURCE_REF is the verbatim source block. A text field that is a whole document is split by a chunk block, as for any source; the section strategy with sectionSource: jsonFields makes each top-level JSON field its own section.

Cloud drives

A clouddrive source reads SharePoint document libraries and OneDrive for Business drives through a clouddrive warehouse. The stock clouddrive-source worker writes one metadata row per file and never downloads file contents. PDFs and photos from the same drives are read by the content workers described below.

spec:
  target:
    namespace: m365-files
  sourceRef:
    kind: clouddrive
    warehouseRef: m365
    drives:
      sites:
        - site: contoso.sharepoint.com:/sites/Projects
          library: Documents
          root: /Contracts               # omit for the whole library
      personal:
        mode: none                       # none | allowlist | all
    include:
      mimeTypes: [application/pdf, "image/*"]
      maxBytes: 536870912
    mapping:
      id: id
      version: file.hashes.quickXorHash
  schedule:
    cron: "0 * * * *"
    leaseSeconds: 3300
  worker:
    image: <acct>.dkr.ecr.us-east-1.amazonaws.com/clouddrive-source:<version>
  scaling:
    replicas:
      min: 0
      max: 1

Source field	Default	Purpose
`warehouseRef`	required	A `Verified` `clouddrive` warehouse in the same namespace.
`drives.sites[].site`	required	The SharePoint site as a Graph site address, `hostname:/sites/Name`. The worker resolves it with Graph `GET /sites/{site}`.
`drives.sites[].library`	required	The document library’s name. It must match exactly one drive on the site. Declare each site and library pair once.
`drives.sites[].root`	the whole library	An absolute folder path in the library, such as `/Contracts`. Only files at or under it become rows. `.` and `..` segments are rejected.
`drives.sites[].roots`	none	More absolute folders of this same library, selected next to `root`; a file is kept when it sits at or under any of them. Use it when the wanted folders are not under one parent, instead of declaring the library twice or widening to the whole library. They never apply to another library, unlike `include.paths`. Matching is case-insensitive. Adding a root re-walks that library once; settled content is not downloaded or parsed again.
`drives.personal.mode`	`none`	OneDrive for Business drives to read: `none`, `allowlist`, or `all`.
`drives.personal.users`	`[]`	User principal names to read. Required with `allowlist` and rejected with the other modes.
`drives.personal.exclude`	`[]`	User principal names to skip. Accepted only with `all`.
`include.mimeTypes`	every type	MIME types to keep, as `type/subtype` or `type/*`.
`include.maxBytes`	no limit	Files larger than this are skipped. Must be positive.
`include.paths`	whole drive root	Folder prefixes, absolute drive paths such as `/Scans`. List several to select several folders of one library. Matching is case-insensitive, as in SharePoint. A file is kept when it sits at or under any of them. Others are counted as `outsidePaths`.
`exclude.items`	none	Exact items to leave out of the selected set, as `[{driveId, itemId}]` pairs. Reaches single files inside a folder you otherwise keep. Counted as `excluded` only when every other filter selected the file, so `items = rows + skipped` shows what the list removed. A file already published and later excluded loses its rows at the next reconcile; add only items you intend to remove. Keep the ids in the live resource, not in git.
`exclude.itemsFile`	none	An absolute path to a mounted file of more items to leave out, one `<driveId> <itemId>` per line (`#` comments and blank lines ignored). Read at the start of every sync, so it is not limited by the resource size and keeps ids out of the resource. An unreadable or malformed file fails that drive’s sync.
`include.items`	none	Exact `[{driveId, itemId}]` files to select, even outside `root`, `roots` and `include.paths`. Size, MIME and `exclude` still apply.
`include.itemsFile`	none	Mounted file of more exact items, same format as `exclude.itemsFile`.
`include.itemsOnly`	`false`	Select exactly `include.items` and `include.itemsFile` and nothing else: the folder rules are not applied, so no sibling in a large folder is admitted. Requires a set. The library is still walked; only the listed files become rows. Skipped files are counted as `notInItems`.
`retention.deselected`	`delete`	What happens to a row already published when selection narrows (a dropped root, `itemsOnly`, an `exclude`). `delete` removes it at the next reconcile. `keep` keeps the row, and the documents built from it, while the file still exists in the drive; they are counted as `retained`, and the row is marked `clouddriveRetained: true` so consumers preserve it without doing new work (the extractor lists it, keeps its published pages and manifest state, and never downloads, parses, OCRs, stages or embeds it; a never-published retained file stays unprocessed). A file deleted from the drive, or without a content hash, is removed either way. New files are always judged by the current rules.
`mapping.id`	required	Must be `id`. Rows are keyed by the Graph item id.
`mapping.version`	required	Must be `file.hashes.quickXorHash`. The content version is Graph’s quickXorHash.
`mapping.attributes`	`[]`	Names from the metadata row list: `name`, `path`, `folder`, `mime`, `size`, `modifiedAt`, `webUrl`, `folderUrl`, `version`, `drive`, `fetch`, `provider`. Other names are rejected. The stock worker writes every metadata attribute whatever the list holds.

The source needs at least one site or a personal mode other than none, and unknown fields are rejected. Personal drives are found through Graph GET /sites/getAllSites, never by listing users, and personal drives take no root. An allowlisted user with no personal site fails the run.

The operator validates the block before it creates the worker and reports a problem on the Pipeline’s Ready condition:

Reason	Cause
`InvalidCloudDriveSource`	The source block fails the rules above.
`InvalidCloudDriveScaling`	`scaling.replicas.max` is not `1`, or `min` is above `1`. Each drive’s delta cursor has one writer.
`InsufficientGraphRoles`	A personal mode other than `none`, without `Sites.Read.All` or `Sites.ReadWrite.All` in the warehouse’s `status.roles`.
`WarehouseNotVerified`	The warehouse is not `Verified`, or its credential Secret changed and has not been verified again.
`CredentialNotFound`	The warehouse’s credential Secret cannot be read.

The operator mounts the warehouse’s credential Secret, injects HEVLAYER_WAREHOUSE as described under Cloud drives / Microsoft Graph, and rolls the worker Deployment with the Recreate strategy so two pods never share a cursor. Set spec.worker.image to the clouddrive-source image in the mesh-account ECR registry. The worker keeps its state in the chart’s snapshot bucket and exits at startup when the operator injects none.

The worker sweeps every selected drive once when its pod starts, then idles. For periodic sync, set spec.schedule with scaling.replicas.min: 0; the operator starts the pod again at each cron window. The window lasts leaseSeconds / 60 + 1 minutes, clamped between 5 and 55, so give it enough time for the first full enumeration.

Metadata rows

The worker writes one row per selected file to spec.target.namespace through the gateway. Rows have no vector, and folders are not rows. Use a dedicated target namespace for each Pipeline.

Attribute	Contents
`id`	The Graph drive item id.
`name`	The file name. Full-text searchable.
`path`	The path from the library root, such as `/Contracts/2026/lease.pdf`. Full-text searchable.
`folder`	The parent folder’s path, such as `/Contracts/2026`. Empty at the library root.
`mime`	Graph’s `file.mimeType`.
`size`	The size in bytes.
`version`	`quickXor:` followed by Graph’s quickXorHash. It changes when the bytes change.
`modifiedAt`	Graph’s `lastModifiedDateTime`.
`webUrl`	The file’s URL, verbatim from Graph’s delta feed. For some files this is an Office viewer URL rather than a direct file URL.
`folderUrl`	The parent folder’s URL, verbatim from Graph.
`drive`	JSON string naming the drive: `{"kind":"site","site":…,"library":…}` or `{"kind":"personal","site":…}`.
`fetch`	JSON string with `provider` (`msgraph`), `driveId`, and `itemId`.
`provider`	JSON string with Graph’s `image` and `photo` facets, `null` when absent.
`clouddrivePipeline`, `clouddriveDriveId`, `clouddriveGeneration`	Scope deletes and reconciliation to this Pipeline and drive.

drive, fetch, and provider are strings because some stores reject object attributes. Decode them as JSON when reading. Links always come from Graph; the worker never builds a URL from a path.

Items that do not become rows are counted by reason: folderOrNonFile, outsideRoot, outsidePaths, maxBytes, mimeType, and missingQuickXorHash (a file with no provider hash).

Delta sync and deletes

Each drive has a manifest in the snapshot bucket at pipelines/{pipelineId}/source-snapshots/clouddrive/{driveId}.json. It holds the Graph delta link, the drive’s item tree, and a fingerprint of each row.

First run. The worker walks the drive’s whole root/delta feed. root filters rows but does not narrow the walk, so the whole library is enumerated.
Later runs. The worker resumes from the saved delta link and reads only changes. It rewrites only rows whose attributes changed, so an unchanged run writes nothing.
Full walks. A drive is walked again from the start when Graph answers the delta link with 410 Gone, when warehouseRef, drives, include, mapping, or the target namespace changes, or when the previous run stopped between its writes and its cursor commit.
Deletes. A deleted file loses its row. A deleted folder removes the rows under it, except files moved out of it in the same delta feed. A file that stops matching root or include loses its row. After a full walk, the drive’s rows from any older walk are deleted.
Moves and renames. A moved or renamed file gets its new path, folder, webUrl, and folderUrl. Graph’s delta feed leaves out the contents of a renamed folder, so the worker fetches them again from Graph to keep their links current.
Commit order. Writes and deletes finish before the delta link is saved. Before its first write, the manifest records that a full walk is needed; a pod that stops mid-write is followed by a full walk that reconciles rows by generation.

A drive that answers 423 Locked is reported as locked and does not fail the sweep. Any other drive failure, or a delta chain that does not reach its final link, reports the drive as failed and leaves its cursor where it was. The other drives still sync, and the worker exits with an error after writing the report. A site or library that does not resolve fails the whole run.

Rate limits

Graph calls follow the warehouse’s rateLimit. requestsPerSecond (default 8) spaces every request from the worker, token requests included. concurrentDownloads (default 4) limits content downloads by the content workers. The budget is per worker pod, so Pipelines that share a warehouse each get their own.

On 429 or 503, the worker waits for Retry-After, in seconds or as an HTTP date. Without one, it waits 1, 2, 4, 8, and 16 seconds. The wait applies to every drive in the sweep. After five retries the request fails.

Reports

Each sweep writes a report to pipelines/{pipelineId}/source-snapshots/clouddrive/report.json in the snapshot bucket, with throttles (throttled responses), elapsedMs, and one entry per drive:

Field	Contents
`driveId`, `driveNumber`	The drive and its position in the sweep.
`status`, `error`	`succeeded`, `locked`, or `failed`, and the error for a drive that did not succeed.
`pages`, `deltaItems`	Delta pages read, and delta entries including repeats.
`items`	Items in the drive’s current tree, folders included.
`rows`, `upserted`, `deleted`	Current rows, and rows written and deleted by this sweep.
`skipped`	Items that are not rows, by reason.
`lastFullWalk`	When the drive was last walked in full.
`bytesDownloaded`	Always `0` for the metadata worker.

For a completed drive, items equals rows plus the skipped counts. Logs carry counts only, never file names, links, or credentials.

In an idle worker pod, clouddrive-source --once runs another sweep, --reset-cursors clears every drive’s delta link so the next sweep walks in full, and --verify checks each drive’s row count against its manifest and compares eight sampled webUrl and folderUrl values with Graph. Run these one at a time, never during a sweep.

PDFs from a drive

The layer-extract extract stage reads a clouddrive source directly: it lists the drives through Graph and downloads each PDF, instead of listing an object-store prefix. Page rows, the write stage, failure reasons, and the sweep report work as on PDF extraction; this section covers what differs.

spec:
  pipelineId: contracts
  target:
    namespace: contracts-pages
  sourceRef:
    kind: clouddrive
    warehouseRef: m365
    dedupe: content                    # content (default) | item
    drives:
      sites:
        - site: contoso.sharepoint.com:/sites/Projects
          library: Documents
          root: /Contracts
    include:
      mimeTypes: [application/pdf]
    mapping:
      id: id
      version: file.hashes.quickXorHash
    parse:
      timeoutSeconds: 120
    rescanSeconds: 3600
  worker:
    image: <acct>.dkr.ecr.us-east-1.amazonaws.com/mesh:layer-extract-<version>
  scaling:
    mode: fixed
    replicas:
      min: 1
      max: 1

The drive fields are the ones in the table above. include.mimeTypes defaults to [application/pdf]. To read scanned images as documents, list the image types in extract.imageMimeTypes (see Scanned images); include.mimeTypes then accepts application/pdf and those types only. Use include.paths to say which folders hold scans.

To read Word, Excel and PowerPoint files (.docx, .xlsx, .pptx, and the legacy .doc, .xls, .ppt), list their MIME types in include.mimeTypes. Any non-PDF document type turns on conversion to PDF inside the extract child, so pages, #page={n} links, locations and OCR work as they do for PDFs, and sets dedupe: item. Image types listed in extract.imageMimeTypes are converted the same way alongside them. Any other type is rejected when the worker starts. parse, cacheUri, and rescanSeconds mean what they do for an objects source; uri and suffixes are ignored. Run the write stage (args: [write]) beside it with the same pipelineId. The extract stage keeps its own drive listing in its cache location and does not need a metadata Pipeline.

`dedupe`	Documents
`content`	One document per quickXorHash. Copies of a file anywhere in the selected drives are downloaded, parsed, staged, and embedded once. The document id is the hash as 40 hex characters, so page ids are `{hash}#p{n}`.
`item`	One document per drive item. The document id is the first 32 hex characters of the sha256 of `clouddrive://{warehouseRef}/{driveId}/{itemId}`.

Content state lives under the cache location in a directory keyed to the gateway, the target namespace, and the dedupe mode. Changing any of them starts from empty state.

Page rows carry the drive’s copies of the document instead of an object key:

Attribute	Contents
`locations`	JSON string listing every copy, each with `drive`, `path`, `webUrl`, and `folderUrl`. Not filterable.
`path`, `name`	The first copy’s path from the library root, and its file name.
`source_modified`	The first copy’s Graph modification timestamp; cleared when the timestamp is absent.
`webUrl`, `folderUrl`	The first copy’s file and parent folder URLs, verbatim from Graph.
`pdf`, `source_uri`	The first copy’s `webUrl` with `#page={n}`.

Copies are ordered by drive id, then item id, so the first copy is stable from run to run.

Unchanged content is not downloaded, parsed, or embedded again. A second sweep over an unchanged drive reports bytes_downloaded: 0 and parsed: 0.
Moves, renames, and added or removed copies patch locations, path, name, webUrl, folderUrl, pdf, source_uri, and source_modified on the existing page rows. Nothing is downloaded, parsed, or embedded.
Deleting the last copy deletes the document’s page rows.
Editing a file gives it a new quickXorHash, so it becomes a new document. The old document’s pages are deleted once no copy has the old hash.

After the first sweep, a sweep starts only when every document in the pipeline is indexed. While the write or embed stage still has work, the sweep fails and runs again after the worker restarts, so an embed in flight cannot write back an old location.

A complete drive listing is authoritative: unlike an objects prefix, a listing with no PDFs left deletes their pages. A locked or failed drive fails the sweep before anything is deleted. After each download the worker reads the item again; if its hash or size changed, the document fails with source_error and is retried on the next sweep. Two items with the same hash and different sizes fail the sweep before any download.

The sweep report adds unique_contents, bytes_downloaded, bytes_converted and locations_updated; documents counts drive items. Reports and logs carry counts only, without document keys or links.

Photos from a drive

The stock clouddrive-images worker embeds photos from a clouddrive source. It syncs the selected drives as the metadata worker does, groups image files by quickXorHash, embeds each distinct image with CLIP, and writes one row per content to the target namespace.

spec:
  target:
    namespace: project-photos
  sourceRef:
    kind: clouddrive
    warehouseRef: m365
    drives:
      sites:
        - site: contoso.sharepoint.com:/sites/Projects
          library: Documents
          root: /Photos
    include:
      mimeTypes: ["image/*"]
      maxBytes: 536870912
    mapping:
      id: id
      version: file.hashes.quickXorHash
  schedule:
    cron: "0 * * * *"
    leaseSeconds: 3300
  worker:
    image: <acct>.dkr.ecr.us-east-1.amazonaws.com/clouddrive-images:<version>
    podSpec:
      containers:
        - name: worker
          resources:
            requests: {cpu: "2", memory: 4Gi}
            limits: {cpu: "4", memory: 8Gi}
  scaling:
    replicas:
      min: 0
      max: 1

include.mimeTypes defaults to ["image/*"], and every entry must be an image/ type. include.maxBytes also caps each download, at 512 MiB when unset. dedupe may be omitted or content; item is rejected. The worker exits at startup without a gateway key (LAYER_GATEWAY_API_KEY) or a snapshot bucket. Like the metadata worker, it runs once per pod start, so use spec.schedule for periodic runs.

The model is openai/clip-vit-base-patch32 at a pinned revision, with the weights in the image, so the worker needs no network access to Hugging Face. CLIP embeds the full decoded image after applying its EXIF orientation. A separate serving copy is re-encoded as JPEG, at most 2048 pixels on its longest side and under 6 MiB, and stored with the blob API. Images over 256 megapixels are rejected. Above about 89 megapixels, a JPEG is decoded at reduced resolution, and other formats are rejected.

Attribute	Contents
`id`	The sha256 of `version`, as hex. One row per distinct content.
`embed_image_url`	The normalized 512-dimensional CLIP vector, indexed for cosine distance.
`image_blob`	The blob reference for the serving JPEG.
`clip_model`	`openai/clip-vit-base-patch32`.
`locations`	JSON string listing every copy, each with `drive`, `driveId`, `itemId`, `path`, `webUrl`, and `folderUrl`. Not filterable.
`name`, `path`, `folder`, `mime`, `size`, `version`, `modifiedAt`, `webUrl`, `folderUrl`	The first copy’s metadata. Copies are ordered by drive id, then item id.
`provider`	JSON string with the first copy’s Graph `image` and `photo` facets. Not filterable.
`photo.<field>`, `image.<field>`	Each scalar field of those facets, such as `photo.takenDateTime`, `photo.cameraModel`, or `image.width`. Numbers are floats, except `image.width` and `image.height`, which are integers.
`clouddrivePipeline`, `clouddriveGeneration`	Scope reconciliation to this Pipeline and run.

Each run first syncs metadata into a durable stage, which commits before each drive’s cursor. The rest of the run needs a complete listing: if any drive is locked or failed, the run stops and existing rows stay. For each content without a cached result, the worker downloads one copy, reads the item again to confirm its hash, runs CLIP, uploads the serving JPEG, and caches the vector and blob reference in the snapshot bucket. It then writes every content row in batches of 100 under a new generation, deletes this Pipeline’s rows from older generations, and reads back every expected id and the row count.

Unchanged content is not downloaded again. An unchanged rerun downloads nothing and uploads no blobs, but still rewrites each row.
Moves, renames, and new copies rewrite the row from the cached result.
Deleting the last copy leaves the content out of the new generation, so the cleanup deletes its row.
A content that fails (a download, decode, or gateway error) fails the run. Cached results survive, so the next run downloads only what is missing.

State lives in the snapshot bucket under pipelines/{pipelineId}/image-content/{targetHash}/v1/, where targetHash identifies the gateway and target namespace: metadata.json is the metadata stage, artifacts/ holds the cached vectors and blob references, and report.json holds the run’s aggregates. A new target starts with no cached results. When the metadata stage is missing, the worker clears the drive cursors and walks every drive again.

Report field	Contents
`items`	Selected image files.
`uniqueContents`, `rows`	Distinct contents, and rows written.
`bytesDownloaded`, `blobsWritten`	Work done by this run.
`resizedContents`	Contents whose serving JPEG has smaller dimensions than the source image.
`likelyNearDuplicatePairs`	Pairs of distinct contents whose 64-bit difference hashes differ by at most four bits. The worker counts them and never merges them.

Variable	Default	Purpose
`CLIP_THREADS`	`4`	Torch threads per CLIP process.
`CLIP_WORKERS`	`1`	CLIP processes, each loading its own model. At most 4, and at most the warehouse’s `concurrentDownloads`. Keep workers times threads within the pod’s CPU limit.
`CLIP_DEVICE`	`cpu`	Torch device, for an image that supplies that runtime.

Downloads and processing run up to the warehouse’s concurrentDownloads at a time. In an idle pod, clouddrive-images --once runs another pass, and --reset-cursors walks every drive in full before reconciling content.

To search photos with text, rank by embed_image_url with an inline Embed on the same CLIP model. The gateway embeds the query with CLIP’s text tower, so it must have its local CLIP model configured:

{"rank_by": ["embed_image_url", "ANN", ["Embed", "kitchen cabinets", {"model": "openai/clip-vit-base-patch32"}]], "top_k": 10}

The worker declares image_url with a local image-modality CLIP profile (dims: 512) and supplies each vector itself, so the gateway doesn’t fetch the image again. The gateway must be a release that accepts precomputed vectors on a local image profile.

Keys

An ApiKey binds to a warehouse with a warehouse.<name> entitlement carrying a list of opaque claims strings. Layer stores and echoes the strings; the application routes on them. No client route reaches a source system — clients touch indexes, not warehouses — so the entitlement grants nothing in Layer, and it inerts when the warehouse is deleted.

Deletion

Deleting a warehouse fences everything drawing from it. A finalizer blocks deletion while status.consumers is non-zero — pipelines extracting from it or keys entitled to it — annotate with hevlayer.com/force-delete: "true" to override.

List Warehouses

The read-only warehouse API requires a Pro gateway connected to Kubernetes. It includes Secret reference names and keys, never Secret contents. All routes require a key with read scope. Create and edit connections by applying the resource or through the dashboard’s apply forms.

curl -H "Authorization: Bearer $LAYER_API_KEY" \
  "$LAYER_BASE_URL/v2/warehouses"

{
  "warehouses": [
    {
      "name": "prod-snowflake",
      "kind": "snowflake",
      "snowflake": {
        "account": "acme-xy12345",
        "user": "SVC_LAYER",
        "role": "SVC_LAYER_ROLE",
        "warehouse": "EXTRACT_WH",
        "keyPairSecretRef": {
          "name": "snowflake-rsa"
        },
        "pool": {
          "size": 5,
          "timeout": "30s"
        }
      },
      "verifyInterval": "1h",
      "status": {
        "phase": "Verified",
        "verifiedAt": "2026-06-10T00:00:00Z",
        "consumers": {
          "pipelines": 2,
          "apiKeys": 1
        },
        "conditions": []
      }
    }
  ]
}

GET /v2/warehouses/{name} returns one object in the same shape. phase is Pending, Verified, or Failed; failed warehouses include status.failureReason. status.consumers counts pipelines and API keys that still reference the warehouse.

Status

status:
  phase: Verified
  verifiedAt: "2026-06-10T00:00:00Z"
  failureReason: null
  sourceRevisions:
    McAuley-Lab/Amazon-Reviews-2023@main: 2b6d039ed471f2ba5fd2acb718bf33b0a7e5598e
  consumers:
    pipelines: 2
    apiKeys: 1

The operator emits Kubernetes Events on phase transitions and counts observed references in status.consumers.

Supported Warehouses

Connection

Snowflake

REST / HTTP JSON API

Login and renewable bearer tokens

RSS and YouTube

Cloud drives / Microsoft Graph

Verification

Rotation

Pipeline source

Snowflake

Hugging Face

REST

Totals in a header

Staying in sync

Cloud drives

Metadata rows

Delta sync and deletes

Rate limits

Reports

PDFs from a drive

Photos from a drive

Keys

Deletion

List Warehouses

Status