You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "jayzhan-synnada (via GitHub)" <gi...@apache.org> on 2024/12/18 08:24:41 UTC
[PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
jayzhan-synnada opened a new pull request, #13823:
URL: https://github.com/apache/datafusion/pull/13823
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #.
## Rationale for this change
Existing `execution_mode` is not enough the represent different combination of execution properties. There are different emission type like incrementally or emit at the end. And, we might have different decision based on the boundedness of the source and emission type, for example, whether it is pipeline breaking (not able to generate result if it has unbounded input and requires to emit at the end)
In this PR we introduce `emission_type` and `boundedness` to replace the existing `execution mode` to represent more execution properties.
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1890197827
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -488,95 +495,152 @@ impl ExecutionPlanProperties for &dyn ExecutionPlan {
self.properties().output_partitioning()
}
- fn execution_mode(&self) -> ExecutionMode {
- self.properties().execution_mode()
- }
-
fn output_ordering(&self) -> Option<&LexOrdering> {
self.properties().output_ordering()
}
+ fn boundedness(&self) -> Boundedness {
+ self.properties().boundedness
+ }
+
+ fn pipeline_behavior(&self) -> EmissionType {
+ self.properties().emission_type
+ }
+
fn equivalence_properties(&self) -> &EquivalenceProperties {
self.properties().equivalence_properties()
}
}
-/// Describes the execution mode of the result of calling
-/// [`ExecutionPlan::execute`] with respect to its size and behavior.
+/// Represents whether a stream of data **generated** by an operator is bounded (finite)
+/// or unbounded (infinite).
///
-/// The mode of the execution plan is determined by the mode of its input
-/// execution plans and the details of the operator itself. For example, a
-/// `FilterExec` operator will have the same execution mode as its input, but a
-/// `SortExec` operator may have a different execution mode than its input,
-/// depending on how the input stream is sorted.
+/// This is used to determine whether an execution plan will eventually complete
+/// processing all its data (bounded) or could potentially run forever (unbounded).
///
-/// There are three possible execution modes: `Bounded`, `Unbounded` and
-/// `PipelineBreaking`.
-#[derive(Clone, Copy, PartialEq, Debug)]
-pub enum ExecutionMode {
- /// The stream is bounded / finite.
- ///
- /// In this case the stream will eventually return `None` to indicate that
- /// there are no more records to process.
+/// For unbounded streams, it also tracks whether the operator requires finite memory
+/// to process the stream or if memory usage could grow unbounded.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Boundedness {
+ /// The data stream is bounded (finite) and will eventually complete
Bounded,
- /// The stream is unbounded / infinite.
- ///
- /// In this case, the stream will never be done (never return `None`),
- /// except in case of error.
- ///
- /// This mode is often used in "Steaming" use cases where data is
- /// incrementally processed as it arrives.
- ///
- /// Note that even though the operator generates an unbounded stream of
- /// results, it can execute with bounded memory and incrementally produces
- /// output.
- Unbounded,
- /// Some of the operator's input stream(s) are unbounded, but the operator
- /// cannot generate streaming results from these streaming inputs.
- ///
- /// In this case, the execution mode will be pipeline breaking, e.g. the
- /// operator requires unbounded memory to generate results. This
- /// information is used by the planner when performing sanity checks
- /// on plans processings unbounded data sources.
- PipelineBreaking,
+ /// The data stream is unbounded (infinite) and could run forever
+ Unbounded {
+ /// Whether this operator requires infinite memory to process the unbounded stream.
+ /// If false, the operator can process an infinite stream with bounded memory.
+ /// If true, memory usage may grow unbounded while processing the stream.
+ requires_infinite_memory: bool,
+ },
}
-impl ExecutionMode {
- /// Check whether the execution mode is unbounded or not.
+impl Boundedness {
pub fn is_unbounded(&self) -> bool {
- matches!(self, ExecutionMode::Unbounded)
+ matches!(self, Boundedness::Unbounded { .. })
}
+}
- /// Check whether the execution is pipeline friendly. If so, operator can
- /// execute safely.
- pub fn pipeline_friendly(&self) -> bool {
- matches!(self, ExecutionMode::Bounded | ExecutionMode::Unbounded)
- }
+/// Represents how an operator emits its output records.
+///
+/// This is used to determine whether an operator emits records incrementally as they arrive,
Review Comment:
Yes, you have the right idea. We will add some comments to clarify.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1890196572
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -488,95 +495,152 @@ impl ExecutionPlanProperties for &dyn ExecutionPlan {
self.properties().output_partitioning()
}
- fn execution_mode(&self) -> ExecutionMode {
- self.properties().execution_mode()
- }
-
fn output_ordering(&self) -> Option<&LexOrdering> {
self.properties().output_ordering()
}
+ fn boundedness(&self) -> Boundedness {
+ self.properties().boundedness
+ }
+
+ fn pipeline_behavior(&self) -> EmissionType {
+ self.properties().emission_type
+ }
+
fn equivalence_properties(&self) -> &EquivalenceProperties {
self.properties().equivalence_properties()
}
}
-/// Describes the execution mode of the result of calling
-/// [`ExecutionPlan::execute`] with respect to its size and behavior.
+/// Represents whether a stream of data **generated** by an operator is bounded (finite)
+/// or unbounded (infinite).
///
-/// The mode of the execution plan is determined by the mode of its input
-/// execution plans and the details of the operator itself. For example, a
-/// `FilterExec` operator will have the same execution mode as its input, but a
-/// `SortExec` operator may have a different execution mode than its input,
-/// depending on how the input stream is sorted.
+/// This is used to determine whether an execution plan will eventually complete
+/// processing all its data (bounded) or could potentially run forever (unbounded).
///
-/// There are three possible execution modes: `Bounded`, `Unbounded` and
-/// `PipelineBreaking`.
-#[derive(Clone, Copy, PartialEq, Debug)]
-pub enum ExecutionMode {
- /// The stream is bounded / finite.
- ///
- /// In this case the stream will eventually return `None` to indicate that
- /// there are no more records to process.
+/// For unbounded streams, it also tracks whether the operator requires finite memory
+/// to process the stream or if memory usage could grow unbounded.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Boundedness {
+ /// The data stream is bounded (finite) and will eventually complete
Bounded,
- /// The stream is unbounded / infinite.
- ///
- /// In this case, the stream will never be done (never return `None`),
- /// except in case of error.
- ///
- /// This mode is often used in "Steaming" use cases where data is
- /// incrementally processed as it arrives.
- ///
- /// Note that even though the operator generates an unbounded stream of
- /// results, it can execute with bounded memory and incrementally produces
- /// output.
- Unbounded,
- /// Some of the operator's input stream(s) are unbounded, but the operator
- /// cannot generate streaming results from these streaming inputs.
- ///
- /// In this case, the execution mode will be pipeline breaking, e.g. the
- /// operator requires unbounded memory to generate results. This
- /// information is used by the planner when performing sanity checks
- /// on plans processings unbounded data sources.
- PipelineBreaking,
+ /// The data stream is unbounded (infinite) and could run forever
+ Unbounded {
+ /// Whether this operator requires infinite memory to process the unbounded stream.
+ /// If false, the operator can process an infinite stream with bounded memory.
Review Comment:
Makes sense, we will add some examples. In your case, `MEDIAN` would be an example (but we can do `MIN`/`MAX` with bounded memory when sorted).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#issuecomment-2550959973
> Thanks for expanding docstrings in JoinType. Can you please move them to separate PR? This would help with review and merging.
Makes sense, we can merge quickly ahead of the main PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak merged PR #13823:
URL: https://github.com/apache/datafusion/pull/13823
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#issuecomment-2551270608
> So maybe we can figure out the FFI stuff, and then wait to merge this PR until 44 is relased. I will try hard to focus on the 44 release later this week
Sure, it makes sense to wait and only proceed with merging after all the PRs that are slated for 44 get merged before this.
One alternative to consider would be making 44 a major release with all the API changes (including this) so that we can proceed with a few "stable" releases after 44. It is also not a great dev-ex to deal with some API change in every single release, albeit small.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#issuecomment-2557107279
> One alternative to consider would be making 44 a major release with all the API changes (including this) so that we can proceed with a few "stable" releases after 44. It is also not a great dev-ex to deal with some API change in every single release, albeit small.
Indeed -- it may be time to consider making the API more stable. Here is a related discussion:
- https://github.com/apache/datafusion/issues/5269
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#issuecomment-2550715538
Thanks for expanding docstrings in JoinType.
Can you please move them to separate PR? This would help with review and merging.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1890196572
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -488,95 +495,152 @@ impl ExecutionPlanProperties for &dyn ExecutionPlan {
self.properties().output_partitioning()
}
- fn execution_mode(&self) -> ExecutionMode {
- self.properties().execution_mode()
- }
-
fn output_ordering(&self) -> Option<&LexOrdering> {
self.properties().output_ordering()
}
+ fn boundedness(&self) -> Boundedness {
+ self.properties().boundedness
+ }
+
+ fn pipeline_behavior(&self) -> EmissionType {
+ self.properties().emission_type
+ }
+
fn equivalence_properties(&self) -> &EquivalenceProperties {
self.properties().equivalence_properties()
}
}
-/// Describes the execution mode of the result of calling
-/// [`ExecutionPlan::execute`] with respect to its size and behavior.
+/// Represents whether a stream of data **generated** by an operator is bounded (finite)
+/// or unbounded (infinite).
///
-/// The mode of the execution plan is determined by the mode of its input
-/// execution plans and the details of the operator itself. For example, a
-/// `FilterExec` operator will have the same execution mode as its input, but a
-/// `SortExec` operator may have a different execution mode than its input,
-/// depending on how the input stream is sorted.
+/// This is used to determine whether an execution plan will eventually complete
+/// processing all its data (bounded) or could potentially run forever (unbounded).
///
-/// There are three possible execution modes: `Bounded`, `Unbounded` and
-/// `PipelineBreaking`.
-#[derive(Clone, Copy, PartialEq, Debug)]
-pub enum ExecutionMode {
- /// The stream is bounded / finite.
- ///
- /// In this case the stream will eventually return `None` to indicate that
- /// there are no more records to process.
+/// For unbounded streams, it also tracks whether the operator requires finite memory
+/// to process the stream or if memory usage could grow unbounded.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Boundedness {
+ /// The data stream is bounded (finite) and will eventually complete
Bounded,
- /// The stream is unbounded / infinite.
- ///
- /// In this case, the stream will never be done (never return `None`),
- /// except in case of error.
- ///
- /// This mode is often used in "Steaming" use cases where data is
- /// incrementally processed as it arrives.
- ///
- /// Note that even though the operator generates an unbounded stream of
- /// results, it can execute with bounded memory and incrementally produces
- /// output.
- Unbounded,
- /// Some of the operator's input stream(s) are unbounded, but the operator
- /// cannot generate streaming results from these streaming inputs.
- ///
- /// In this case, the execution mode will be pipeline breaking, e.g. the
- /// operator requires unbounded memory to generate results. This
- /// information is used by the planner when performing sanity checks
- /// on plans processings unbounded data sources.
- PipelineBreaking,
+ /// The data stream is unbounded (infinite) and could run forever
+ Unbounded {
+ /// Whether this operator requires infinite memory to process the unbounded stream.
+ /// If false, the operator can process an infinite stream with bounded memory.
Review Comment:
Makes sense, we will add some examples. In your case, `MEDIAN` would be an example (we can do `MIN`/`MAX` with bounded memory).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "jayzhan-synnada (via GitHub)" <gi...@apache.org>.
jayzhan-synnada commented on PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#issuecomment-2556214155
If we can merge this soon, I think we don't need to separate Join doc, just get in together. Thanks all for the review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#issuecomment-2556407929
Great, then let's wait for a few hours in case any other feedback arrives and then merge this. This way, we will avoid API churn on FFI between DataFusion 44 and 45.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1889834200
##########
datafusion/ffi/src/plan_properties.rs:
##########
@@ -220,50 +235,89 @@ impl TryFrom<FFI_PlanProperties> for PlanProperties {
RErr(e) => Err(DataFusionError::Plan(e.to_string())),
}?;
- let execution_mode: ExecutionMode =
- unsafe { (ffi_props.execution_mode)(&ffi_props).into() };
-
let eq_properties = match orderings {
Some(ordering) => {
EquivalenceProperties::new_with_orderings(Arc::new(schema), &[ordering])
}
None => EquivalenceProperties::new(Arc::new(schema)),
};
+ let emission_type: EmissionType =
+ unsafe { (ffi_props.emission_type)(&ffi_props).into() };
+
+ let boundedness: Boundedness =
+ unsafe { (ffi_props.boundedness)(&ffi_props).into() };
+
Ok(PlanProperties::new(
eq_properties,
partitioning,
- execution_mode,
+ emission_type,
+ boundedness,
))
}
}
-/// FFI safe version of [`ExecutionMode`].
+/// FFI safe version of [`Boundedness`].
#[repr(C)]
#[allow(non_camel_case_types)]
#[derive(Clone, StableAbi)]
-pub enum FFI_ExecutionMode {
+pub enum FFI_Boundedness {
Bounded,
- Unbounded,
Review Comment:
Was the purpose of FFI to provide a stable ABI? cc @timsaucer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1890174570
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -488,95 +495,152 @@ impl ExecutionPlanProperties for &dyn ExecutionPlan {
self.properties().output_partitioning()
}
- fn execution_mode(&self) -> ExecutionMode {
- self.properties().execution_mode()
- }
-
fn output_ordering(&self) -> Option<&LexOrdering> {
self.properties().output_ordering()
}
+ fn boundedness(&self) -> Boundedness {
+ self.properties().boundedness
+ }
+
+ fn pipeline_behavior(&self) -> EmissionType {
+ self.properties().emission_type
+ }
+
fn equivalence_properties(&self) -> &EquivalenceProperties {
self.properties().equivalence_properties()
}
}
-/// Describes the execution mode of the result of calling
-/// [`ExecutionPlan::execute`] with respect to its size and behavior.
+/// Represents whether a stream of data **generated** by an operator is bounded (finite)
+/// or unbounded (infinite).
///
-/// The mode of the execution plan is determined by the mode of its input
-/// execution plans and the details of the operator itself. For example, a
-/// `FilterExec` operator will have the same execution mode as its input, but a
-/// `SortExec` operator may have a different execution mode than its input,
-/// depending on how the input stream is sorted.
+/// This is used to determine whether an execution plan will eventually complete
+/// processing all its data (bounded) or could potentially run forever (unbounded).
///
-/// There are three possible execution modes: `Bounded`, `Unbounded` and
-/// `PipelineBreaking`.
-#[derive(Clone, Copy, PartialEq, Debug)]
-pub enum ExecutionMode {
- /// The stream is bounded / finite.
- ///
- /// In this case the stream will eventually return `None` to indicate that
- /// there are no more records to process.
+/// For unbounded streams, it also tracks whether the operator requires finite memory
+/// to process the stream or if memory usage could grow unbounded.
Review Comment:
Limit and TopK can be given as examples
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1890183926
##########
datafusion/ffi/src/plan_properties.rs:
##########
@@ -220,50 +235,89 @@ impl TryFrom<FFI_PlanProperties> for PlanProperties {
RErr(e) => Err(DataFusionError::Plan(e.to_string())),
}?;
- let execution_mode: ExecutionMode =
- unsafe { (ffi_props.execution_mode)(&ffi_props).into() };
-
let eq_properties = match orderings {
Some(ordering) => {
EquivalenceProperties::new_with_orderings(Arc::new(schema), &[ordering])
}
None => EquivalenceProperties::new(Arc::new(schema)),
};
+ let emission_type: EmissionType =
+ unsafe { (ffi_props.emission_type)(&ffi_props).into() };
+
+ let boundedness: Boundedness =
+ unsafe { (ffi_props.boundedness)(&ffi_props).into() };
+
Ok(PlanProperties::new(
eq_properties,
partitioning,
- execution_mode,
+ emission_type,
+ boundedness,
))
}
}
-/// FFI safe version of [`ExecutionMode`].
+/// FFI safe version of [`Boundedness`].
#[repr(C)]
#[allow(non_camel_case_types)]
#[derive(Clone, StableAbi)]
-pub enum FFI_ExecutionMode {
+pub enum FFI_Boundedness {
Bounded,
- Unbounded,
Review Comment:
Yes, the idea is to be stable. I don't think the real problem here is the change from `FFI_ExecutionMode` to `FFI_Boundness` but rather the way it breaks libraries working across the boundary that will expect one and receive the other. I think this current impact is minimal since the FFI interfaces have *just* been released and I don't think many people have had an opportunity to use them. However I will open an issue about adding a feature to validate the versioning across the boundary. I think that will be necessary going forward.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1890158218
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -488,95 +495,152 @@ impl ExecutionPlanProperties for &dyn ExecutionPlan {
self.properties().output_partitioning()
}
- fn execution_mode(&self) -> ExecutionMode {
- self.properties().execution_mode()
- }
-
fn output_ordering(&self) -> Option<&LexOrdering> {
self.properties().output_ordering()
}
+ fn boundedness(&self) -> Boundedness {
+ self.properties().boundedness
+ }
+
+ fn pipeline_behavior(&self) -> EmissionType {
+ self.properties().emission_type
+ }
+
fn equivalence_properties(&self) -> &EquivalenceProperties {
self.properties().equivalence_properties()
}
}
-/// Describes the execution mode of the result of calling
-/// [`ExecutionPlan::execute`] with respect to its size and behavior.
+/// Represents whether a stream of data **generated** by an operator is bounded (finite)
+/// or unbounded (infinite).
///
-/// The mode of the execution plan is determined by the mode of its input
-/// execution plans and the details of the operator itself. For example, a
-/// `FilterExec` operator will have the same execution mode as its input, but a
-/// `SortExec` operator may have a different execution mode than its input,
-/// depending on how the input stream is sorted.
+/// This is used to determine whether an execution plan will eventually complete
+/// processing all its data (bounded) or could potentially run forever (unbounded).
///
-/// There are three possible execution modes: `Bounded`, `Unbounded` and
-/// `PipelineBreaking`.
-#[derive(Clone, Copy, PartialEq, Debug)]
-pub enum ExecutionMode {
- /// The stream is bounded / finite.
- ///
- /// In this case the stream will eventually return `None` to indicate that
- /// there are no more records to process.
+/// For unbounded streams, it also tracks whether the operator requires finite memory
+/// to process the stream or if memory usage could grow unbounded.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Boundedness {
+ /// The data stream is bounded (finite) and will eventually complete
Bounded,
- /// The stream is unbounded / infinite.
- ///
- /// In this case, the stream will never be done (never return `None`),
- /// except in case of error.
- ///
- /// This mode is often used in "Steaming" use cases where data is
- /// incrementally processed as it arrives.
- ///
- /// Note that even though the operator generates an unbounded stream of
- /// results, it can execute with bounded memory and incrementally produces
- /// output.
- Unbounded,
- /// Some of the operator's input stream(s) are unbounded, but the operator
- /// cannot generate streaming results from these streaming inputs.
- ///
- /// In this case, the execution mode will be pipeline breaking, e.g. the
- /// operator requires unbounded memory to generate results. This
- /// information is used by the planner when performing sanity checks
- /// on plans processings unbounded data sources.
- PipelineBreaking,
+ /// The data stream is unbounded (infinite) and could run forever
+ Unbounded {
+ /// Whether this operator requires infinite memory to process the unbounded stream.
+ /// If false, the operator can process an infinite stream with bounded memory.
Review Comment:
Maybe we can give an example of each type?
Like is it the case that
```rust
Unbounded {
requires_infinite_memory: true,
}
```
Would describe a min/max aggregate?
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -488,95 +495,152 @@ impl ExecutionPlanProperties for &dyn ExecutionPlan {
self.properties().output_partitioning()
}
- fn execution_mode(&self) -> ExecutionMode {
- self.properties().execution_mode()
- }
-
fn output_ordering(&self) -> Option<&LexOrdering> {
self.properties().output_ordering()
}
+ fn boundedness(&self) -> Boundedness {
+ self.properties().boundedness
+ }
+
+ fn pipeline_behavior(&self) -> EmissionType {
+ self.properties().emission_type
+ }
+
fn equivalence_properties(&self) -> &EquivalenceProperties {
self.properties().equivalence_properties()
}
}
-/// Describes the execution mode of the result of calling
-/// [`ExecutionPlan::execute`] with respect to its size and behavior.
+/// Represents whether a stream of data **generated** by an operator is bounded (finite)
+/// or unbounded (infinite).
///
-/// The mode of the execution plan is determined by the mode of its input
-/// execution plans and the details of the operator itself. For example, a
-/// `FilterExec` operator will have the same execution mode as its input, but a
-/// `SortExec` operator may have a different execution mode than its input,
-/// depending on how the input stream is sorted.
+/// This is used to determine whether an execution plan will eventually complete
+/// processing all its data (bounded) or could potentially run forever (unbounded).
///
-/// There are three possible execution modes: `Bounded`, `Unbounded` and
-/// `PipelineBreaking`.
-#[derive(Clone, Copy, PartialEq, Debug)]
-pub enum ExecutionMode {
- /// The stream is bounded / finite.
- ///
- /// In this case the stream will eventually return `None` to indicate that
- /// there are no more records to process.
+/// For unbounded streams, it also tracks whether the operator requires finite memory
+/// to process the stream or if memory usage could grow unbounded.
Review Comment:
Can we clarify in these comments how Boundedness is related to the input stream?
For example, if the the input stream is unbounded, does that imply the output is also unbounded?
Maybe some examples would help to clarify this
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -488,95 +495,152 @@ impl ExecutionPlanProperties for &dyn ExecutionPlan {
self.properties().output_partitioning()
}
- fn execution_mode(&self) -> ExecutionMode {
- self.properties().execution_mode()
- }
-
fn output_ordering(&self) -> Option<&LexOrdering> {
self.properties().output_ordering()
}
+ fn boundedness(&self) -> Boundedness {
+ self.properties().boundedness
+ }
+
+ fn pipeline_behavior(&self) -> EmissionType {
+ self.properties().emission_type
+ }
+
fn equivalence_properties(&self) -> &EquivalenceProperties {
self.properties().equivalence_properties()
}
}
-/// Describes the execution mode of the result of calling
-/// [`ExecutionPlan::execute`] with respect to its size and behavior.
+/// Represents whether a stream of data **generated** by an operator is bounded (finite)
+/// or unbounded (infinite).
///
-/// The mode of the execution plan is determined by the mode of its input
-/// execution plans and the details of the operator itself. For example, a
-/// `FilterExec` operator will have the same execution mode as its input, but a
-/// `SortExec` operator may have a different execution mode than its input,
-/// depending on how the input stream is sorted.
+/// This is used to determine whether an execution plan will eventually complete
+/// processing all its data (bounded) or could potentially run forever (unbounded).
///
-/// There are three possible execution modes: `Bounded`, `Unbounded` and
-/// `PipelineBreaking`.
-#[derive(Clone, Copy, PartialEq, Debug)]
-pub enum ExecutionMode {
- /// The stream is bounded / finite.
- ///
- /// In this case the stream will eventually return `None` to indicate that
- /// there are no more records to process.
+/// For unbounded streams, it also tracks whether the operator requires finite memory
+/// to process the stream or if memory usage could grow unbounded.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Boundedness {
+ /// The data stream is bounded (finite) and will eventually complete
Bounded,
- /// The stream is unbounded / infinite.
- ///
- /// In this case, the stream will never be done (never return `None`),
- /// except in case of error.
- ///
- /// This mode is often used in "Steaming" use cases where data is
- /// incrementally processed as it arrives.
- ///
- /// Note that even though the operator generates an unbounded stream of
- /// results, it can execute with bounded memory and incrementally produces
- /// output.
- Unbounded,
- /// Some of the operator's input stream(s) are unbounded, but the operator
- /// cannot generate streaming results from these streaming inputs.
- ///
- /// In this case, the execution mode will be pipeline breaking, e.g. the
- /// operator requires unbounded memory to generate results. This
- /// information is used by the planner when performing sanity checks
- /// on plans processings unbounded data sources.
- PipelineBreaking,
+ /// The data stream is unbounded (infinite) and could run forever
+ Unbounded {
+ /// Whether this operator requires infinite memory to process the unbounded stream.
+ /// If false, the operator can process an infinite stream with bounded memory.
+ /// If true, memory usage may grow unbounded while processing the stream.
+ requires_infinite_memory: bool,
+ },
}
-impl ExecutionMode {
- /// Check whether the execution mode is unbounded or not.
+impl Boundedness {
pub fn is_unbounded(&self) -> bool {
- matches!(self, ExecutionMode::Unbounded)
+ matches!(self, Boundedness::Unbounded { .. })
}
+}
- /// Check whether the execution is pipeline friendly. If so, operator can
- /// execute safely.
- pub fn pipeline_friendly(&self) -> bool {
- matches!(self, ExecutionMode::Bounded | ExecutionMode::Unbounded)
- }
+/// Represents how an operator emits its output records.
+///
+/// This is used to determine whether an operator emits records incrementally as they arrive,
Review Comment:
I suggest documenting when someone would expect incremental output -- I think it is basically when the operator can emit a RecordBatches of `batch_size` rows
Specifically I think it would help to say something like "EmissionType::Incremental generates output incrementally but the operator may still buffer data internally until `batch_size` records arrive
##########
datafusion/ffi/src/plan_properties.rs:
##########
@@ -220,50 +235,89 @@ impl TryFrom<FFI_PlanProperties> for PlanProperties {
RErr(e) => Err(DataFusionError::Plan(e.to_string())),
}?;
- let execution_mode: ExecutionMode =
- unsafe { (ffi_props.execution_mode)(&ffi_props).into() };
-
let eq_properties = match orderings {
Some(ordering) => {
EquivalenceProperties::new_with_orderings(Arc::new(schema), &[ordering])
}
None => EquivalenceProperties::new(Arc::new(schema)),
};
+ let emission_type: EmissionType =
+ unsafe { (ffi_props.emission_type)(&ffi_props).into() };
+
+ let boundedness: Boundedness =
+ unsafe { (ffi_props.boundedness)(&ffi_props).into() };
+
Ok(PlanProperties::new(
eq_properties,
partitioning,
- execution_mode,
+ emission_type,
+ boundedness,
))
}
}
-/// FFI safe version of [`ExecutionMode`].
+/// FFI safe version of [`Boundedness`].
#[repr(C)]
#[allow(non_camel_case_types)]
#[derive(Clone, StableAbi)]
-pub enum FFI_ExecutionMode {
+pub enum FFI_Boundedness {
Bounded,
- Unbounded,
Review Comment:
I think so too -- Maybe we can leave the `FFI_ExecutionMode` present and mark it deprecated or something 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1890178866
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -488,95 +495,152 @@ impl ExecutionPlanProperties for &dyn ExecutionPlan {
self.properties().output_partitioning()
}
- fn execution_mode(&self) -> ExecutionMode {
- self.properties().execution_mode()
- }
-
fn output_ordering(&self) -> Option<&LexOrdering> {
self.properties().output_ordering()
}
+ fn boundedness(&self) -> Boundedness {
+ self.properties().boundedness
+ }
+
+ fn pipeline_behavior(&self) -> EmissionType {
+ self.properties().emission_type
+ }
+
fn equivalence_properties(&self) -> &EquivalenceProperties {
self.properties().equivalence_properties()
}
}
-/// Describes the execution mode of the result of calling
-/// [`ExecutionPlan::execute`] with respect to its size and behavior.
+/// Represents whether a stream of data **generated** by an operator is bounded (finite)
+/// or unbounded (infinite).
///
-/// The mode of the execution plan is determined by the mode of its input
-/// execution plans and the details of the operator itself. For example, a
-/// `FilterExec` operator will have the same execution mode as its input, but a
-/// `SortExec` operator may have a different execution mode than its input,
-/// depending on how the input stream is sorted.
+/// This is used to determine whether an execution plan will eventually complete
+/// processing all its data (bounded) or could potentially run forever (unbounded).
///
-/// There are three possible execution modes: `Bounded`, `Unbounded` and
-/// `PipelineBreaking`.
-#[derive(Clone, Copy, PartialEq, Debug)]
-pub enum ExecutionMode {
- /// The stream is bounded / finite.
- ///
- /// In this case the stream will eventually return `None` to indicate that
- /// there are no more records to process.
+/// For unbounded streams, it also tracks whether the operator requires finite memory
+/// to process the stream or if memory usage could grow unbounded.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Boundedness {
+ /// The data stream is bounded (finite) and will eventually complete
Bounded,
- /// The stream is unbounded / infinite.
- ///
- /// In this case, the stream will never be done (never return `None`),
- /// except in case of error.
- ///
- /// This mode is often used in "Steaming" use cases where data is
- /// incrementally processed as it arrives.
- ///
- /// Note that even though the operator generates an unbounded stream of
- /// results, it can execute with bounded memory and incrementally produces
- /// output.
- Unbounded,
- /// Some of the operator's input stream(s) are unbounded, but the operator
- /// cannot generate streaming results from these streaming inputs.
- ///
- /// In this case, the execution mode will be pipeline breaking, e.g. the
- /// operator requires unbounded memory to generate results. This
- /// information is used by the planner when performing sanity checks
- /// on plans processings unbounded data sources.
- PipelineBreaking,
+ /// The data stream is unbounded (infinite) and could run forever
+ Unbounded {
+ /// Whether this operator requires infinite memory to process the unbounded stream.
+ /// If false, the operator can process an infinite stream with bounded memory.
Review Comment:
min/max over unordered keys are `requires_infinite_memory: true` by default, but if the key cardinality is finite, then it is false. That could be added as an example.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#issuecomment-2557093668
🚀
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1889838140
##########
datafusion/common/src/join_type.rs:
##########
@@ -28,21 +28,26 @@ use crate::{DataFusionError, Result};
/// Join type
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Hash)]
pub enum JoinType {
- /// Inner Join
+ /// Inner Join - Returns rows where there is a match in both tables.
Review Comment:
if we want to get precise, this should also say that all matches are returned (n-to-m)
##########
datafusion/common/src/join_type.rs:
##########
@@ -28,21 +28,26 @@ use crate::{DataFusionError, Result};
/// Join type
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Hash)]
pub enum JoinType {
- /// Inner Join
+ /// Inner Join - Returns rows where there is a match in both tables.
Inner,
- /// Left Join
+ /// Left Join - Returns all rows from the left table and matching rows from the right table.
+ /// If no match, NULL values are returned for columns from the right table.
Left,
- /// Right Join
+ /// Right Join - Returns all rows from the right table and matching rows from the left table.
+ /// If no match, NULL values are returned for columns from the left table.
Right,
- /// Full Join
+ /// Full Join - Returns all rows when there is a match in either table.
+ /// Rows without a match in one table will have NULL values for columns from that table.
Review Comment:
> there is a match in either table
"a match" implies a matching row in left and right tables.
also see above, this should also imply cardinality
maybe
```
/// Full Join - returns all matches between left and right table,
/// plus left table rows without a match (with NULL values for right table columns)
/// plus right table rows without a match (with NULL values for left table columns)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1890188171
##########
datafusion/ffi/src/plan_properties.rs:
##########
@@ -220,50 +235,89 @@ impl TryFrom<FFI_PlanProperties> for PlanProperties {
RErr(e) => Err(DataFusionError::Plan(e.to_string())),
}?;
- let execution_mode: ExecutionMode =
- unsafe { (ffi_props.execution_mode)(&ffi_props).into() };
-
let eq_properties = match orderings {
Some(ordering) => {
EquivalenceProperties::new_with_orderings(Arc::new(schema), &[ordering])
}
None => EquivalenceProperties::new(Arc::new(schema)),
};
+ let emission_type: EmissionType =
+ unsafe { (ffi_props.emission_type)(&ffi_props).into() };
+
+ let boundedness: Boundedness =
+ unsafe { (ffi_props.boundedness)(&ffi_props).into() };
+
Ok(PlanProperties::new(
eq_properties,
partitioning,
- execution_mode,
+ emission_type,
+ boundedness,
))
}
}
-/// FFI safe version of [`ExecutionMode`].
+/// FFI safe version of [`Boundedness`].
#[repr(C)]
#[allow(non_camel_case_types)]
#[derive(Clone, StableAbi)]
-pub enum FFI_ExecutionMode {
+pub enum FFI_Boundedness {
Bounded,
- Unbounded,
Review Comment:
https://github.com/apache/datafusion/issues/13827
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Replace `execution_mode` with `emission_type` and `boundedness` [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #13823:
URL: https://github.com/apache/datafusion/pull/13823#discussion_r1890195326
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -488,95 +495,152 @@ impl ExecutionPlanProperties for &dyn ExecutionPlan {
self.properties().output_partitioning()
}
- fn execution_mode(&self) -> ExecutionMode {
- self.properties().execution_mode()
- }
-
fn output_ordering(&self) -> Option<&LexOrdering> {
self.properties().output_ordering()
}
+ fn boundedness(&self) -> Boundedness {
+ self.properties().boundedness
+ }
+
+ fn pipeline_behavior(&self) -> EmissionType {
+ self.properties().emission_type
+ }
+
fn equivalence_properties(&self) -> &EquivalenceProperties {
self.properties().equivalence_properties()
}
}
-/// Describes the execution mode of the result of calling
-/// [`ExecutionPlan::execute`] with respect to its size and behavior.
+/// Represents whether a stream of data **generated** by an operator is bounded (finite)
+/// or unbounded (infinite).
///
-/// The mode of the execution plan is determined by the mode of its input
-/// execution plans and the details of the operator itself. For example, a
-/// `FilterExec` operator will have the same execution mode as its input, but a
-/// `SortExec` operator may have a different execution mode than its input,
-/// depending on how the input stream is sorted.
+/// This is used to determine whether an execution plan will eventually complete
+/// processing all its data (bounded) or could potentially run forever (unbounded).
///
-/// There are three possible execution modes: `Bounded`, `Unbounded` and
-/// `PipelineBreaking`.
-#[derive(Clone, Copy, PartialEq, Debug)]
-pub enum ExecutionMode {
- /// The stream is bounded / finite.
- ///
- /// In this case the stream will eventually return `None` to indicate that
- /// there are no more records to process.
+/// For unbounded streams, it also tracks whether the operator requires finite memory
+/// to process the stream or if memory usage could grow unbounded.
Review Comment:
Sure, we can expand the comments. Basically, boundedness of the output stream is a function of the boundedness of the input stream *and* the nature of the operator. For example, a simple SISO operator that transforms it data preserves the boundedness of its input, while an operator with fetch can have an unbounded input stream but a bounded output stream.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org