pathwaycom / pathwaycom/pathway
`pw.io.s3` needs a `write` counterpart
Nobody has claimed this yet.
- Dominant language
- Python
- Stars
- 62.3k
- Forks
- 1.7k
- PR merge metrics
- No merged PRs in 30d
Description
Is your feature request related to a problem? Please describe.
pw.io.s3 and pw.io.minio currently expose only a read method. There is no native way to write Pathway pipeline output back to S3 or MinIO, forcing users to implement custom Python output connectors which bypass engine-level batching and reliability guarantees.
Unlike a local filesystem, S3 and MinIO do not support appending to existing objects — every write must be a complete new object. This rules out the row-by-row write strategy used by file system connectors and requires a dedicated batching approach.
Describe the solution you'd like
Add pw.io.s3.write and pw.io.minio.write, implemented in Rust, following the same connector patterns as the existing output connectors.
API:
pw.io.s3.write(
table,
path, # S3 prefix / directory to write into
format, # "csv" | "json" | ...
*,
write_interval_ms, # how often to flush a new object, in milliseconds
name=None,
)
pw.io.minio.write(
table,
path,
format,
*,
write_interval_ms,
name=None,
)
Implementation:
The connector is split into a formatter and a writer, consistent with the existing connector architecture:
- The formatter serializes incoming rows into the chosen format (CSV, JSON, etc.), producing a byte string per row.
- The writer appends each formatted row to an in-memory buffer. The buffer is never written to S3/MinIO directly during normal row processing.
Flushing works as follows: when the engine calls flush on the write connector, the writer checks whether now - last_write_timestamp >= write_interval. If yes, it uploads the buffer as a new S3/MinIO object (with a timestamped or sequenced key under the configured prefix), and on success updates last_write_timestamp and clears the buffer. If no, the flush is a no-op — the buffer is retained as-is and last_write_timestamp is not updated, so the data will be included in the next successful flush. This ensures no data is lost between flush cycles and that each uploaded object is a complete, self-contained chunk.
Object naming within the prefix should be deterministic and monotonically increasing (e.g. using the flush timestamp or a sequence number) so that pw.io.s3.read / pw.io.minio.read can consume the output directory correctly.
Describe alternatives you've considered
Uploading one object per row would generate excessive S3 API call costs and small-object overhead. Streaming via multipart upload could allow appending within a single object, but adds significant complexity and still requires a final CompleteMultipartUpload call, making it unsuitable for a continuously running pipeline where the "end" of a file is never reached.
Additional context
Testing: integration tests should follow the existing pattern for S3/MinIO connectors and use a MinIO container (which is S3-compatible) in the Docker Compose test suite. Coverage should include:
- Basic write and read-back: write rows via
pw.io.s3.write, read back viapw.io.s3.read, assert equality. - Flush timing: verify that no object is written before
write_intervalelapses and that one is written shortly after. - Buffer retention on early flush: verify that rows are not lost when a flush occurs before
write_intervaland are included in the next successful flush. - Multiple flush cycles: verify that successive objects under the prefix are all readable by
pw.io.s3.readand that the full dataset is correct. - Format coverage: at minimum CSV and JSON.
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start by reading the existing S3 and MinIO read connectors and output-connector patterns, then inspect the Docker Compose integration-test setup using MinIO. Implement the formatter, buffered writer, interval-based flush behavior, and deterministic object naming described in the issue. Done means CSV and JSON write/read-back, timing, buffer retention, and multiple-flush tests pass.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- aws, python, rust
- Domain
- cloud, data-engineering
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 38/100