This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark-connect-rust.git
The following commit(s) were added to refs/heads/master by this push:
new 0748b05 [SPARK-60549][CONNECT][FOLLOWUP] Assert shared plan identity
directly
0748b05 is described below
commit 0748b0570aa8091a4ccbdd49a5ec18dc0be3a095
Author: DB Tsai <[email protected]>
AuthorDate: Tue Oct 27 14:29:37 2026 -0700
[SPARK-60549][CONNECT][FOLLOWUP] Assert shared plan identity directly
### What changes were proposed in this pull request?
Follow up on [the review of
#102](https://github.com/apache/spark-connect-rust/pull/102#pullrequestreview-5672008601)
for [SPARK-60549](https://issues.apache.org/jira/browse/SPARK-60549).
Replace exact `Arc::strong_count` assertions in
`related_builders_share_the_source_plan` with direct `LogicalPlan` pointer
comparisons. Cover both batch writers, the streaming writer, the merge writer,
`TableArg`, and its clone. Four crate-private accessors, compiled only for
tests, expose the builders' plan references.
### Why are the changes needed?
An extra `Arc` handle can break an exact-count assertion even when every
builder still shares the original plan. Pointer comparisons directly check that
each builder retains the source allocation and reject separately allocated
copies.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- `cargo fmt --all --check` passed.
- 392 release-mode Rust tests passed: 368 library tests and 24
expression-sharing/golden integration tests.
- All core integration targets compiled in release mode, both with and
without `wasm-udf`.
- Five CI-tooling tests passed.
- Workspace Clippy was blocked by the locally unavailable
`wasm32-unknown-unknown` target. A core-only Clippy run reached existing
`clippy::approx_constant` errors in unchanged tests; neither Clippy run passed.
- The live-server and full official parity suites were not rerun for this
test-only change.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6)
This pull request and its description were written by Isaac.
Closes #104 from dbtsai/codex/shared-plan-pointer-identity.
Authored-by: DB Tsai <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
---
crates/spark-connect/src/dataframe.rs | 23 ++++++-----------------
crates/spark-connect/src/merge.rs | 5 +++++
crates/spark-connect/src/readwriter.rs | 10 ++++++++++
crates/spark-connect/src/streaming.rs | 5 +++++
4 files changed, 26 insertions(+), 17 deletions(-)
diff --git a/crates/spark-connect/src/dataframe.rs
b/crates/spark-connect/src/dataframe.rs
index 0cf1229..0a273cc 100644
--- a/crates/spark-connect/src/dataframe.rs
+++ b/crates/spark-connect/src/dataframe.rs
@@ -2686,30 +2686,19 @@ mod plan_construction_tests {
fn related_builders_share_the_source_plan() {
let spark = session();
let source = spark.range(10).unwrap().limit(5);
- assert_eq!(Arc::strong_count(&source.plan), 1);
let writer = source.write();
- assert_eq!(Arc::strong_count(&source.plan), 2);
+ assert!(std::ptr::eq(source.plan(), writer.plan()));
let writer_v2 = source.write_to("target");
- assert_eq!(Arc::strong_count(&source.plan), 3);
+ assert!(std::ptr::eq(source.plan(), writer_v2.plan()));
let stream_writer = source.write_stream();
- assert_eq!(Arc::strong_count(&source.plan), 4);
+ assert!(std::ptr::eq(source.plan(), stream_writer.plan()));
let merge_writer = source.merge_into("target",
crate::column::col("id"));
- assert_eq!(Arc::strong_count(&source.plan), 5);
+ assert!(std::ptr::eq(source.plan(), merge_writer.plan()));
let table_arg = crate::table_arg::TableArg::new(source.clone());
- assert_eq!(Arc::strong_count(&source.plan), 6);
+ assert!(std::ptr::eq(source.plan(), table_arg.plan()));
let cloned_table_arg = table_arg.clone();
- assert_eq!(Arc::strong_count(&source.plan), 7);
-
- drop((
- writer,
- writer_v2,
- stream_writer,
- merge_writer,
- table_arg,
- cloned_table_arg,
- ));
- assert_eq!(Arc::strong_count(&source.plan), 1);
+ assert!(std::ptr::eq(source.plan(), cloned_table_arg.plan()));
}
#[test]
diff --git a/crates/spark-connect/src/merge.rs
b/crates/spark-connect/src/merge.rs
index fa22172..1c723db 100644
--- a/crates/spark-connect/src/merge.rs
+++ b/crates/spark-connect/src/merge.rs
@@ -50,6 +50,11 @@ pub struct MergeIntoWriter {
}
impl MergeIntoWriter {
+ #[cfg(test)]
+ pub(crate) fn plan(&self) -> &LogicalPlan {
+ self.source_plan.as_ref()
+ }
+
pub(crate) fn new(
session: SparkSession,
source_plan: Arc<LogicalPlan>,
diff --git a/crates/spark-connect/src/readwriter.rs
b/crates/spark-connect/src/readwriter.rs
index 7b01d0a..54fafeb 100644
--- a/crates/spark-connect/src/readwriter.rs
+++ b/crates/spark-connect/src/readwriter.rs
@@ -344,6 +344,11 @@ pub struct DataFrameWriter {
}
impl DataFrameWriter {
+ #[cfg(test)]
+ pub(crate) fn plan(&self) -> &LogicalPlan {
+ self.input_plan.as_ref()
+ }
+
/// Create a new DataFrameWriter.
pub(crate) fn new(session: SparkSession, input_plan: Arc<LogicalPlan>) ->
Self {
DataFrameWriter {
@@ -522,6 +527,11 @@ pub struct DataFrameWriterV2 {
}
impl DataFrameWriterV2 {
+ #[cfg(test)]
+ pub(crate) fn plan(&self) -> &LogicalPlan {
+ self.input_plan.as_ref()
+ }
+
/// Create a new DataFrameWriterV2 targeting `table_name`.
pub(crate) fn new(
session: SparkSession,
diff --git a/crates/spark-connect/src/streaming.rs
b/crates/spark-connect/src/streaming.rs
index 86ee11a..841d885 100644
--- a/crates/spark-connect/src/streaming.rs
+++ b/crates/spark-connect/src/streaming.rs
@@ -259,6 +259,11 @@ pub struct DataStreamWriter {
}
impl DataStreamWriter {
+ #[cfg(test)]
+ pub(crate) fn plan(&self) -> &LogicalPlan {
+ self.plan.as_ref()
+ }
+
/// Create a new DataStreamWriter.
pub(crate) fn new(session: SparkSession, plan: Arc<LogicalPlan>) -> Self {
DataStreamWriter {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
