Update for nested fields

dbt_dry_run/models/table.py

@@ -52,6 +52,11 @@ def validate_type_field(cls, field: str) -> BigQueryFieldType:
 TableField.model_rebuild()
 
 
+class FieldLineage(BaseModel):
+    lineage: str
+    field: TableField
+
+
 class Table(BaseModel):
     fields: List[TableField]
 

dbt_dry_run/schema_change_handlers.py

@@ -4,6 +4,7 @@
 from dbt_dry_run.models import OnSchemaChange, Table
 from dbt_dry_run.models.dry_run_result import DryRunResult
 from dbt_dry_run.models.report import DryRunStatus
+from dbt_dry_run.utils import find_missing_fields, build_predicted_fields
 
 
 def ignore_handler(dry_run_result: DryRunResult, target_table: Table) -> DryRunResult:
@@ -15,15 +16,11 @@ def append_new_columns_handler(
 ) -> DryRunResult:
     if dry_run_result.table is None:
         return dry_run_result
-
-    target_column_names = set(field.name for field in target_table.fields)
-    new_columns = [
-        new_field
-        for new_field in dry_run_result.table.fields
-        if new_field.name not in target_column_names
-    ]
-    final_fields = target_table.fields + new_columns
-    return dry_run_result.replace_table(Table(fields=final_fields))
+    missing_fields = find_missing_fields(
+        dry_run_result.table.fields, target_table.fields
+    )
+    predicted_fields = build_predicted_fields(target_table, missing_fields)
+    return dry_run_result.replace_table(Table(fields=predicted_fields))
 
 
 def sync_all_columns_handler(
@@ -43,7 +40,13 @@ def sync_all_columns_handler(
         for existing_field in target_table.fields
         if existing_field.name in predicted_column_names
     ]
-    final_fields = existing_columns + new_columns
+    final_field_names = set(field.name for field in existing_columns + new_columns)
+    missing_fields = find_missing_fields(dry_run_result.table.fields, existing_columns)
+    final_fields = build_predicted_fields(
+        target_table=target_table,
+        missing_fields=missing_fields,
+        included_top_level_field_names=final_field_names,
+    )
     return dry_run_result.replace_table(Table(fields=final_fields))