From aac9704ff4a941169fcf542badf2159b48447970 Mon Sep 17 00:00:00 2001 From: Gene Bordegaray Date: Thu, 27 Aug 2026 09:42:43 -0400 Subject: [PATCH 1/5] fix: preserve projection metadata --- datafusion/physical-plan/src/projection.rs | 256 ++++++++++++++++++++- 1 file changed, 244 insertions(+), 12 deletions(-) diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index c4096457c168a..3bedcfa3d216e 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -83,6 +83,9 @@ pub struct ProjectionExec { metrics: ExecutionPlanMetricsSet, /// Cache holding plan properties like equivalences, output partitioning etc. cache: Arc, + /// Whether the output metadata differs from the metadata derived from the + /// projection expressions and input schema. + overrides_metadata: bool, } impl ProjectionExec { @@ -144,7 +147,7 @@ impl ProjectionExec { let expr_arc = expr.into_iter().map(Into::into).collect::>(); let projection = ProjectionExprs::from_expressions(expr_arc); let projector = projection.make_projector(&input_schema)?; - Self::try_from_projector(projector, input) + Self::try_from_projector(projector, input, false) } /// Create a projection using field and schema metadata from @@ -172,14 +175,17 @@ impl ProjectionExec { let projection = ProjectionExprs::from_expressions(expr_arc); let projector = projection .make_projector_with_schema_metadata(&input_schema, projected_schema)?; - Self::try_from_projector(projector, input) + let overrides_metadata = + Self::compute_overrides_metadata(&projector, &input_schema)?; + Self::try_from_projector(projector, input, overrides_metadata) } fn try_from_projector( projector: Projector, input: Arc, + overrides_metadata: bool, ) -> Result { - Self::try_from_projector_with_eq_group(projector, input, None) + Self::try_from_projector_with_eq_group(projector, input, None, overrides_metadata) } /// As [`Self::try_from_projector`], but `reuse_from` may carry the previous @@ -204,6 +210,7 @@ impl ProjectionExec { projector: Projector, input: Arc, reuse_from: Option<(&EquivalenceProperties, &EquivalenceProperties)>, + overrides_metadata: bool, ) -> Result { // Construct a map from the input expressions to the output expression of the Projection let projection_mapping = @@ -219,6 +226,7 @@ impl ProjectionExec { input, metrics: ExecutionPlanMetricsSet::new(), cache: Arc::new(cache), + overrides_metadata, }) } @@ -270,6 +278,33 @@ impl ProjectionExec { )) } + /// Returns whether `projector`'s output metadata differs from the metadata + /// derived from its expressions and `input_schema`. + fn compute_overrides_metadata( + projector: &Projector, + input_schema: &Schema, + ) -> Result { + let output_schema = projector.output_schema(); + if input_schema.metadata() != output_schema.metadata() { + return Ok(true); + } + for (projection, output_field) in + projector.projection().iter().zip(output_schema.fields()) + { + let derived_field = projection.expr.return_field(input_schema)?; + if derived_field.metadata() != output_field.metadata() { + return Ok(true); + } + } + Ok(false) + } + + /// Returns whether this projection's output metadata differs from the + /// metadata derived when the projection was constructed. + fn overrides_metadata(&self) -> bool { + self.overrides_metadata + } + /// Collect reverse alias mapping from projection expressions. /// The result hash map is a map from aliased Column in parent to original expr. fn collect_reverse_alias( @@ -406,10 +441,17 @@ impl ExecutionPlan for ProjectionExec { self.input.equivalence_properties(), self.cache.equivalence_properties(), )); + let input = children.swap_remove(0); + let projector = self.projector.clone(); + let overrides_metadata = ProjectionExec::compute_overrides_metadata( + &projector, + input.schema().as_ref(), + )?; ProjectionExec::try_from_projector_with_eq_group( - self.projector.clone(), - children.swap_remove(0), + projector, + input, reuse_from, + overrides_metadata, ) .map(|p| Arc::new(p) as _) } @@ -631,6 +673,8 @@ impl ExecutionPlan for ProjectionExec { metrics: _, // Derived plan properties, recomputed on decode. cache: _, + // Derived metadata comparison, recomputed with the projector. + overrides_metadata: _, } = self; let projection_exprs = projector.projection().as_ref(); let input = ctx.encode_child(input)?; @@ -1019,7 +1063,12 @@ pub fn remove_unnecessary_projections( if is_projection_removable(projection) { return Ok(Transformed::yes(Arc::clone(projection.input()))); } - // If it does, check if we can push it under its child(ren): + // Swapping a projection with observable metadata can change query results + // by changing the metadata visible to its child expressions. + if projection.overrides_metadata() { + return Ok(Transformed::no(plan)); + } + // Otherwise, check if we can push it under its child(ren): projection .input() .try_swapping_with_projection(projection)? @@ -1031,6 +1080,7 @@ pub fn remove_unnecessary_projections( /// Compare the inputs and outputs of the projection. All expressions must be /// columns without alias, and projection does not change the order of fields. +/// The input and output schemas must also match exactly to preserve metadata. /// For example, if the input schema is `a, b`, `SELECT a, b` is removable, /// but `SELECT b, a` and `SELECT a+1, b` and `SELECT a AS c, b` are not. fn is_projection_removable(projection: &ProjectionExec) -> bool { @@ -1041,6 +1091,7 @@ fn is_projection_removable(projection: &ProjectionExec) -> bool { }; col.name() == proj_expr.alias && col.index() == idx }) && exprs.len() == projection.input().schema().fields().len() + && projection.schema() == projection.input().schema() } /// Given the expression set of a projection, checks if the projection causes @@ -1074,13 +1125,17 @@ pub fn new_projections_for_columns( } /// Creates a new [`ProjectionExec`] instance with the given child plan and -/// projected expressions. +/// projected expressions, preserving the original output metadata. pub fn make_with_child( projection: &ProjectionExec, child: &Arc, ) -> Result> { - ProjectionExec::try_new(projection.expr().to_vec(), Arc::clone(child)) - .map(|e| Arc::new(e) as _) + ProjectionExec::try_new_with_schema_metadata( + projection.expr().to_vec(), + Arc::clone(child), + projection.schema().as_ref(), + ) + .map(|e| Arc::new(e) as _) } /// Returns `true` if all the expressions in the argument are `Column`s. @@ -1331,12 +1386,20 @@ pub fn update_join_filter( fn try_collapse_projection_chain( outer: &ProjectionExec, ) -> Result>> { + if outer.overrides_metadata() { + return Ok(None); + } + let mut current_exprs: Vec = outer.expr().to_vec(); let mut current_input: Arc = Arc::clone(outer.input()); let mut column_ref_map: HashMap = HashMap::new(); let mut collapsed_any = false; 'outer: while let Some(inner_proj) = current_input.downcast_ref::() { + if inner_proj.overrides_metadata() { + break; + } + // Collect the column references usage in the outer projection. column_ref_map.clear(); for proj_expr in ¤t_exprs { @@ -1386,8 +1449,13 @@ fn try_collapse_projection_chain( } // To unify 3 or more sequential projections: + // Preserve the outer projection's output metadata. let unified: Arc = - Arc::new(ProjectionExec::try_new(current_exprs, current_input)?); + Arc::new(ProjectionExec::try_new_with_schema_metadata( + current_exprs, + current_input, + outer.schema().as_ref(), + )?); remove_unnecessary_projections(unified).data().map(Some) } @@ -1517,13 +1585,16 @@ mod tests { use crate::test; use crate::test::exec::StatisticsExec; + use arrow::array::StringArray; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::ScalarValue; use datafusion_common::stats::{ColumnStatistics, Precision, Statistics}; - use datafusion_expr::Operator; + use datafusion_expr::{Operator, ScalarUDF}; + use datafusion_functions::core::arrow_metadata::ArrowMetadataFunc; + use datafusion_physical_expr::ScalarFunctionExpr; use datafusion_physical_expr::expressions::{ - BinaryExpr, Column, DynamicFilterPhysicalExpr, Literal, binary, col, lit, + BinaryExpr, Column, DynamicFilterPhysicalExpr, Literal, binary, col, is_null, lit, }; #[test] @@ -1566,6 +1637,166 @@ mod tests { Ok(()) } + fn identity_projection_with_metadata( + input: Arc, + field_metadata: HashMap, + schema_metadata: HashMap, + ) -> Result> { + let metadata_schema = Schema::new_with_metadata( + vec![Field::new("i", DataType::Int32, true).with_metadata(field_metadata)], + schema_metadata, + ); + Ok(Arc::new(ProjectionExec::try_new_with_schema_metadata( + [ProjectionExpr { + expr: Arc::new(Column::new("i", 0)), + alias: "i".to_string(), + }], + input, + &metadata_schema, + )?)) + } + + #[test] + fn test_field_metadata_projection_is_not_removable() -> Result<()> { + let projection = identity_projection_with_metadata( + test::scan_partitioned(1), + HashMap::from([("event_field".to_string(), "true".to_string())]), + HashMap::new(), + )?; + let expected_schema = projection.schema(); + + let optimized = remove_unnecessary_projections(projection)?.data; + + assert!(optimized.downcast_ref::().is_some()); + assert_eq!(optimized.schema(), expected_schema); + Ok(()) + } + + #[test] + fn test_schema_metadata_projection_is_not_removable() -> Result<()> { + let projection = identity_projection_with_metadata( + test::scan_partitioned(1), + HashMap::new(), + HashMap::from([("schema-key".to_string(), "schema-value".to_string())]), + )?; + let expected_schema = projection.schema(); + + let optimized = remove_unnecessary_projections(projection)?.data; + + assert!(optimized.downcast_ref::().is_some()); + assert_eq!(optimized.schema(), expected_schema); + Ok(()) + } + + #[test] + fn test_make_with_child_preserves_output_metadata() -> Result<()> { + let projection = identity_projection_with_metadata( + test::scan_partitioned(1), + HashMap::from([("event_field".to_string(), "true".to_string())]), + HashMap::from([("schema-key".to_string(), "schema-value".to_string())]), + )?; + let projection = projection + .downcast_ref::() + .expect("test plan should be a ProjectionExec"); + + let rebuilt = make_with_child(projection, &test::scan_partitioned(1))?; + + assert_eq!(rebuilt.schema(), projection.schema()); + Ok(()) + } + + #[tokio::test] + async fn test_metadata_observing_parent_blocks_projection_collapse() -> Result<()> { + let inner = identity_projection_with_metadata( + test::scan_partitioned(1), + HashMap::from([("event_field".to_string(), "true".to_string())]), + HashMap::new(), + )?; + let arrow_metadata = ScalarFunctionExpr::new( + "arrow_metadata", + Arc::new(ScalarUDF::new_from_impl(ArrowMetadataFunc::new())), + vec![ + Arc::new(Column::new("i", 0)), + Arc::new(Literal::new(ScalarValue::Utf8(Some( + "event_field".to_string(), + )))), + ], + Arc::new(Field::new("arrow_metadata", DataType::Utf8, true)), + Arc::new(ConfigOptions::default()), + ); + let outer: Arc = Arc::new(ProjectionExec::try_new( + [ProjectionExpr { + expr: Arc::new(arrow_metadata), + alias: "metadata".to_string(), + }], + inner, + )?); + + let outer_projection = outer + .downcast_ref::() + .expect("test plan should be a ProjectionExec"); + assert!(try_collapse_projection_chain(outer_projection)?.is_none()); + + let optimized = remove_unnecessary_projections(outer)?.data; + let batches = + collect(optimized.execute(0, Arc::new(TaskContext::default()))?).await?; + let values = batches[0] + .column(0) + .as_any() + .downcast_ref::() + .expect("metadata expression should return Utf8"); + assert_eq!(values.value(0), "true"); + Ok(()) + } + + #[tokio::test] + async fn test_metadata_observing_filter_blocks_projection_pushdown() -> Result<()> { + let widened: Arc = Arc::new(ProjectionExec::try_new( + [ + ProjectionExpr { + expr: Arc::new(Column::new("i", 0)), + alias: "i".to_string(), + }, + ProjectionExpr { + expr: Arc::new(Column::new("i", 0)), + alias: "j".to_string(), + }, + ], + test::scan_partitioned(1), + )?); + let arrow_metadata = Arc::new(ScalarFunctionExpr::new( + "arrow_metadata", + Arc::new(ScalarUDF::new_from_impl(ArrowMetadataFunc::new())), + vec![ + Arc::new(Column::new("i", 0)), + Arc::new(Literal::new(ScalarValue::Utf8(Some( + "event_field".to_string(), + )))), + ], + Arc::new(Field::new("arrow_metadata", DataType::Utf8, true)), + Arc::new(ConfigOptions::default()), + )); + let filter: Arc = + Arc::new(FilterExec::try_new(is_null(arrow_metadata)?, widened)?); + let projection = identity_projection_with_metadata( + filter, + HashMap::from([("event_field".to_string(), "true".to_string())]), + HashMap::new(), + )?; + let expected_schema = projection.schema(); + + let optimized = remove_unnecessary_projections(projection)?.data; + assert_eq!(optimized.schema(), expected_schema); + let batches = + collect(optimized.execute(0, Arc::new(TaskContext::default()))?).await?; + + assert_eq!( + batches.iter().map(|batch| batch.num_rows()).sum::(), + 100 + ); + Ok(()) + } + #[test] fn test_collect_column_indices() -> Result<()> { let expr = Arc::new(BinaryExpr::new( @@ -2510,6 +2741,7 @@ mod tests { let recomputed = ProjectionExec::try_from_projector( projection.projector.clone(), tightened_child, + projection.overrides_metadata(), )?; assert_same_properties(replaced.as_ref(), &recomputed); From 64e9b97d648d4fac1234349d9d9eb94addcff5b2 Mon Sep 17 00:00:00 2001 From: Gene Bordegaray Date: Wed, 2 Sep 2026 20:41:16 -0400 Subject: [PATCH 2/5] test: add mutation testing coverage --- datafusion/physical-plan/src/projection.rs | 35 ++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index 3bedcfa3d216e..ddf75d491e626 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -1688,6 +1688,41 @@ mod tests { Ok(()) } + #[test] + fn test_replace_children_recomputes_metadata_override() -> Result<()> { + let field_metadata = + HashMap::from([("event_field".to_string(), "true".to_string())]); + let projection = identity_projection_with_metadata( + test::scan_partitioned(1), + field_metadata.clone(), + HashMap::new(), + )?; + assert!( + projection + .downcast_ref::() + .expect("test plan should be a ProjectionExec") + .overrides_metadata() + ); + + let replacement_schema = Arc::new(Schema::new(vec![ + Field::new("i", DataType::Int32, true).with_metadata(field_metadata), + ])); + let replacement: Arc = + Arc::new(EmptyExec::new(replacement_schema)); + let replaced = projection.replace_children( + vec![replacement], + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), + )?; + + assert!( + !replaced + .downcast_ref::() + .expect("replaced plan should be a ProjectionExec") + .overrides_metadata() + ); + Ok(()) + } + #[test] fn test_make_with_child_preserves_output_metadata() -> Result<()> { let projection = identity_projection_with_metadata( From 90f6c365dd814ff710a15f39d52a8a32955e0430 Mon Sep 17 00:00:00 2001 From: Tim Saucer Date: Fri, 4 Sep 2026 12:56:18 -0400 Subject: [PATCH 3/5] test: cover schema-level metadata override in ProjectionExec `compute_overrides_metadata` compares schema-level metadata separately from per-field metadata, but nothing exercised that first comparison: deleting it left every test passing. The gap was reachable. A projection that overrides only schema-level metadata has field metadata matching what its expressions derive, so the field loop never flags it, and `is_projection_removable` only declines to remove it. With the schema-level comparison gone it would be handed to `try_swapping_with_projection` and silently lose its metadata. Add a test that pins it. The identity shape does not narrow the schema, so `FilterExec::try_swapping_with_projection` falls through to `try_embed_projection`, which rebuilds the plan from the projection expressions alone and drops the schema metadata. Extending the existing `test_schema_metadata_projection_is_not_removable` would not have worked: `TestMemoryExec` has no `try_swapping_with_projection` impl, so no swap is ever attempted there. Co-Authored-By: Claude Opus 5 (1M context) --- datafusion/physical-plan/src/projection.rs | 55 ++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index ddf75d491e626..f2c58c0c710ec 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -1832,6 +1832,61 @@ mod tests { Ok(()) } + /// A projection that only overrides *schema-level* metadata must still be + /// recognized as overriding, so it is never handed to an operator that + /// re-derives its schema. + /// + /// The identity shape below does not narrow the schema, so + /// [`FilterExec::try_swapping_with_projection`] falls through to + /// [`try_embed_projection`], which rebuilds the plan from the projection + /// expressions alone and would drop the schema metadata. Only the + /// schema-level comparison in + /// [`ProjectionExec::compute_overrides_metadata`] catches this case: every + /// field's metadata matches what the expressions derive, and + /// [`is_projection_removable`] merely declines to remove the projection. + #[tokio::test] + async fn test_schema_level_metadata_blocks_projection_embedding() -> Result<()> { + let scan = test::scan_partitioned(1); + let predicate = binary( + col("i", &scan.schema())?, + Operator::Gt, + lit(ScalarValue::Int32(Some(-1))), + &scan.schema(), + )?; + let filter: Arc = + Arc::new(FilterExec::try_new(predicate, scan)?); + let projection = identity_projection_with_metadata( + filter, + HashMap::new(), + HashMap::from([("schema-key".to_string(), "schema-value".to_string())]), + )?; + // Field metadata alone cannot flag this projection -- the override is + // carried entirely by the schema-level entry. + let projection_exec = projection + .downcast_ref::() + .expect("test plan should be a ProjectionExec"); + assert!(projection_exec.overrides_metadata()); + let expected_schema = projection.schema(); + + let optimized = remove_unnecessary_projections(projection)?.data; + + assert!(optimized.downcast_ref::().is_some()); + assert_eq!(optimized.schema(), expected_schema); + assert_eq!( + optimized.schema().metadata(), + &HashMap::from([("schema-key".to_string(), "schema-value".to_string())]) + ); + + // The projection is still evaluated, so the filter's rows survive it. + let batches = + collect(optimized.execute(0, Arc::new(TaskContext::default()))?).await?; + assert_eq!( + batches.iter().map(|batch| batch.num_rows()).sum::(), + 100 + ); + Ok(()) + } + #[test] fn test_collect_column_indices() -> Result<()> { let expr = Arc::new(BinaryExpr::new( From 4e956f16b6ef5b2558c4b4c8efd2485a85f75ac3 Mon Sep 17 00:00:00 2001 From: Tim Saucer Date: Fri, 4 Sep 2026 12:58:00 -0400 Subject: [PATCH 4/5] refactor: drop unreachable metadata guard in projection collapse `try_collapse_projection_chain` bailed out when the outer projection overrode metadata. That branch cannot be taken. The function's only production caller is `ProjectionExec::try_swapping_with_projection`, which is reached exclusively from `remove_unnecessary_projections` after it has already returned early on an overriding projection. Deleting the guard leaves every test in the workspace passing, including the sqllogictest suite. The check was also redundant on its own terms. An inner projection that overrides metadata breaks the loop, so every collapsed projection derives the same metadata its expressions would, and the unified projection is built with the outer projection's output schema regardless. Replace it with a doc comment recording why the caller's guard is sufficient, so the invariant is stated rather than re-implemented. Co-Authored-By: Claude Opus 5 (1M context) --- datafusion/physical-plan/src/projection.rs | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index f2c58c0c710ec..28c1548dd6acb 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -1383,13 +1383,16 @@ pub fn update_join_filter( /// Collapse a chain of consecutive [`ProjectionExec`]s into one. Returns /// `None` if nothing could be merged. +/// +/// `outer` is not checked for a metadata override here. Its only production +/// caller is [`ExecutionPlan::try_swapping_with_projection`] on +/// [`ProjectionExec`], reached exclusively from +/// [`remove_unnecessary_projections`], which already bails out on such a +/// projection. The unified projection below is still built with `outer`'s output +/// schema so that its metadata survives regardless. fn try_collapse_projection_chain( outer: &ProjectionExec, ) -> Result>> { - if outer.overrides_metadata() { - return Ok(None); - } - let mut current_exprs: Vec = outer.expr().to_vec(); let mut current_input: Arc = Arc::clone(outer.input()); let mut column_ref_map: HashMap = HashMap::new(); From 889cdfb2236cdee777cc96f79bb052be3baeb6d9 Mon Sep 17 00:00:00 2001 From: Gene Bordegaray Date: Fri, 4 Sep 2026 13:58:54 -0400 Subject: [PATCH 5/5] clarify comments --- datafusion/physical-plan/src/projection.rs | 29 ++++++---------------- 1 file changed, 7 insertions(+), 22 deletions(-) diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index 28c1548dd6acb..7851e5705934e 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -1384,12 +1384,10 @@ pub fn update_join_filter( /// Collapse a chain of consecutive [`ProjectionExec`]s into one. Returns /// `None` if nothing could be merged. /// -/// `outer` is not checked for a metadata override here. Its only production -/// caller is [`ExecutionPlan::try_swapping_with_projection`] on -/// [`ProjectionExec`], reached exclusively from -/// [`remove_unnecessary_projections`], which already bails out on such a -/// projection. The unified projection below is still built with `outer`'s output -/// schema so that its metadata survives regardless. +/// The projection-removal optimizer checks `outer.overrides_metadata()` before +/// reaching this helper. The unified projection also keeps `outer`'s schema, so +/// collapsing cannot lose its output metadata. Inner projections still need the +/// check below because outer expressions may observe their metadata. fn try_collapse_projection_chain( outer: &ProjectionExec, ) -> Result>> { @@ -1835,18 +1833,8 @@ mod tests { Ok(()) } - /// A projection that only overrides *schema-level* metadata must still be - /// recognized as overriding, so it is never handed to an operator that - /// re-derives its schema. - /// - /// The identity shape below does not narrow the schema, so - /// [`FilterExec::try_swapping_with_projection`] falls through to - /// [`try_embed_projection`], which rebuilds the plan from the projection - /// expressions alone and would drop the schema metadata. Only the - /// schema-level comparison in - /// [`ProjectionExec::compute_overrides_metadata`] catches this case: every - /// field's metadata matches what the expressions derive, and - /// [`is_projection_removable`] merely declines to remove the projection. + // A schema-only metadata override must block projection embedding. The filter + // rebuilds the schema from expressions and would otherwise drop this metadata. #[tokio::test] async fn test_schema_level_metadata_blocks_projection_embedding() -> Result<()> { let scan = test::scan_partitioned(1); @@ -1863,8 +1851,7 @@ mod tests { HashMap::new(), HashMap::from([("schema-key".to_string(), "schema-value".to_string())]), )?; - // Field metadata alone cannot flag this projection -- the override is - // carried entirely by the schema-level entry. + // Field metadata matches, so this checks the schema-level comparison. let projection_exec = projection .downcast_ref::() .expect("test plan should be a ProjectionExec"); @@ -1873,14 +1860,12 @@ mod tests { let optimized = remove_unnecessary_projections(projection)?.data; - assert!(optimized.downcast_ref::().is_some()); assert_eq!(optimized.schema(), expected_schema); assert_eq!( optimized.schema().metadata(), &HashMap::from([("schema-key".to_string(), "schema-value".to_string())]) ); - // The projection is still evaluated, so the filter's rows survive it. let batches = collect(optimized.execute(0, Arc::new(TaskContext::default()))?).await?; assert_eq!(