FileSource — the full guide¶

The csv-file connector ingests over 3 transports: local, SFTP, S3 (and S3-compatibles: MinIO/R2/GCS).

YAML structure (connector manifest)¶

What a csv-file connector manifest may legally contain. Transport hooks and max_file_size are deliberately absent — they belong to the one-shot surface below and are refused here (see the capability table):

spec:
  transport:
    kind: local | sftp | s3
    # kind-specific fields below — nothing else
  after_process: archive | delete | move | none   # spec level (none = leave in place)
                                                  # local transport: only `none` is accepted
  archive_path: /archive/                          # required when after_process=move
  # Timing (crash-safety, deliberate): on connector runs a file is archived/deleted only
  # AFTER the next chunk confirms its rows were committed — never inside the read that
  # ingests it. A crash in between leaves the file in place; the next run skips
  # re-ingestion (dedup) and applies the action then. At-least-once — the file and its
  # rows are never lost together.
  parser: { ... }
  mapping: { ... }             # optional dataset/row/cell hooks (aggregator refused)
  target: { tags: [...], indexes: [...] }
  schedule: "0 6 * * *"

Full key-by-key reference: csv-file — Manifest reference.

Scheduled connector vs one-shot `lm csv ingest`¶

Two ingestion surfaces share the transports but not the features — the validator refuses what a scheduled run would silently ignore:

Capability	Connector manifest (scheduled runs)	One-shot `lm csv ingest`
Transports	local, sftp, s3	local upload, sftp, s3
`schedule` (cron)	yes	no (one shot, by definition)
`mapping` hooks (dataset/row/cell)	yes (`aggregator` refused)	no — never applied
Transport hooks (`file_filter`/`pre_process`/`post_process`)	refused at validate/apply	yes
`max_file_size`	refused at validate/apply	yes (default 500 MB, over-limit aborts)
`after_process` on sftp/s3	yes (deferred until the cursor proves the commit)	yes (deferred until the ingest transaction commits)
`after_process` on local	only `none` — archive/delete/move refused	yes
Malformed rows	isolated per-row to the DLQ, run continues	abort on the first malformed row (whole ingest rolls back)
Target table	`connector.t_<name>` (auto-schema)	`ext.<name>` (hybrid JSONB + promoted columns)
Naming a multi-file source	that is what a manifest is — one connector over a glob	`name` + more than one matched file is refused (see below)

One-shot `lm csv ingest` — what actually happens¶

One transaction: sample → infer → write. The service samples the first rows (default 100), infers a schema, creates a dormant connector + endpoint (enabled: false, no schedule — inspect, promote, or delete before enabling anything), then writes all rows to a hybrid ext.<name> table (full row as JSONB + typed promoted columns).

Type inference (widest type across the sample; on conflict TEXT wins): integers → BIGINT, decimals → DECIMAL, true/false → BOOLEAN, everything else TEXT. Date-looking values deliberately stay TEXT in v1.
Reserved columns are never promoted (they already exist in every ext.* table): id, tenant_id, external_id, data, indexed_at, connector_id, endpoint_id. A source column with one of these names survives inside the JSONB data, just not as a typed column.
Abort, not DLQ: the first malformed row aborts the ingest and the transaction rolls back — nothing is half-written, and no source file is archived: an aborted run leaves every file where it was, so the whole batch can simply be re-run. Per-row DLQ isolation is a connector-run behavior.
Delimiter and encoding are auto-detected per file (same detectors as the connector path).
Remote (sftp/s3): each matching file is checked against max_file_size (default 500 MB, over-limit aborts), buffered to a temp file and run through the same pipeline; zero-byte files are skipped at listing. after_process (and post_process) run after the transaction commits, for every ingested file at once — the commit is the proof the rows landed, so a crash can never leave archived files whose rows were rolled back. If the archive step itself fails once the rows are committed, it is logged as a WARNING per file and the run still succeeds: the file simply stays on the source.

--name needs a single file

Each matched file registers its own dormant connector under the name you give, so a name over a glob that matches more than one file is refused before anything is downloaded — one name cannot label two connectors:

API error 400: name 'clients' would ingest 3 files into one dataset, which the one-shot
path cannot register: each file creates its own dormant connector under that name. Point
at a single file, or drop the name to get one dataset per file. To fuse a folder into one
dataset, declare a connector manifest with a glob path.

Mind that sftp://host/dir/ globs *.csv by default (--pattern), so the plain folder gesture is multi-file. Three ways forward:

one dataset per file — drop --name; each file gets ext.<filename> and its own dormant connector.
one specific file — point the URL at it (or narrow --pattern / --suffix), and --name works as documented.
the whole folder as one dataset — that is a connector manifest with a glob path: one connector, one table, and a schedule if you want it. See CSV file manifest.

Local transport¶

transport:
  kind: local
  path: /var/imports/*.csv

Glob semantics: * and ? only, matched against the filename segment — the directory part must be literal, and there is no ** recursion. No name is excluded implicitly (a *.csv glob is how you keep .tmp files out).

Back-compat: legacy (pre-manifest) connector configs with a top-level path: instead of a transport block are wrapped as {kind: local, path: ...} at runtime. Manifests must declare the transport block.

Drop-directory contract: names must grow lexically

On scheduled runs the local cursor resumes by the current file's position in the lexically sorted glob result. A file dropped later with a lexically earlier name sorts before the cursor and is treated as already ingested — it is never read. Name incoming files monotonically (date-prefixed: 2026-07-25-orders.csv); if your producer cannot guarantee that, use an sftp/s3 transport, whose cursor tracks files by modification time instead.

SFTP transport¶

transport:
  kind: sftp
  host: sftp.example.com
  port: 22
  user: lumnik
  key_env: ACME_SFTP_KEY            # resolved via SecretResolver (DB → env)
  # OR password_env: ACME_SFTP_PWD
  path: /export/*.csv
  strict_host_key: true
# spec level, next to transport:
after_process: move
archive_path: /archive/

after_process: archive on sftp relocates too (archive and move are the same runtime action); with archive the archive_path is optional and defaults to /archive/ — only move makes it mandatory. Zero-byte files are skipped at listing (sftp and s3 alike).

S3 transport¶

transport:
  kind: s3
  region: eu-west-3
  endpoint: https://...           # optional (MinIO/R2/GCS)
  bucket: my-bucket
  prefix: billing/
  suffix: .csv
  access_key_env: AWS_S3_ACCESS_KEY
  secret_key_env: AWS_S3_SECRET_KEY
  archive_bucket: my-bucket-archive    # destination when after_process=archive
  archive_prefix: billing-processed/
# spec level, next to transport:
after_process: archive   # relocates via archive_bucket/archive_prefix
                         # ('move' would additionally require spec.archive_path)

Kotlin hooks (3 extension points — one-shot `lm csv ingest` only)¶

Honest boundary: these hooks and max_file_size run only on the one-shot lm csv ingest path. A connector manifest carrying them is refused at validate/apply — scheduled connector runs would silently ignore them. Same for after_process: archive|delete|move on a local connector: the local scheduled reader has no per-file completion tracking, so the manifest is refused (none stays valid; sftp/s3 honor it).

transport:
  file_filter:  { kind: kotlin, script_ref: reference/skip-if-too-small.kt }
  pre_process:  { kind: kotlin, script_ref: reference/decompress-gzip.kt }
  post_process: { kind: kotlin, script_ref: reference/notify-webhook.kt }

3 reference hooks ship under META-INF/hooks/connectors/csv-file/reference/.

Signatures:

fun shouldProcess(file: FileDescriptor, source: Any?): Boolean   // file_filter
fun transform(file: FileDescriptor, input: InputStream): InputStream   // pre_process
fun postProcess(file: FileDescriptor, result: Any?)   // post_process

after_process accepts only archive | delete | move | none (validator). The Kotlin hooks above run in addition to the declared action — custom-kotlin is not an after_process value.

CLI¶

lm csv ingest <local-file>
lm csv ingest sftp://user@host[:port]/path-or-glob --key-secret X | --password-secret X
lm csv ingest s3://bucket/prefix --region eu-west-3 [--endpoint X] [--suffix .csv]

Troubleshooting¶

Error	Cause	Fix
`sftp connect failed: auth`	key/password missing or invalid	`lm secret list` then `lm secret set ACME_SFTP_KEY ...`
`unable to verify host key`	strict_host_key=true and no known fingerprint	`strict_host_key: false` for a demo
`S3 403 Forbidden`	invalid S3 credentials	check `AWS_S3_ACCESS_KEY` / `AWS_S3_SECRET_KEY`
`file exceeds max_file_size`	file > 500 MB	raise `max_file_size` or split producer-side
`Hook script not found`	script_ref points to a missing file	check `META-INF/hooks/connectors/csv-file/<script_ref>`