feat(connectors): add OpenDAL sink - #4123
George-Miao wants to merge 2 commits into
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #4123 +/- ##
============================================
+ Coverage 86.03% 86.06% +0.02%
Complexity 1449 1449
============================================
Files 1248 1250 +2
Lines 195841 196099 +258
Branches 161139 161397 +258
============================================
+ Hits 168501 168776 +275
+ Misses 23206 23164 -42
- Partials 4134 4159 +25
🚀 New features to boost your workflow:
|
| )); | ||
| } | ||
|
|
||
| opendal::init_default_registry(); |
There was a problem hiding this comment.
install_default calles init_default_registry() then install transport, so s3 would not fail. Maybe consider to add a test against minio container.
| opendal::init_default_registry(); | |
| opendal::install_default(); |
mattp5657
left a comment
mattp5657
left a comment
There was a problem hiding this comment.
2 things:
-
This connector is missing from the release/version-bump tooling. Absent from scripts/bump-version.sh, the connector_plugins default in _build_rust_artifacts.yml, and the connector list in edge-release.yml. It'll pass CI but won't ship in edge or tagged release artifacts, and a future version bump will silently skip it. Not unique to this PR (a handful of existing sinks are missing from these same lists too), but worth fixing here or in an immediate follow-up rather than adding a fourth thing to that pile.
-
I think we could use more unit and integration test coverage, happy and unhappy paths for every function. Untested today: path.rs's render_template and sanitize_path_segment beyond one edge case, lib.rs's consume()/write_message() unhappy path, the retry policy past one attempt, and S3/GCS/Azure end-to-end (fs is the only backend any test exercises)
| Ok(()) | ||
| } | ||
|
|
||
| async fn consume( |
There was a problem hiding this comment.
Should these be batched instead of one operator.write() per message?
s3_sink batches and flushes a combined object per batch, this sink
issues a separate write per message, so at the default batch_length = 100 that's 100 S3 PUTs instead of 1. Worth aligning with s3_sink's
approach for cost/throughput at scale?
| #[serde(default = "default_path_template")] | ||
| pub path_template: String, | ||
| #[serde(default)] | ||
| pub options: BTreeMap<String, String>, |
There was a problem hiding this comment.
Credentials (S3 access keys, etc.) flow
through options: BTreeMap<String, String> instead of SecretString,
diverging from this codebase's own stated convention for
credential-bearing config. A hand-rolled Debug impl redacts values in
one specific logging path but gives no type-level guarantee anywhere
else.
| let seconds = (micros / 1_000_000) as i64; | ||
| let nanoseconds = ((micros % 1_000_000) * 1_000) as u32; | ||
| DateTime::<Utc>::from_timestamp(seconds, nanoseconds).ok_or_else(|| { | ||
| Error::CannotStoreData(format!( |
There was a problem hiding this comment.
Should an out-of-range timestamp return Error::InvalidRecordValue
instead of Error::CannotStoreData?
3 participants
Which issue does this PR address?
Closes #4071
Rationale
OpenDAL provides an uniformed interface for writing to various services, which is useful for iggy to support multiple new sink types with a very small amount of change & maintainance effort.
What changed?
Added opendal-sink crate that implemnets a sink adapter for OpenDAL. I intentionally didn't touch the other part, e.g., CI. Those can be done in another follow-up PR.
Local Execution
AI Usage
Uses codex & omp for reviewing the changes & write some part of the doc.