Commits and parquet file sink #197
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
jacksonrnewhouse
force-pushed
the
commits_and_parquet_file_sink
branch
2 times, most recently
from
5941084 to
cdb52e6
Compare
Closed
mwylde
reviewed
Jul 18, 2023
arroyo-connectors/src/parquet.rs
Outdated
| connection_type: ConnectionType::Sink, | ||
| schema: schema | ||
| .map(|s| s.to_owned()) | ||
| .ok_or_else(|| anyhow!("No schema defined for SSE source"))?, |
| data_recovery: Vec<Self::DataRecovery>, | ||
| ) -> Result<()>; | ||
| async fn insert_record(&mut self, record: &Record<K, T>) -> Result<()>; | ||
| // TODO: figure out how to have the relevant vectors be of pointers across async boundaries. |
There was a problem hiding this comment.
they can be pointers across async bounderies so long as they are &mut (because &T is send only if T is sync, but &mut T is send if T is send).
|
|
||
| async fn on_close(&mut self, ctx: &mut crate::engine::Context<(), ()>) { | ||
| info!("waiting for commit message"); | ||
| if let Some(ControlMessage::Commit { epoch }) = ctx.control_rx.recv().await { |
There was a problem hiding this comment.
This logic assumes that the next control message on the queue is a commit message -- is that guaranteed?
There was a problem hiding this comment.
It is except for if the source data ends, which right now is only for test sources.
Merged
jacksonrnewhouse
force-pushed
the
commits_and_parquet_file_sink
branch
4 times, most recently
from
85184df to
8c147ce
Compare
|
_ |
mwylde
reviewed
Jul 19, 2023
…rquet, refactor writing code.
jacksonrnewhouse
force-pushed
the
commits_and_parquet_file_sink
branch
from
57a5eef to
eefcbe4
Compare
mwylde
approved these changes
Jul 20, 2023
| @@ -283,6 +283,26 @@ export function FormInner({ | |||
| </Stack> | |||
| </fieldset> | |||
| ); | |||
| } else if (values[key] > 0) { | |||
There was a problem hiding this comment.
I think this should be (values[key].properties?.length || 0) > 0 -- values[key] is an object and can't be (meaningfully) compared to a number
Merged
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This adds a parquet sink and introduces a committing phase during checkpointing. This allows for the support of exactly once sinks, the first of which is implemented to support writing Parquet to S3. Parquet on S3 allows for the production of typed parquet files to an S3 directory. The files are created using S3's multipart upload API, and we support the creation of parquet files across multiple checkpoints.
When restoring from checkpoints the subtask with id 0 will be responsible for finishing any files that were previously in-flight, as well as committing any files that were ready to be committed.
Right now we assume idempotency for the completion of multipart uploads, which seems to work. According to the S3 Glacier documentation, "Complete Multipart Upload is an idempotent operation. After your first successful complete multipart upload, if you call the operation again within a short period, the operation will succeed and return the same archive ID." However, the normal S3 docs say that trying to complete an already finished upload might return a 404: "Description: The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed."
I've manually tested the behavior under a variety of cases, including stopping it through the UI, uncontrolled stopping, and failing to log commits. In all cases it was able to resume writes without losing or duplicating data.