I/O: Add CSV file import, with transformations

[amotl]

About

Support CSV file imports with special needs.

Poem

I sniff the CSV, a ribboned trail,
Pipes and separators wag their tail,
Polars hum, macro-steps unfold,
Chunks hop into Crate, stories told.

References

• Import earth observation data from CSV with cells in JSON format crate-clients-tools#319

Backlog

• Other supported file formats, like Parquet.
• Documentation.

[coderabbitai]

Walkthrough

Added a CSV import adapter and wiring: new CSV loader module with URL parsing, batching, optional MacroPipe transformations, router dispatch, tests/fixtures, model/util adjustments, pyproject/CI tweaks, and changelog entry. No API-breaking changes to existing public functions beyond adding TableAddress.if_exists.

Changes

Cohort / File(s)	Summary
CSV I/O Implementation cratedb_toolkit/io/file/csv.py	New module providing CsvFileAddress (URL parsing, storage options, load_table) and from_csv() which calls polars_to_cratedb() with batch handling and S3 auth retry logic.
Router Integration cratedb_toolkit/io/router.py	IoRouter.load_table() detects CSV-like sources (scheme/path) and dispatches to from_csv() after adjusting the URL.
Model Update cratedb_toolkit/model.py	Added TableAddress.if_exists: Optional[str] and DatabaseAddress.with_table_address(table_address) to produce a DB address for a specific table (injects if-exists when set).
I/O Utility cratedb_toolkit/io/util.py	polars_to_cratedb() now parses target_url, pops if-exists from query params, and builds DatabaseAddress from the cleaned URL so if-exists is applied to the first batch.
Cluster API cratedb_toolkit/cluster/core.py	StandaloneCluster.load_table now respects the target argument by constructing an address via with_table_address() when provided.
Tests & Fixtures tests/io/file/test_csv.py, tests/io/file/data/climate_ddl.sql	Added SQL fixture and integration tests for CSV import scenarios (local, pipeline, S3 skipped).
Project Config & Dependencies pyproject.toml	Added macropipe[geo]==0.0.0 to optional-dependencies.io-recipe; extended Ruff exclude to include examples.
Workflows .github/workflows/release-oci-full.yml, .github/workflows/release-oci-ingest.yml	Re-enabled pull_request trigger in both release workflows (uncommented/added).
Changelog CHANGES.md	Added Unreleased entry: “Added CSV file import, with transformations.”
Misc (small edits) cratedb_toolkit/util/database.py, cratedb_toolkit/testing/testcontainers/cratedb.py, tests/cluster/test_import.py	Minor control-flow/assignment tweaks and a TODO/comment update; adjusted a test log assertion string.

Sequence Diagram(s)

sequenceDiagram
    participant User as "User"
    participant Router as "IoRouter"
    participant CsvAddr as "CsvFileAddress"
    participant Polars as "Polars"
    participant Macro as "MacroPipe"
    participant Crate as "CrateDB"

    User->>Router: load_table(csv_source, target)
    Router->>CsvAddr: from_url(csv_source)
    CsvAddr->>CsvAddr: parse URI, extract batch/pipeline/storage opts
    CsvAddr->>Polars: scan_csv(path, sep, quote, storage_options)
    Polars-->>CsvAddr: LazyFrame
    alt pipeline present
        CsvAddr->>Macro: MacroPipe.from_recipes(pipeline)
        Macro->>Macro: apply(LazyFrame)
        Macro-->>CsvAddr: transformed LazyFrame
    end
    CsvAddr->>Polars: collect() / chunk
    Polars-->>CsvAddr: DataFrame chunk
    CsvAddr->>Crate: polars_to_cratedb(chunk, target_url, if_exists)
    Crate-->>CsvAddr: insert result
    CsvAddr-->>User: success/failure

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly related PRs

• I/O: Adapter for PostgreSQL #445 — touches StandaloneCluster.load_table; related changes to cluster load routing and ingestion paths.
• Chore: Refactor ctk.io package #679 — modifies router/cluster/model handling similarly; overlaps in CSV/IO dispatch and address handling.
• OCI: Start producing image cratedb-toolkit-ingest #508 — updates the same release workflow files; likely related CI trigger changes.

Suggested reviewers

• seut
• matriv

Poem

🐰 I sniff the CSV, a ribboned trail,

Pipes and separators wag their tail,
Polars hum, macro-steps unfold,
Chunks hop into Crate, stories told,
I nibble carrots, then leap — import complete! 🥕

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 65.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and specifically describes the main change: adding CSV file import functionality with transformations, which is well-supported by the raw summary showing new CSV module, dataclass, and utility functions.
Description check	✅ Passed	The PR description directly relates to the changeset, describing CSV file import support with transformations and referencing the relevant issue.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch io-files

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[amotl]

This is where Macropipe comes into play, providing a concisely configurable transformation unit to your ingress channel.

[amotl]

You can configure Macropipe by describing transformation steps through compact URL parameters.

&pipe=json_array_to_wkt_point:geo_location&pipe=python_to_json:data

That's two pipeline elements, processed sequentially:

• json_array_to_wkt_point:geo_location (docs)
• python_to_json:data (docs)