You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "QinXi-ai (via GitHub)" <gi...@apache.org> on 2026/09/26 05:52:04 UTC
[PR] feat: serialize LazyMemoryExec through generator hooks [datafusion]
QinXi-ai opened a new pull request, #25765:
URL: https://github.com/apache/datafusion/pull/25765
## Which issue does this PR close?
Part of #25395. This is the wire-compatible generator-hook step; it does not close the multi-generator serialization gap.
## Rationale for this change
Serializing a built-in `LazyMemoryExec` currently requires `datafusion-proto` to recognize each concrete generator type. A generator cannot supply its own encoding even when it can produce the existing generate-series representation.
This change lets the generator provide that representation and keeps decoding alongside the concrete generator types. Existing single-generator plans retain their protobuf bytes and execution behavior.
## What changes are included in this PR?
- Add an optional `LazyBatchGenerator::try_to_proto` hook and implement `LazyMemoryExec::try_to_proto` using it.
- Implement the hook for empty, integer and timestamp/date series, behind a new `datafusion-functions-table/proto` feature. Generic series values can supply their argument encoding; the default remains extension-codec fallback.
- Move built-in decoding into `datafusion-functions-table` and remove the generator downcasts from the central serializer.
- Preserve extension fallback for unsupported, empty and multiple-generator plans, and propagate real encoding errors.
- Fix two pre-existing Windows Clippy blockers with the repository's Rust 1.98.1 toolchain: remove a redundant semicolon in conservative floating-point rounding, and document the intentional async API of the Windows-only stub for the FIFO example.
- Mark the literal `file:///` example in the 14.0.0 changelog as code so the link checker does not treat it as a local filesystem link on Windows.
The fixed built-in decoder still requires the `datafusion-functions-table` dependency. Removing it would require a separate injection/registration design. The existing wire message has a single argument payload, so multiple generators also need a separate protocol change.
## What is the testing strategy for this PR?
The new integration tests construct legacy wire messages independently of the hooks and check exact re-encoding, output values, batch boundaries and repeatable execution. They cover `generate_series` and `range`, empty/integer/timestamp/date payloads, a generator absent from the old downcast chain, extension fallback, malformed input and batch-size overflow.
- `cargo check --locked -p datafusion-functions-table --no-default-features`: passed.
- `cargo test --locked -p datafusion-proto -p datafusion-functions-table --all-features`: 301 tests passed, including 276 integration tests and 4 doctests; no failures.
- `cargo fmt --all` and `git diff --check`: passed.
- `cargo clippy --locked --all-targets --all-features -- -D warnings`: passed.
- All checks from `dev/rust_lint.sh` passed individually, including its Clippy feature combination, formatting, licenses, spelling, generated docs, links, security audit, dependency checks, Rust docs and Sphinx docs. On native Windows, the Sphinx build used `uv run --package datafusion-docs bash ./docs/build.sh` to launch the shell script explicitly.
- The physical-plan crate passed all 2,390 tests in the extended-workspace configuration.
The extended Windows workspace run is not fully green. Four datasource URL/cache tests and four floating-point statistics/interval tests failed; all eight reproduce on an unmodified source export of base `6a792c6`, with the same assertions. A repartition spill timeout test failed in the concurrent run but passed its exact isolated rerun (12 internal attempts, 104.23 seconds). Remaining long-running integration suites are still in progress. This PR stays draft pending completion of platform/CI validation.
## Are there any user-facing changes?
New defaulted Rust trait methods and an opt-in `proto` feature. Existing downstream generators need no changes. Generate-series helper methods are removed from `PhysicalPlanNodeExt`; direct callers of its decoder can use `datafusion_functions_table::generate_series::proto::try_from_proto`. The built-in single-generator wire format is unchanged; multi-generator serialization remains outside this first step.
--
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] feat: serialize LazyMemoryExec through generator hooks [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #25765:
URL: https://github.com/apache/datafusion/pull/25765#discussion_r4282992988
##########
datafusion/physical-plan/src/memory.rs:
##########
@@ -159,6 +159,19 @@ pub trait LazyBatchGenerator: Send + Sync + fmt::Debug + fmt::Display {
/// Returns a new instance with the state reset.
fn reset_state(&self) -> Arc<RwLock<dyn LazyBatchGenerator>>;
+
+ /// Encode this generator using the built-in generate-series wire format.
Review Comment:
Could we clarify the contract for this hook? Returning `Some(node)` means the plan will be reconstructed as a built-in series generator, so its rows, batch boundaries, and schema should match the original generator after reset. It would also help to mention that `Err` stops serialization, while only `Ok(None)` allows extension fallback.
##########
datafusion/proto/tests/cases/plans/lazy_memory.rs:
##########
@@ -0,0 +1,345 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::any::Any;
+use std::fmt;
+use std::sync::Arc;
+
+use arrow::array::{Int64Array, RecordBatch, TimestampNanosecondArray};
+use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
+use datafusion_common::{Result, assert_contains, not_impl_err};
+use datafusion_execution::TaskContext;
+use datafusion_physical_plan::ExecutionPlan;
+use datafusion_physical_plan::common::collect;
+use datafusion_physical_plan::memory::{LazyBatchGenerator, LazyMemoryExec};
+use datafusion_proto::physical_plan::{
+ AsExecutionPlan, DefaultPhysicalExtensionCodec, PhysicalExtensionCodec,
+ PhysicalProtoConverterExtension,
+};
+use datafusion_proto::protobuf::{self, PhysicalPlanNode};
+use parking_lot::RwLock;
+use prost::Message;
+
+fn plan_node(node: protobuf::GenerateSeriesNode) -> PhysicalPlanNode {
+ PhysicalPlanNode {
+ physical_plan_type: Some(
+ protobuf::physical_plan_node::PhysicalPlanType::GenerateSeries(node),
+ ),
+ }
+}
+
+fn int_node() -> Result<protobuf::GenerateSeriesNode> {
+ let schema = Schema::new(vec![Field::new("v", DataType::Int64, false)]);
+ Ok(protobuf::GenerateSeriesNode {
+ schema: Some((&schema).try_into()?),
+ target_batch_size: 2,
+ args: Some(protobuf::generate_series_node::Args::Int64Args(
+ protobuf::GenerateSeriesArgsInt64 {
+ start: 3,
+ end: 1,
+ step: -1,
+ include_end: true,
+ name: protobuf::GenerateSeriesName::GsGenerateSeries as i32,
+ },
+ )),
+ })
+}
+
+/// Construct the legacy wire messages independently of the generator hooks,
+/// then check exact re-encoding and actual execution, including batch sizes.
+#[tokio::test]
+async fn lazy_memory_legacy_wire_roundtrip() -> Result<()> {
+ use protobuf::generate_series_node::Args;
+ for (name, include_end) in [
+ (protobuf::GenerateSeriesName::GsGenerateSeries, true),
+ (protobuf::GenerateSeriesName::GsRange, false),
+ ] {
+ let mut integer = int_node()?;
+ let Some(Args::Int64Args(args)) = integer.args.as_mut() else {
+ unreachable!()
+ };
+ args.name = name as i32;
+ args.include_end = include_end;
+ let mut empty = integer.clone();
+ empty.target_batch_size = 8192;
+ empty.args = Some(Args::ContainsNull(
+ protobuf::GenerateSeriesArgsContainsNull { name: name as i32 },
+ ));
+ let expected = if include_end {
+ vec![3, 2, 1]
+ } else {
+ vec![3, 2]
+ };
+ let mut cases = vec![(integer, expected.clone()), (empty, vec![])];
+ for tz in [None, Some("+08:00")] {
+ let schema = Schema::new(vec![Field::new(
+ "v",
+ DataType::Timestamp(TimeUnit::Nanosecond, tz.map(Into::into)),
+ false,
+ )]);
+ let step = Some(datafusion_proto_common::IntervalMonthDayNanoValue {
+ months: 0,
+ days: 0,
+ nanos: -1,
+ });
+ let args = match tz {
+ Some(tz) => Args::TimestampArgs(protobuf::GenerateSeriesArgsTimestamp {
+ start: 3,
+ end: 1,
+ step,
+ include_end,
+ name: name as i32,
+ tz: Some(tz.to_string()),
+ }),
+ None => Args::DateArgs(protobuf::GenerateSeriesArgsDate {
+ start: 3,
+ end: 1,
+ step,
+ include_end,
+ name: name as i32,
+ }),
+ };
+ cases.push((
+ protobuf::GenerateSeriesNode {
+ schema: Some((&schema).try_into()?),
+ target_batch_size: 2,
+ args: Some(args),
+ },
+ expected.clone(),
+ ));
+ }
+ for (node, expected) in cases {
+ let legacy = plan_node(node);
+ let bytes = legacy.encode_to_vec();
+ let decoded = PhysicalPlanNode::decode(bytes.as_slice()).unwrap();
+ let ctx = Arc::new(TaskContext::default());
+ let plan = decoded
+ .try_into_physical_plan(&ctx, &DefaultPhysicalExtensionCodec {})?;
+ let encoded = PhysicalPlanNode::try_from_physical_plan(
+ Arc::clone(&plan),
+ &DefaultPhysicalExtensionCodec {},
+ )?;
+ assert_eq!(encoded.encode_to_vec(), bytes);
+ // Repeated execute calls must still produce fresh independent streams.
+ for _ in 0..2 {
+ let batches = collect(plan.execute(0, Arc::clone(&ctx))?).await?;
+ let sizes: Vec<_> = batches.iter().map(RecordBatch::num_rows).collect();
+ let expected_sizes: Vec<_> =
+ expected.chunks(2).map(<[i64]>::len).collect();
+ assert_eq!(sizes, expected_sizes);
+ let actual: Vec<i64> = batches
+ .iter()
+ .flat_map(|batch| {
+ let column = batch.column(0);
+ if let Some(array) = column.as_any().downcast_ref::<Int64Array>()
+ {
+ array.values().to_vec()
+ } else {
+ column
+ .as_any()
+ .downcast_ref::<TimestampNanosecondArray>()
+ .unwrap()
+ .values()
+ .to_vec()
+ }
+ })
+ .collect();
+ assert_eq!(actual, expected);
+ }
+ }
+ }
+ Ok(())
+}
+
+#[derive(Debug, Clone)]
+struct HookGenerator(protobuf::GenerateSeriesNode);
+
+impl fmt::Display for HookGenerator {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ write!(f, "HookGenerator")
+ }
+}
+
+impl LazyBatchGenerator for HookGenerator {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+ fn generate_next_batch(&mut self) -> Result<Option<RecordBatch>> {
+ Ok(None)
+ }
+ fn reset_state(&self) -> Arc<RwLock<dyn LazyBatchGenerator>> {
+ Arc::new(RwLock::new(self.clone()))
+ }
+ fn try_to_proto(&self) -> Result<Option<protobuf::GenerateSeriesNode>> {
+ Ok(Some(self.0.clone()))
+ }
+}
+
+#[test]
+fn lazy_memory_dispatches_to_generator_hook() -> Result<()> {
Review Comment:
Could we optionally add a custom generator that delegates execution and reset to a real series, then compare its rows, batch sizes, and schema before and after serialization? The current `HookGenerator` returns no rows even though its payload represents `[3, 2, 1]`, so this would help demonstrate the execution contract for custom generators.
--
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] feat: serialize LazyMemoryExec through generator hooks [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on PR #25765:
URL: https://github.com/apache/datafusion/pull/25765#issuecomment-6261796769
🚀
@QinXi-ai
Thank you for your contribution.
--
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] feat: serialize LazyMemoryExec through generator hooks [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew merged PR #25765:
URL: https://github.com/apache/datafusion/pull/25765
--
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] feat: serialize LazyMemoryExec through generator hooks [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #25765:
URL: https://github.com/apache/datafusion/pull/25765#issuecomment-5948299137
<!-- semver-check-comment -->
Thank you for opening this pull request!
Reviewer note: [cargo-semver-checks](https://github.com/obi1kenobi/cargo-semver-checks) reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).
<details>
<summary>Details</summary>
```
Cloning apache/main
Building datafusion-functions-table v55.1.0 (current)
Built [ 47.130s] (current)
Parsing datafusion-functions-table v55.1.0 (current)
Parsed [ 0.012s] (current)
Building datafusion-functions-table v55.1.0 (baseline)
Built [ 42.232s] (baseline)
Parsing datafusion-functions-table v55.1.0 (baseline)
Parsed [ 0.012s] (baseline)
Checking datafusion-functions-table v55.1.0 -> v55.1.0 (no change; assume patch)
Checked [ 0.091s] 223 checks: 223 pass, 31 skip
Summary no semver update required
Finished [ 90.688s] datafusion-functions-table
Building datafusion-physical-plan v55.1.0 (current)
Built [ 40.565s] (current)
Parsing datafusion-physical-plan v55.1.0 (current)
Parsed [ 0.174s] (current)
Building datafusion-physical-plan v55.1.0 (baseline)
Built [ 40.377s] (baseline)
Parsing datafusion-physical-plan v55.1.0 (baseline)
Parsed [ 0.173s] (baseline)
Checking datafusion-physical-plan v55.1.0 -> v55.1.0 (no change; assume patch)
Checked [ 0.636s] 223 checks: 223 pass, 31 skip
Summary no semver update required
Finished [ 83.279s] datafusion-physical-plan
Building datafusion-proto v55.1.0 (current)
Built [ 56.731s] (current)
Parsing datafusion-proto v55.1.0 (current)
Parsed [ 0.018s] (current)
Building datafusion-proto v55.1.0 (baseline)
Built [ 57.749s] (baseline)
Parsing datafusion-proto v55.1.0 (baseline)
Parsed [ 0.019s] (baseline)
Checking datafusion-proto v55.1.0 -> v55.1.0 (no change; assume patch)
Checked [ 0.128s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip
--- failure trait_method_missing: pub trait method removed or renamed ---
Description:
A trait method is no longer callable, and may have been renamed or removed entirely.
ref: https://doc.rust-lang.org/cargo/reference/semver.html#trait-item-signature
impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/trait_method_missing.ron
Failed in:
method generate_series_name_to_str of trait PhysicalPlanNodeExt, previously in file /home/runner/work/datafusion/datafusion/target/semver-checks/git-apache_main/a1ffa10ebf54831f42c65f00a084297c63e72f39/datafusion/proto/src/physical_plan/mod.rs:1416
method try_into_generate_series_physical_plan of trait PhysicalPlanNodeExt, previously in file /home/runner/work/datafusion/datafusion/target/semver-checks/git-apache_main/a1ffa10ebf54831f42c65f00a084297c63e72f39/datafusion/proto/src/physical_plan/mod.rs:1423
method str_to_generate_series_name of trait PhysicalPlanNodeExt, previously in file /home/runner/work/datafusion/datafusion/target/semver-checks/git-apache_main/a1ffa10ebf54831f42c65f00a084297c63e72f39/datafusion/proto/src/physical_plan/mod.rs:1506
method try_from_lazy_memory_exec of trait PhysicalPlanNodeExt, previously in file /home/runner/work/datafusion/datafusion/target/semver-checks/git-apache_main/a1ffa10ebf54831f42c65f00a084297c63e72f39/datafusion/proto/src/physical_plan/mod.rs:1514
Summary semver requires new major version: 1 major and 0 minor checks failed
Finished [ 116.183s] datafusion-proto
```
</details>
--
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] feat: serialize LazyMemoryExec through generator hooks [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #25765:
URL: https://github.com/apache/datafusion/pull/25765#issuecomment-5948631033
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/25765?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `88.00000%` with `21 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 82.61%. Comparing base ([`4d167a1`](https://app.codecov.io/gh/apache/datafusion/commit/4d167a167228a424ef8cc29fd6bea5f17fc23c2a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`507eb0f`](https://app.codecov.io/gh/apache/datafusion/commit/507eb0f454d7d714207ec32c2db7b662e70f9d91?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/25765?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [datafusion/functions-table/src/generate\_series.rs](https://app.codecov.io/gh/apache/datafusion/pull/25765?src=pr&el=tree&filepath=datafusion%2Ffunctions-table%2Fsrc%2Fgenerate_series.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9mdW5jdGlvbnMtdGFibGUvc3JjL2dlbmVyYXRlX3Nlcmllcy5ycw==) | 82.71% | [11 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/25765?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...usion/functions-table/src/generate\_series/proto.rs](https://app.codecov.io/gh/apache/datafusion/pull/25765?src=pr&el=tree&filepath=datafusion%2Ffunctions-table%2Fsrc%2Fgenerate_series%2Fproto.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9mdW5jdGlvbnMtdGFibGUvc3JjL2dlbmVyYXRlX3Nlcmllcy9wcm90by5ycw==) | 91.78% | [2 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/25765?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [datafusion/physical-plan/src/memory.rs](https://app.codecov.io/gh/apache/datafusion/pull/25765?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Fmemory.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9tZW1vcnkucnM=) | 95.00% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/25765?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #25765 +/- ##
==========================================
+ Coverage 82.60% 82.61% +0.01%
==========================================
Files 1145 1146 +1
Lines 444057 444064 +7
Branches 444057 444064 +7
==========================================
+ Hits 366796 366875 +79
+ Misses 54996 54922 -74
- Partials 22265 22267 +2
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/25765?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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