remove the is_internal part and clean up

Author: kumarUjjawal

datafusion/core/src/dataframe/mod.rs

@@ -2482,7 +2482,6 @@ impl DataFrame {
                             relation: None,
                             name: field.name().to_string(),
                             metadata: None,
-                            is_internal: false,
                         }),
                         Err(_) => col(field.name()),
                     }

datafusion/core/src/physical_planner.rs

@@ -2472,7 +2472,10 @@ pub fn create_window_expr(
 ) -> Result<Arc<dyn WindowExpr>> {
     // unpack aliased logical expressions, e.g. "sum(col) over () as total"
     let (name, e) = match e {
-        Expr::Alias(alias) => (alias.name.clone(), e.clone().unalias_nested().data),
+        Expr::Alias(alias) => (
+            alias.name.clone(),
+            alias.expr.as_ref().clone().unalias_nested().data,
+        ),
         _ => (e.schema_name().to_string(), e.clone()),
     };
     create_window_expr_with_name(&e, name, logical_schema, execution_props)
@@ -3269,9 +3272,10 @@ mod tests {
     use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs};
     use datafusion_expr::{
         Accumulator, AggregateUDF, AggregateUDFImpl, ExprFunctionExt, LogicalPlanBuilder,
-        Signature, TableSource, UserDefinedLogicalNodeCore, Volatility, col, lit,
+        Signature, TableSource, UserDefinedLogicalNodeCore, Volatility,
+        WindowFunctionDefinition, col, lit,
     };
-    use datafusion_functions_aggregate::count::count_all;
+    use datafusion_functions_aggregate::count::{count_all, count_udaf};
     use datafusion_functions_aggregate::expr_fn::sum;
     use datafusion_functions_aggregate::first_last::first_value_udaf;
     use datafusion_physical_expr::EquivalenceProperties;
@@ -3303,6 +3307,35 @@ mod tests {
         Ok(displayable(physical_plan.as_ref()).indent(true).to_string())
     }
 
+    #[test]
+    fn test_create_window_expr_unwraps_alias_with_metadata() -> Result<()> {
+        use std::collections::HashMap;
+
+        use datafusion_common::metadata::FieldMetadata;
+
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "column1",
+            DataType::Int64,
+            true,
+        )]));
+        let logical_schema = schema.as_ref().clone().to_dfschema_ref()?;
+        let metadata = FieldMetadata::from(HashMap::from([(
+            "some_key".to_string(),
+            "some_value".to_string(),
+        )]));
+        let expr = Expr::from(WindowFunction::new(
+            WindowFunctionDefinition::AggregateUDF(count_udaf()),
+            vec![col("column1")],
+        ))
+        .alias_with_metadata("window_alias", Some(metadata));
+
+        let window_expr =
+            create_window_expr(&expr, &logical_schema, &ExecutionProps::new())?;
+
+        assert_eq!(window_expr.name(), "window_alias");
+        Ok(())
+    }
+
     #[derive(Debug, Default)]
     struct NullAccumulator;
 

datafusion/expr/src/expr.rs

@@ -670,7 +670,6 @@ pub struct Alias {
     pub relation: Option<TableReference>,
     pub name: String,
     pub metadata: Option<FieldMetadata>,
-    pub is_internal: bool,
 }
 
 impl Hash for Alias {
@@ -679,7 +678,6 @@ impl Hash for Alias {
         self.relation.hash(state);
         self.name.hash(state);
         self.metadata.hash(state);
-        self.is_internal.hash(state);
     }
 }
 
@@ -712,22 +710,6 @@ impl Alias {
             relation: relation.map(|r| r.into()),
             name: name.into(),
             metadata: None,
-            is_internal: false,
-        }
-    }
-
-    #[doc(hidden)]
-    pub fn new_internal(
-        expr: Expr,
-        relation: Option<impl Into<TableReference>>,
-        name: impl Into<String>,
-    ) -> Self {
-        Self {
-            expr: Box::new(expr),
-            relation: relation.map(|r| r.into()),
-            name: name.into(),
-            metadata: None,
-            is_internal: true,
         }
     }
 
@@ -749,14 +731,12 @@ impl Alias {
             relation,
             name,
             metadata,
-            is_internal,
         } = self;
         Ok(Expr::Alias(Alias {
             expr: Box::new(f(*expr)?),
             relation,
             name,
             metadata,
-            is_internal,
         }))
     }
 }
@@ -1896,11 +1876,6 @@ impl Expr {
         Expr::Alias(Alias::new(self, None::<&str>, name.into()))
     }
 
-    #[doc(hidden)]
-    pub fn alias_internal(self, name: impl Into<String>) -> Expr {
-        Expr::Alias(Alias::new_internal(self, None::<&str>, name.into()))
-    }
-
     /// Return `self AS name` alias expression with metadata
     ///
     /// The metadata will be attached to the Arrow Schema field when the expression
@@ -1932,15 +1907,6 @@ impl Expr {
         Expr::Alias(Alias::new(self, relation, name.into()))
     }
 
-    #[doc(hidden)]
-    pub fn alias_qualified_internal(
-        self,
-        relation: Option<impl Into<TableReference>>,
-        name: impl Into<String>,
-    ) -> Expr {
-        Expr::Alias(Alias::new_internal(self, relation, name.into()))
-    }
-
     /// Return `self AS name` alias expression with a specific qualifier and metadata
     ///
     /// The metadata will be attached to the Arrow Schema field when the expression
@@ -2461,20 +2427,17 @@ impl NormalizeEq for Expr {
                     relation: self_relation,
                     name: self_name,
                     metadata: self_metadata,
-                    is_internal: self_is_internal,
                 }),
                 Expr::Alias(Alias {
                     expr: other_expr,
                     relation: other_relation,
                     name: other_name,
                     metadata: other_metadata,
-                    is_internal: other_is_internal,
                 }),
             ) => {
                 self_name == other_name
                     && self_relation == other_relation
                     && self_metadata == other_metadata
-                    && self_is_internal == other_is_internal
                     && self_expr.normalize_eq(other_expr)
             }
             (
@@ -2821,12 +2784,10 @@ impl HashNode for Expr {
                 relation,
                 name,
                 metadata,
-                is_internal,
             }) => {
                 relation.hash(state);
                 name.hash(state);
                 metadata.hash(state);
-                is_internal.hash(state);
             }
             Expr::Column(column) => {
                 column.hash(state);
@@ -4297,24 +4258,6 @@ mod test {
         );
     }
 
-    #[test]
-    fn test_internal_alias_changes_equality() {
-        let user_alias = col("id").alias("count(*)");
-        let internal_alias = col("id").alias_internal("count(*)");
-
-        assert_ne!(user_alias, internal_alias);
-        assert_ne!(internal_alias, user_alias);
-    }
-
-    #[test]
-    fn test_internal_alias_changes_normalized_identity() {
-        let user_alias = col("id").alias("count(*)");
-        let internal_alias = col("id").alias_internal("count(*)");
-
-        assert!(!user_alias.normalize_eq(&internal_alias));
-        assert!(!internal_alias.normalize_eq(&user_alias));
-    }
-
     #[test]
     fn test_unalias_nested_respects_user_metadata() {
         use std::collections::HashMap;
@@ -4332,9 +4275,6 @@ mod test {
         );
         assert_eq!(empty_metadata_alias.unalias_nested().data, base_expr);
 
-        let internal_alias = base_expr.clone().alias_internal("alias");
-        assert_eq!(internal_alias.unalias_nested().data, base_expr);
-
         let user_metadata = FieldMetadata::from(HashMap::from([(
             "some_key".to_string(),
             "some_value".to_string(),
@@ -4346,14 +4286,6 @@ mod test {
         let user_alias =
             Expr::Alias(user_alias.with_metadata(Some(user_metadata.clone())));
         assert_eq!(user_alias.clone().unalias_nested().data, user_alias);
-
-        let Expr::Alias(internal_alias) = base_expr.clone().alias_internal("alias")
-        else {
-            unreachable!();
-        };
-        let internal_alias =
-            Expr::Alias(internal_alias.with_metadata(Some(user_metadata.clone())));
-        assert_eq!(internal_alias.clone().unalias_nested().data, internal_alias);
     }
 
     fn wildcard_options(

datafusion/expr/src/expr_rewriter/mod.rs

@@ -309,7 +309,6 @@ pub enum SavedName {
     Saved {
         relation: Option<TableReference>,
         name: String,
-        is_internal_generated: bool,
     },
     /// Name is not preserved
     None,
@@ -345,17 +344,10 @@ impl NamePreserver {
                 Expr::Alias(alias) => SavedName::Saved {
                     relation: alias.relation.clone(),
                     name: alias.name.clone(),
-                    is_internal_generated: alias.is_internal,
                 },
                 _ => {
-                    // Rewrites preserve this output name for schema stability, but the
-                    // alias itself is planner-generated rather than user-written.
                     let (relation, name) = expr.qualified_name();
-                    SavedName::Saved {
-                        relation,
-                        name,
-                        is_internal_generated: true,
-                    }
+                    SavedName::Saved { relation, name }
                 }
             }
         } else {
@@ -368,18 +360,10 @@ impl SavedName {
     /// Ensures the qualified name of the rewritten expression is preserved
     pub fn restore(self, expr: Expr) -> Expr {
         match self {
-            SavedName::Saved {
-                relation,
-                name,
-                is_internal_generated,
-            } => {
+            SavedName::Saved { relation, name } => {
                 let (new_relation, new_name) = expr.qualified_name();
                 if new_relation != relation || new_name != name {
-                    if is_internal_generated {
-                        expr.alias_qualified_internal(relation, name)
-                    } else {
-                        expr.alias_qualified(relation, name)
-                    }
+                    expr.alias_qualified(relation, name)
                 } else {
                     expr
                 }
@@ -567,20 +551,6 @@ mod test {
         );
     }
 
-    #[test]
-    fn test_name_preserver_marks_non_alias_restore_as_internal() {
-        let saved_name = NamePreserver::new_for_projection().save(&col("a"));
-        let restored = saved_name.restore(col("b"));
-
-        let Expr::Alias(alias) = restored else {
-            panic!("expected alias");
-        };
-
-        assert!(alias.is_internal);
-        assert_eq!(alias.name, "a");
-        assert_eq!(*alias.expr, col("b"));
-    }
-
     /// rewrites `expr_from` to `rewrite_to` while preserving the original qualified name
     /// by using the `NamePreserver`
     fn test_rewrite(expr_from: Expr, rewrite_to: Expr) {

datafusion/expr/src/expr_schema.rs

@@ -1070,14 +1070,14 @@ mod tests {
     }
 
     #[test]
-    fn test_internal_alias_metadata_is_preserved_in_field_metadata() {
+    fn test_alias_metadata_is_preserved_in_field_metadata() {
         let schema = MockExprSchema::new().with_data_type(DataType::Int32);
         let alias_metadata = FieldMetadata::from(HashMap::from([(
             "some_key".to_string(),
             "some_value".to_string(),
         )]));
 
-        let Expr::Alias(alias) = col("foo").alias_internal("alias") else {
+        let Expr::Alias(alias) = col("foo").alias("alias") else {
             unreachable!();
         };
         let expr = Expr::Alias(alias.with_metadata(Some(alias_metadata.clone())));

datafusion/expr/src/tree_node.rs

@@ -158,14 +158,12 @@ impl TreeNode for Expr {
                 relation,
                 name,
                 metadata,
-                is_internal,
             }) => expr.map_elements(f)?.update_data(|expr| {
                 Expr::Alias(Alias {
                     expr,
                     relation,
                     name,
                     metadata,
-                    is_internal,
                 })
             }),
             Expr::InSubquery(InSubquery {

datafusion/functions-aggregate/src/count.rs

@@ -102,7 +102,7 @@ pub fn count_distinct(expr: Expr) -> Expr {
 /// let expr = col(expr.schema_name().to_string());
 /// ```
 pub fn count_all() -> Expr {
-    count(Expr::Literal(COUNT_STAR_EXPANSION, None)).alias_internal("count(*)")
+    count(Expr::Literal(COUNT_STAR_EXPANSION, None)).alias("count(*)")
 }
 
 /// Creates window aggregation to count all rows.

datafusion/optimizer/src/analyzer/resolve_grouping_function.rs

@@ -111,7 +111,7 @@ fn replace_grouping_exprs(
                     &group_expr_to_bitmap_index,
                     grouping_id_type,
                 )?;
-                projection_exprs.push(Expr::Alias(Alias::new_internal(
+                projection_exprs.push(Expr::Alias(Alias::new(
                     grouping_expr,
                     column.relation,
                     column.name,

datafusion/optimizer/src/analyzer/type_coercion.rs

@@ -1275,21 +1275,19 @@ fn project_with_column_index(
         .enumerate()
         .map(|(i, e)| match e {
             Expr::Alias(Alias { ref name, .. }) if name != schema.field(i).name() => {
-                Ok(e.unalias().alias_internal(schema.field(i).name()))
+                Ok(e.unalias().alias(schema.field(i).name()))
             }
             Expr::Column(Column {
                 relation: _,
                 ref name,
                 spans: _,
-            }) if name != schema.field(i).name() => {
-                Ok(e.alias_internal(schema.field(i).name()))
-            }
+            }) if name != schema.field(i).name() => Ok(e.alias(schema.field(i).name())),
             Expr::Alias { .. } | Expr::Column { .. } => Ok(e),
             #[expect(deprecated)]
             Expr::Wildcard { .. } => {
                 plan_err!("Wildcard should be expanded before type coercion")
             }
-            _ => Ok(e.alias_internal(schema.field(i).name())),
+            _ => Ok(e.alias(schema.field(i).name())),
         })
         .collect::<Result<Vec<_>>>()?;