[SDK-parquet] parquet sized buffer and gcs handler #602
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
This stack of pull requests is managed by Graphite. Learn more about stacking. |
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
3 times, most recently
from
a0cf889 to
1f2a615
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_default_processor_extractor_step
branch
from
9bff5f3 to
9c6dc91
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
2 times, most recently
from
0592350 to
4fc6f8c
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_default_processor_extractor_step
branch
from
472b366 to
daf3f6e
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
4fc6f8c to
0d05560
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_default_processor_extractor_step
branch
from
daf3f6e to
f77aaac
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
0d05560 to
9a4a1e2
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_default_processor_extractor_step
branch
from
f77aaac to
eb396de
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
9a4a1e2 to
2514fd7
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_default_processor_extractor_step
branch
3 times, most recently
from
2f91ffa to
2dfdb15
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
2 times, most recently
from
c2e7723 to
6f12cd4
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_default_processor_extractor_step
branch
from
2dfdb15 to
1bc8672
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
2 times, most recently
from
4d7551c to
c59ea51
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_default_processor_extractor_step
branch
2 times, most recently
from
ef311b0 to
8f43e54
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
c59ea51 to
f1ca3f9
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
f1ca3f9 to
4d56db7
Compare
dermanyang
reviewed
Nov 14, 2024
rust/sdk-processor/src/steps/parquet_default_processor/size_buffer.rs
Outdated
Show resolved
Hide resolved
dermanyang
reviewed
Nov 14, 2024
rust/sdk-processor/src/steps/parquet_default_processor/size_buffer.rs
Outdated
Show resolved
Hide resolved
rtso
reviewed
Nov 14, 2024
rust/sdk-processor/src/steps/parquet_default_processor/size_buffer.rs
Outdated
Show resolved
Hide resolved
rust/sdk-processor/src/steps/parquet_default_processor/size_buffer.rs
Outdated
Show resolved
Hide resolved
rust/sdk-processor/src/steps/parquet_default_processor/size_buffer.rs
Outdated
Show resolved
Hide resolved
rust/sdk-processor/src/steps/parquet_default_processor/size_buffer.rs
Outdated
Show resolved
Hide resolved
rust/sdk-processor/src/steps/parquet_default_processor/size_buffer.rs
Outdated
Show resolved
Hide resolved
rust/sdk-processor/src/steps/parquet_default_processor/size_buffer.rs
Outdated
Show resolved
Hide resolved
rust/sdk-processor/src/steps/parquet_default_processor/size_buffer.rs
Outdated
Show resolved
Hide resolved
|
|
||
| #[async_trait] | ||
| pub trait Uploadable { | ||
| async fn handle_buffer( |
There was a problem hiding this comment.
Should this be called upload_buffer? handle is very generic-sounding
There was a problem hiding this comment.
yeah, upload makes more sense!
There was a problem hiding this comment.
Doesn't look like this was addressed so opening back up!
There was a problem hiding this comment.
oops I think Imissed, updated!
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
2 times, most recently
from
d0a5380 to
946b726
Compare
dermanyang
reviewed
Nov 16, 2024
dermanyang
reviewed
Nov 16, 2024
|
|
||
| #[tokio::test] | ||
| #[allow(clippy::needless_return)] | ||
| async fn test_parquet_buffer_step_trigger_upload() { |
dermanyang
approved these changes
Nov 16, 2024
rtso
reviewed
Nov 18, 2024
| pub fn update_current_batch_metadata(&mut self, cur_batch_metadata: &TransactionMetadata) { | ||
| if let Some(buffer_metadata) = &mut self.current_batch_metadata { | ||
| // Update metadata fields with the current batch's end information | ||
| buffer_metadata.end_version = cur_batch_metadata.end_version; |
There was a problem hiding this comment.
Before we set, the new versions, should we add some validation that there are not gaps in data (buffer_metadata.end_version + 1 = cur_batch_metadata.start_version? It seems like everything is parsed in order 🤔
There was a problem hiding this comment.
yeah it shouldn't happen but it's good to add a safety here!
There was a problem hiding this comment.
Let's try to avoid panics if possible and return a Result
| { | ||
| internal_buffers: HashMap<ParquetTypeEnum, ParquetBuffer>, | ||
| pub poll_interval: Duration, | ||
| pub buffer_uploader: U, |
There was a problem hiding this comment.
Does buffer_uploader need to be a trait? It looks like the type will always be GCSUploader.
There was a problem hiding this comment.
right, we can remove the generics
Comment on lines
149
to
160
| // if it wasn't uploaded -> we update only end_version, size, and last timestamp | ||
| if file_uploaded { | ||
| if let Some(buffer_metadata) = &mut buffer.current_batch_metadata { | ||
| buffer_metadata.start_version = cur_batch_metadata.start_version; | ||
| buffer_metadata.start_transaction_timestamp = | ||
| cur_batch_metadata.start_transaction_timestamp.clone(); | ||
| } | ||
| } |
There was a problem hiding this comment.
Is this still necessary if we already did line 147?
There was a problem hiding this comment.
yeah, it's slightly different b/c here we are updating the start_version and start_trxn_timestamp, which only needs to be updated after we upload the file
There was a problem hiding this comment.
WOuldn't 2nd part of the function (lines 45-46) take care of updating after uploading the file too?
There was a problem hiding this comment.
oh right, if we set none after upload, it will be handled when we actually handle the current batch.
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_default_processor_extractor_step
branch
from
e939fee to
2198641
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
2 times, most recently
from
c7c2586 to
168ac53
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_default_processor_extractor_step
branch
2 times, most recently
from
4455b67 to
61ecc68
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
168ac53 to
40b32d4
Compare
rtso
approved these changes
Nov 18, 2024
| pub fn update_current_batch_metadata(&mut self, cur_batch_metadata: &TransactionMetadata) { | ||
| if let Some(buffer_metadata) = &mut self.current_batch_metadata { | ||
| // Update metadata fields with the current batch's end information | ||
| buffer_metadata.end_version = cur_batch_metadata.end_version; |
There was a problem hiding this comment.
Let's try to avoid panics if possible and return a Result
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
40b32d4 to
191f514
Compare
yuunlimm
changed the base branch from
11-12-_sdk-parquet_parquet_default_processor_extractor_step
to
graphite-base/602
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
191f514 to
55389e1
Compare
yuunlimm
force-pushed
the
graphite-base/602
branch
from
61ecc68 to
552ecc7
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
55389e1 to
6c2a18e
Compare
|
updated test with concrete GCSUploader instance, verified that it won't upload anything since data is empty. |
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
6c2a18e to
d8954bd
Compare
yuunlimm
force-pushed
the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
from
d8954bd to
b482e4e
Compare
yuunlimm
deleted the
11-12-_sdk-parquet_parquet_sized_buffer_and_gcs_handler
branch
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.
add a parquet specific step that stores parquet structs in a buffer and triggers uploads using gcs_handler which can handle any of parquet types defined in the enum