csv-file — Manifest reference¶

Declarative YAML manifest for the csv-file connector type. Apply via lm apply -f <file>. Validate before apply with lm validate -f <file>.

1. Envelope¶

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: <kebab-case-name>     # required
  connector_type: csv-file    # required
  tags: [<tag>, ...]          # optional, merged with target.tags
spec:
  transport: { ... }
  parser: { ... }
  schedule: "<cron>"          # optional
  after_process: archive|delete|move|none  # required
  archive_path: <path>        # required when after_process=move
  mapping: { ... }            # optional — transform at ingestion (see §6)
  target:
    tags: [<tag>, ...]        # recommended (warning if absent)
    indexes: [{ columns: [<col>, ...], unique: <bool> }, ...]  # optional (see §5)

Connector manifests vs one-shot lm csv ingest

Transport-level Kotlin hooks (file_filter, pre_process, post_process) and max_file_size belong to the one-shot lm csv ingest path only — a manifest carrying them is refused at validate/apply, because scheduled runs would silently ignore them. Same for after_process: archive|delete|move on a local transport (see §4). The full split: capability table in FileSource — the full guide.

2. Transport block¶

Three transport kinds: local, sftp, s3. Each has its own required fields, enforced by validateCrossField.

Local¶

transport:
  kind: local
  path: /imports/*.csv        # glob OK

Glob limits: * and ? only, and only in the final filename segment — the directory part must be literal (/imports/*/x.csv never matches; there is no ** recursion). A plain path (no glob characters) that doesn't exist simply yields an empty run.

Legacy alias: pre-manifest connector configs may carry a top-level path: instead of a transport block — the runtime wraps it as {kind: local, path: ...}. Manifests always declare the transport block (it is required).

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:
  kind: sftp
  host: sftp.acme.io
  port: 22                    # optional, default 22
  user: lumnik
  path: /export/*.csv         # glob OK
  key_env: ACME_SFTP_KEY      # OR password_env, at least one required
  # password_env: ACME_SFTP_PWD

key_env and password_env reference an entry from lm secret or an OS env var (resolution: DB secret registry first, env var fallback).

S3 (and S3-compatible: MinIO, Cloudflare R2, GCS-S3-compat)¶

transport:
  kind: s3
  bucket: acme-exports         # required
  prefix: billing/             # optional key prefix filter, default ""
  region: eu-west-3            # optional, default eu-west-3
  endpoint: https://...        # optional, for non-AWS S3
  suffix: .csv                 # optional filename filter, default "" (no filter)
  access_key_env: AWS_ACCESS_KEY_ID   # optional, default AWS_ACCESS_KEY_ID
  secret_key_env: AWS_SECRET_ACCESS_KEY # optional, default AWS_SECRET_ACCESS_KEY
  archive_bucket: acme-archive # optional, for after_process=archive|move (default: source bucket)
  archive_prefix: archive/     # optional, default "archive/"

3. Parser block¶

parser:
  delimiter: ","                # optional, auto-detected if absent
  encoding: UTF-8               # optional, auto-detected if absent
  has_header: true              # default true
  bom_strip: true               # default true
  null_values: ["", "NULL", "null", "#N/A"]  # default as shown
  columns: [code, name, email]  # REQUIRED when has_header: false (names the columns); ignored otherwise

Auto-detection mechanics (per file, when the key is absent): encoding via ICU4J CharsetDetector (fallback UTF-8); delimiter by counting the candidates , ; TAB | over the first 8 KB — most frequent wins, fallback ,. An explicit key overrides only that aspect; the other stays auto-detected.

Chunking is fixed, not a knob

Local connector runs commit in fixed 500-row chunks; remote transports (sftp/s3) read one whole file per transaction. A chunk_size key is ignored (the validator warns). Very large remote files: split producer-side.

Malformed rows — per-row DLQ¶

A malformed row never fails the file: it is isolated to the dead-letter queue (reason MalformedSource) and the run continues. Inspect with lm dlq list / lm dlq get ID; the cursor counts consumed events, so a resume never re-emits already-dead-lettered rows.

4. Scheduling and lifecycle¶

schedule — optional cron expression. 5 or 6 fields supported (0 6 * * * = daily at 6am). Validated at apply time.
after_process — required. One of:
archive — relocate processed files (sftp: archive_path, default /archive/; s3: archive_bucket/archive_prefix, defaults source bucket + archive/)
delete — remove the file after success
move — move to archive_path (required when this mode is selected). At runtime archive and move are the same relocation action — move just makes archive_path mandatory.
none — leave the file in place (useful for testing)

Local transport: only none is valid in a manifest

The local scheduled reader (row-level glob cursor) has no per-file completion tracking — archive/delete/move never execute there, so the validator refuses them on a local connector rather than let files silently stay in place. sftp/s3 honor all four values; the one-shot lm csv ingest path honors them on every transport.

Crash-safety timing (sftp/s3): a file is archived/deleted only after a later chunk's committed cursor proves its rows landed — never inside the read that ingests it. A crash in between leaves the file in place; dedup skips re-ingestion and the action runs on the next read. At-least-once: the file and its rows are never lost together.

Zero-byte files are skipped by the sftp/s3 listings (and by the one-shot path) — they are never ingested and never archived.

5. Target block¶

target.tags enables RAG scope filtering — strongly recommended. The connector warns at validate-time if missing.

target.indexes declares Postgres indexes on the hub table — a list of { columns: [...], unique: <bool> } objects (unique optional, default false):

target:
  tags: [billing]
  indexes:
    - { columns: [email], unique: true }
    - { columns: [bpartner_id, dateordered] }

Each entry needs a non-empty columns list of strings; malformed entries fail validation with a per-entry error instead of being silently dropped.

Tables are created in the connector schema as connector.t_<name> (one per file target). A source column named id collides with the auto-generated id BIGSERIAL primary key — rename it in the mapping before ingesting.

Deleting the connector (lm connector delete) purges its connector.t_* tables and registry rows — the data goes with it.

6. Transform at ingestion (`spec.mapping`)¶

Legacy exports rarely arrive clean. An optional mapping block repairs them as they flow in — three hook tiers applied in order (dataset → row → cell) on every connector run before the rows are written. Dataset hooks are chunk-scoped by contract (they may rewrite the whole chunk, never more); the one row hook that is refused on csv-file runs is aggregator — its cross-chunk buffer would be silently dropped at every chunk boundary:

  mapping:
    dataset:
      - filter-where: { column: status, equals: ACTIVE }
    cell:
      - fix-mojibake: { columns: [name, city] }      # "BÃ©nÃ©dict" → "Bénédict"
      - parse-date:
          column: signup_date
          formats: ["dd/MM/yyyy", "yyyy-MM-dd"]      # legacy dates → real dates

Full parameter reference for all 33 built-in hooks, the Kotlin escape hatch included: Transform at ingestion. Note: hooks run on connector runs only — the one-shot lm csv ingest path does not apply them.

7. Full example¶

See lumnik-hub/src/test/resources/validator/csv-sftp-valid.yaml for the canonical SFTP example used in tests.

8. Validation rules summary¶

Rule	Severity	Path
`transport.kind` must be in `{local, sftp, s3}`	ERROR	`spec.transport.kind`
Local/SFTP require `path`; S3 requires `bucket`	ERROR	`spec.transport.<field>`
SFTP requires `host`, `user`	ERROR	`spec.transport.<field>`
SFTP requires `key_env` OR `password_env`	ERROR	`spec.transport`
`after_process=move` requires `archive_path`	ERROR	`spec.archive_path`
`after_process: archive\\|delete\\|move` refused on local transport (not executed on local runs)	ERROR	`spec.after_process`
Transport hooks (`file_filter`/`pre_process`/`post_process`) and `max_file_size` refused (one-shot-only)	ERROR	`spec.transport.<key>`
`parser.columns` required when `has_header: false`	ERROR	`spec.parser.columns`
Unknown key in `parser`/`target` (closed blocks — e.g. the dead `chunk_size` knob)	WARNING	`spec.parser.<key>`
`target.indexes` entries are `{ columns: [non-empty strings], unique?: bool }`	ERROR	`spec.target.indexes[i]`
`schedule` (if present) is a valid cron	ERROR	`spec.schedule`
`aggregator` row hook is refused (chunked runs would lose buffered rows)	ERROR	`spec.mapping.row[i].aggregator`
`target.tags` present	WARNING	`spec.target.tags`