-
Notifications
You must be signed in to change notification settings - Fork 3.8k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
execinfra: explicitly pass the output RowReceiver to Processor.Run #98651
Conversation
482bcd2
to
8251d20
Compare
The first commit is in #98654, so it should be ignored here. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks for making this! It will be very helpful for multiple active portals.
I just reviewed the 3rd commit and it lgtm. Just left nit comments.
Would like to have the queries team to finalize the review as I'm not familiar with this part of the codebase.
Reviewed 2 of 3 files at r2, 78 of 78 files at r3, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @DrewKimball and @yuzefovich)
pkg/sql/execinfra/processorsbase.go
line 723 at r3 (raw file):
// Run is part of the Processor interface. func (pb *ProcessorBaseNoHelper) Run(ctx context.Context, output RowReceiver) { pb.self.Start(ctx)
nit: maybe we should still have a nil check for output
pkg/sql/flowinfra/flow.go
line 369 at r3 (raw file):
// SetProcessors overrides the current f.processors and f.outputs with the // provided slices. func (f *FlowBase) SetProcessors(
nit: might be more explicit if rename it to SetProcessorsAndOutputs
pkg/sql/flowinfra/flow.go
line 409 at r3 (raw file):
ctx context.Context, processors []execinfra.Processor, outputs []execinfra.RowReceiver, ) error { log.VEventf(
nit : do we want to assert len(processors) == len(outputs)
?
pkg/sql/flowinfra/flow.go
line 478 at r3 (raw file):
// Run is part of the Flow interface. func (f *FlowBase) Run(ctx context.Context) { defer f.Wait()
nit: ditto, assert the lengths are equal?
pkg/sql/rowexec/columnbackfiller.go
line 63 at r3 (raw file):
processorID int32, spec execinfrapb.BackfillerSpec, post *execinfrapb.PostProcessSpec,
nit: this post
arg is not used. Remove?
pkg/sql/rowexec/indexbackfiller.go
line 72 at r3 (raw file):
ctx context.Context, flowCtx *execinfra.FlowCtx, processorID int32,
nit: processorID
and post
are never used.
pkg/sql/rowexec/ordinality.go
line 42 at r3 (raw file):
flowCtx *execinfra.FlowCtx, processorID int32, spec *execinfrapb.OrdinalitySpec,
nit: spec
is never used
pkg/sql/rowexec/processors.go
line 117 at r3 (raw file):
) (execinfra.Processor, error) { if core.Noop != nil { if err := checkNumIn(inputs, 1); err != nil {
I'm not familiar with this part of code, but is this length check for input and output important? Do we need to implement something similar for the output in FlowBase.Run()
?
pkg/sql/rowflow/row_based_flow.go
line 205 at r3 (raw file):
} func (f *rowBasedFlow) makeProcessor(
nit : rename it to makeProcessorAndReceiver
?
Approved for backupccl |
This commit refactors the vectorized flow creator to directly embed two structs that implement interfaces used by the creator. This allows for those structs to not be allocated on the main code path. Also, `flowCreatorHelper` no longer implements the `Releasable` interface. Release note: None
This commit modifies how the output of a processor is plumbed through. Previously, we would pass it on the processor construction and store in the `ProcessorBaseNoHelper` to later be used when calling `execinfra.Run` method. However, the upcoming work on multiple active portals will require updating the output when resuming the flow, so this commit makes it so that we now pass the output explicitly into `execinfra.Processor.Run` method. This will make the work on multiple active portals cleaner, but also this change seems like a good one independent of that because we're able to store all outputs in the single place (`FlowBase.outputs`). Release note: None
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks for taking a look both! Addressed the comments so far.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @DrewKimball and @ZhouXing19)
pkg/sql/flowinfra/flow.go
line 478 at r3 (raw file):
Previously, ZhouXing19 (Jane Xing) wrote…
nit: ditto, assert the lengths are equal?
I added the check to FlowBase.SetProcessorsAndOuputs
.
pkg/sql/rowexec/ordinality.go
line 42 at r3 (raw file):
Previously, ZhouXing19 (Jane Xing) wrote…
nit:
spec
is never used
I'll keep this one for consistency with other processors.
pkg/sql/rowexec/processors.go
line 117 at r3 (raw file):
Previously, ZhouXing19 (Jane Xing) wrote…
I'm not familiar with this part of code, but is this length check for input and output important? Do we need to implement something similar for the output in
FlowBase.Run()
?
I don't think we need to something like this in FlowBase.Run
. The idea here is that we perform this check on a per-processor basis, so it should only be done in NewProcessor
(since each processor can have different number of inputs). I simplified this function to only check the number of inputs since all processors have exactly one output, and that is now "encoded" into execinfra.Processor.Run
signature, so no need to check the number of outputs here.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Reviewed 3 of 3 files at r1, 3 of 3 files at r2, 78 of 78 files at r3, 78 of 78 files at r4, 78 of 78 files at r5, all commit messages.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @ZhouXing19)
TFTRs! bors r+ |
Build failed (retrying...): |
Build succeeded: |
colflow: simplify the vectorized flow creator a bit
This commit refactors the vectorized flow creator to directly embed two
structs that implement interfaces used by the creator. This allows for
those structs to not be allocated on the main code path. Also,
flowCreatorHelper
no longer implements theReleasable
interface.Release note: None
execinfra: explicitly pass the output RowReceiver to Processor.Run
This commit modifies how the output of a processor is plumbed through.
Previously, we would pass it on the processor construction and store in
the
ProcessorBaseNoHelper
to later be used when callingexecinfra.Run
method. However, the upcoming work on multiple activeportals will require updating the output when resuming the flow, so this
commit makes it so that we now pass the output explicitly into
execinfra.Processor.Run
method. This will make the work on multipleactive portals cleaner, but also this change seems like a good one
independent of that because we're able to store all outputs in the
single place (
FlowBase.outputs
).Epic: CRDB-17622
Release note: None