remove new field from SourceDefinition

core/dbt/artifacts/resources/v1/source_definition.py

@@ -58,7 +58,6 @@ class ParsedSourceMandatory(GraphResource, HasRelationMetadata):
 class SourceDefinition(ParsedSourceMandatory):
     quoting: Quoting = field(default_factory=Quoting)
     loaded_at_field: Optional[str] = None
-    loaded_at_field_present: Optional[bool] = None
     freshness: Optional[FreshnessThreshold] = None
     external: Optional[ExternalTable] = None
     description: str = ""

core/dbt/contracts/graph/nodes.py

@@ -1190,7 +1190,6 @@ def same_freshness(self, other: "SourceDefinition") -> bool:
         return (
             self.freshness == other.freshness
             and self.loaded_at_field == other.loaded_at_field
-            and self.loaded_at_field_present == other.loaded_at_field_present
             and True
         )
 

core/dbt/parser/schemas.py

@@ -123,10 +123,7 @@ def yaml_from_file(source_file: SchemaSourceFile) -> Optional[Dict[str, Any]]:
         # When loaded_loaded_at_field is defined as None or null, it shows up in
         # the dict but when it is not defined, it does not show up in the dict
         # We need to capture this to be able to override source level settings later.
-        # It defaults to False so only need to set to true here.
         for source in contents.get("sources", []):
-            if "loaded_at_field" in source:
-                source["loaded_at_field_present"] = True
             for table in source.get("tables", []):
                 if "loaded_at_field" in table:
                     table["loaded_at_field_present"] = True

core/dbt/parser/sources.py

@@ -142,8 +142,6 @@ def parse_source(self, target: UnpatchedSourceDefinition) -> SourceDefinition:
         else:
             loaded_at_field = source.loaded_at_field  # may be None, that's okay
 
-        loaded_at_field_present = table.loaded_at_field_present
-
         freshness = merge_freshness(source.freshness, table.freshness)
         quoting = source.quoting.merged(table.quoting)
         # path = block.path.original_file_path
@@ -191,7 +189,6 @@ def parse_source(self, target: UnpatchedSourceDefinition) -> SourceDefinition:
             meta=meta,
             loader=source.loader,
             loaded_at_field=loaded_at_field,
-            loaded_at_field_present=loaded_at_field_present,
             freshness=freshness,
             quoting=quoting,
             resource_type=NodeType.Source,

tests/functional/sources/test_source_loaded_at_field.py

@@ -15,7 +15,6 @@
     loaded_at_field: updated_at
     tables:
       - name: table1
-        identifier: example
         loaded_at_field: null
 """
 
@@ -32,7 +31,6 @@
     loaded_at_field: updated_at
     tables:
       - name: table1
-        identifier: example
 """
 
 loaded_at_field_defined_schema_yml = """
@@ -48,27 +46,25 @@
     loaded_at_field: updated_at
     tables:
       - name: table1
-        identifier: example
         loaded_at_field: updated_at_another_place
 """
 
 
-class TestLoadedAtField:
+class TestParsingLoadedAtField:
     @pytest.fixture(scope="class")
     def models(self):
         return {"schema.yml": loaded_at_field_null_schema_yml}
 
     def test_loaded_at_field(self, project):
-        # falls back to meteadata-based freshness but postgres doesn't support it
+        # test setting loaded_at_field to null explicitly at table level
         run_dbt(["parse"])
         manifest = get_manifest(project.project_root)
 
-        # test setting loaded_at_field at source level, should trickle to table
         assert "source.test.test_source.table1" in manifest.sources
         assert manifest.sources.get("source.test.test_source.table1").loaded_at_field is None
 
-        # test setting loaded_at_field at source level, and explicitly set to
-        # null at table level, end up with source level being None
+        # test setting loaded_at_field at source level, do not set at table level
+        # end up with source level loaded_at_field
         write_file(
             loaded_at_field_missing_schema_yml, project.project_root, "models", "schema.yml"
         )
@@ -79,7 +75,16 @@ def test_loaded_at_field(self, project):
             manifest.sources.get("source.test.test_source.table1").loaded_at_field == "updated_at"
         )
 
-        # test setting loaded_at_field at table level overrides source level
+        # test setting loaded_at_field to null explicitly again to make sure the change is picked up
+        # by parser
+        write_file(loaded_at_field_null_schema_yml, project.project_root, "models", "schema.yml")
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+
+        assert "source.test.test_source.table1" in manifest.sources
+        assert manifest.sources.get("source.test.test_source.table1").loaded_at_field is None
+
+        # test setting loaded_at_field at table level to a value - it should override source level
         write_file(
             loaded_at_field_defined_schema_yml, project.project_root, "models", "schema.yml"
         )