Fix remaining

Author: shangyian

datajunction-server/datajunction_server/internal/deployment/orchestrator.py

@@ -49,6 +49,7 @@
 from datajunction_server.internal.impact import (
     compute_impact,
     _ExplicitDiffSpec,
+    _ProposedColumn,
     _build_proposed_columns,
     _detect_source_column_changes,
     ColumnChangeType,
@@ -417,12 +418,20 @@ async def _compute_pre_deploy_diffs(
                 if link.foreign_key_column_names & columns_removed:
                     dim_links_removed.add(link.dimension.name)
 
+            if proposed_columns is None:
+                # No query change (e.g. metadata-only update): use current columns
+                # so the BFS sees the real column state and doesn't falsely
+                # invalidate downstream nodes.
+                proposed_columns = [
+                    _ProposedColumn(name=c.name, type=c.type)
+                    for c in pre_node.current.columns
+                ]
             specs[name] = _ExplicitDiffSpec(
                 node_type=node_spec.node_type,
                 explicit_columns_removed=columns_removed,
                 explicit_columns_changed=columns_changed,
                 dim_links_removed=dim_links_removed,
-                explicit_proposed_columns=proposed_columns or [],
+                explicit_proposed_columns=proposed_columns,
             )
         return specs
 

datajunction-server/datajunction_server/internal/deployment/validation.py

@@ -15,7 +15,10 @@
 
 from datajunction_server.api.helpers import _resolve_required_dimensions
 from datajunction_server.database.dimensionlink import DimensionLink
-from datajunction_server.internal.validation import validate_metric_query
+from datajunction_server.internal.validation import (
+    update_ast_column_types,
+    validate_metric_query,
+)
 from datajunction_server.sql.dag import get_dimensions
 from datajunction_server.database.node import Node, NodeRevision
 from datajunction_server.models.node import NodeStatus, NodeType
@@ -234,7 +237,7 @@ def _validate_dimension_link_specs(
                                         code=ErrorCode.INVALID_SQL_QUERY,
                                         message=(
                                             f"join_on for dimension link '{dim_name}' references"
-                                            f" unknown table '{node_ref}' — expected"
+                                            f" unknown node '{node_ref}' — expected"
                                             f" '{node_name}' or '{dim_name}'"
                                         ),
                                     ),
@@ -302,6 +305,7 @@ async def validate_query_node(
             await parsed_ast.bake_ctes().extract_dependencies(
                 self.context.compile_context,
             )
+            update_ast_column_types(parsed_ast)
             parsed_ast.select.add_aliases_to_unnamed_columns()
 
             # If _skip_validation is set (e.g., copying from validated source),
@@ -324,6 +328,7 @@ async def validate_query_node(
                     ]
                     if err is not None
                 ] + type_inference_errors
+                print("parsed", spec.rendered_name, "and got", errors)
 
             dep_names = self.context.node_graph.get(spec.rendered_name, [])
             invalid_parents = [
@@ -617,7 +622,12 @@ def _create_column_spec(
         if existing_spec and existing_spec.type:
             column_type = existing_spec.type
         else:
-            column_type = str(ast_column.type)
+            inferred = ast_column.type
+            if inferred is None:
+                raise TypeError(
+                    f"Cannot resolve type of column `{column_name}` in node",
+                )
+            column_type = str(inferred)
 
         if existing_spec:
             return ColumnSpec(

datajunction-server/datajunction_server/internal/impact.py

@@ -546,6 +546,7 @@ async def _validate_downstream_node(
         # would return the existing column type for every projected column even when
         # the upstream column was removed via column_overrides, masking the impact.
         spec = spec.model_copy(update={"columns": []})
+        print("upstream_proposed", upstream_proposed)
         results = await bulk_validate_node_data(
             [spec],
             session,

datajunction-server/datajunction_server/internal/validation.py

@@ -25,6 +25,22 @@
 from datajunction_server.sql.parsing.types import ListType, MapType, StructType
 
 
+def update_ast_column_types(node: ast.Node) -> None:
+    """Recursively update AST Column._type from parsed database column types.
+
+    Call this after re-parsing parent column types (via parse_rule) so that
+    type inference on the query AST sees the properly-typed objects rather than
+    raw ColumnType strings.
+    """
+    if isinstance(node, ast.Column) and node._table and hasattr(node._table, "dj_node"):
+        for db_col in node._table.dj_node.columns:
+            if db_col.name == node.name.name:
+                node._type = db_col.type
+                break
+    for child in node.children:
+        update_ast_column_types(child)
+
+
 @dataclass
 class NodeValidator:
     """
@@ -135,26 +151,9 @@ async def validate_node_data(
                     # If parsing fails, leave the original type
                     pass
 
-    # Update AST Column nodes with the newly parsed types
+    # Update AST Column nodes with the newly parsed types.
     # During extract_dependencies/compilation, AST Columns were created with _type from
     # database columns. Now that we've parsed the types, update the AST Column cache.
-    def update_ast_column_types(node):
-        """Recursively update AST Column _type from parsed database columns"""
-        if (
-            isinstance(node, ast.Column)
-            and node._table
-            and hasattr(node._table, "dj_node")
-        ):
-            # Find the corresponding database column
-            for db_col in node._table.dj_node.columns:
-                if db_col.name == node.name.name:
-                    # Update the cached type
-                    node._type = db_col.type
-                    break
-        # Recursively update child nodes
-        for child in node.children:
-            update_ast_column_types(child)
-
     update_ast_column_types(query_ast)
 
     # Add aliases for any unnamed columns and confirm that all column types can be inferred

datajunction-server/datajunction_server/sql/parsing/ast.py

@@ -1728,7 +1728,10 @@ async def compile(self, ctx: CompileContext):
                 self.set_dj_node(dj_node)
             # Use proposed column overrides when available (impact preview / dryrun).
             # Applied after dj_node is set so is_compiled() stays consistent.
-            if table_name in ctx.column_overrides:
+            if (
+                table_name in ctx.column_overrides
+                and ctx.column_overrides[table_name] is not None
+            ):
                 self._columns = [
                     Column(Name(col.name), _type=col.type, _table=self)
                     for col in ctx.column_overrides[table_name]