storage: use fueled output handles in sources #27814
Conversation
petrosagg
force-pushed
the
columnated-ingestion
branch
8 times, most recently
from
dd3f894
to
9cb5037
Compare
Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
petrosagg
force-pushed
the
columnated-ingestion
branch
from
9cb5037
to
90fd919
Compare
petrosagg
changed the title
|
MitigationsCompleting required mitigations increases Resilience Coverage.
Risk Summary:The risk score for this pull request is high at 80, indicating a significant chance of introducing a bug, especially since one of the modified files is a known bug hotspot. The repository's predicted bug trend is on the rise, although the observed trend has remained steady. It's important to note that, historically, pull requests with similar characteristics to this one are 106% more likely to cause a bug compared to the repository's baseline. The predictors driving this risk score are the sum of bug reports of the files affected and the change in executable lines of code. Note: The risk score is not based on semantic analysis but on historical predictors of bug occurrence in the repository. The attributes above were deemed the strongest predictors based on that history. Predictors and the score may change as the PR evolves in code, time, and review activity. Bug Hotspots:
|
| impl Region for SourceMessageRegion { | ||
| type Item = SourceMessage; | ||
|
|
||
| unsafe fn copy(&mut self, item: &Self::Item) -> Self::Item { |
There was a problem hiding this comment.
This crate is wildly unsafe
😨
(from the columnation docstring )
There was a problem hiding this comment.
heh, yeah.. fortunately we will soon migrate to flatcontainer which is safer
| }); | ||
| row_temp.push(c); | ||
| } | ||
| Ok(std::mem::take(&mut row_temp)) |
There was a problem hiding this comment.
this works because Vec::default() doesn't allocate, right?
|
Will wait for a Nightly run before merging https://buildkite.com/materialize/nightly/builds/8246 |
| @@ -13,6 +13,7 @@ workspace = true | |||
| differential-dataflow = "0.12.0" | |||
| either = "1" | |||
| lgalloc = "0.3" | |||
| columnation = { git = "https://github.com/frankmcsherry/columnation" } | |||
There was a problem hiding this comment.
Nit: Don't depend on it, but use timely::container::columnation.
| @@ -102,6 +106,62 @@ pub struct SourceMessage { | |||
| pub metadata: Row, | |||
| } | |||
|
|
|||
| mod columnation { | |||
Motivation
This PR makes it so that sources produce collections backed by
StackWrapper<T>timely containers as opposed toVec<T>containers. TheStackWrappercontainer stores its data into flat regions (currently backed byColumnation) which allow us to easily measure the heap size of the data produced so far. Currently these container survive until the reclocking boundary where they turn back to normalVec<T>containers in the mz scope. We will probably want to use those region allocated containers throughout the storage dataflows but I didn't want to implement a bigger change than required for this feature.Creating an output in an async operator with
StackWrappercontainer and the newly introducedAccountedStackBuildercontainer builder unlocks a.give_fueled()API which will automatically yield back to timely once certain amount of MBs have been emitted into the dataflow. The limit is currently set to 128MB.I have gone through the feature benchmark and confirmed that this change does not produce a performance regression for any of the ingestion workloads. You can find the results here https://buildkite.com/materialize/nightly/builds/8237
Closes #27211
Tips for reviewer
The PR is made of 5 simple commits that add
Columnationimplementations for relevant types and change the return type required bySourceRender::renderto be aCollectionthat usesStackWrappercontainers.Then there are 4 separate commits, one for each source type, that have source specific changes required to produce these stack-container-based collections.
The files that each commit touches are disjoint so this can also be reviewed using the full diff.
Checklist
$T ⇔ Proto$Tmapping (possibly in a backwards-incompatible way), then it is tagged with aT-protolabel.