implements missing tests

Author: rudolfix

dlt/common/schema/utils.py

@@ -763,9 +763,8 @@ def get_inherited_table_hint(
     tables: TSchemaTables, table_name: str, table_hint_name: str, allow_none: bool = False
 ) -> Any:
     table = tables.get(table_name, {})
-    hint = table.get(table_hint_name)
-    if hint:
-        return hint
+    if table_hint_name in table:
+        return table[table_hint_name]  # type: ignore[literal-required]
 
     if is_nested_table(table):
         return get_inherited_table_hint(tables, table.get("parent"), table_hint_name, allow_none)

dlt/destinations/utils.py

@@ -120,7 +120,6 @@ def verify_schema_merge_disposition(
                         " merge keys defined."
                         " dlt will fall back to `append` for this table."
                     )
-                    raise RuntimeError()
             elif merge_strategy == "upsert":
                 if not has_column_with_prop(table, "primary_key"):
                     exception_log.append(

dlt/extract/extractors.py

@@ -232,13 +232,13 @@ def _compute_tables(
     ) -> List[TTableSchema]:
         """Computes a schema for a new or dynamic table and normalizes identifiers"""
         root_table_schema = resource.compute_table_schema(items, meta)
+        nested_tables_schema = resource.compute_nested_table_schemas(
+            root_table_schema["name"], self.naming, items, meta
+        )
         # we need to re-normalize name to support legacy normalization mode which we will
         # drop in next major version
         # TODO: drop in 2.0 also drop SCHEMA__USE_BREAK_PATH_ON_NORMALIZE
         root_table_schema["name"] = self._normalize_table_identifier(root_table_schema["name"])
-        nested_tables_schema = resource.compute_nested_table_schemas(
-            root_table_schema["name"], self.naming, items, meta
-        )
         return [
             utils.normalize_table_identifiers(table_schema, self.naming)
             for table_schema in (root_table_schema, *nested_tables_schema)

dlt/extract/source.py

@@ -320,16 +320,16 @@ def schema(self) -> Schema:
     def schema(self, value: Schema) -> None:
         self._schema = value
 
-    def discover_schema(self, item: TDataItem = None) -> Schema:
+    def discover_schema(self, item: TDataItem = None, meta: Any = None) -> Schema:
         """Computes table schemas for all selected resources in the source and merges them with a copy of current source schema. If `item` is provided,
         dynamic tables will be evaluated, otherwise those tables will be ignored."""
         schema = self._schema.clone(update_normalizers=True)
         for r in self.selected_resources.values():
             # names must be normalized here
             with contextlib.suppress(DataItemRequiredForDynamicTableHints):
-                root_table_schema = r.compute_table_schema(item)
+                root_table_schema = r.compute_table_schema(item, meta)
                 nested_tables_schema = r.compute_nested_table_schemas(
-                    root_table_schema["name"], schema.naming, item
+                    root_table_schema["name"], schema.naming, item, meta
                 )
                 # NOTE must ensure that `schema.update_table()` is called in an order that respect parent-child relationships
                 for table_schema in (root_table_schema, *nested_tables_schema):

dlt/normalize/items_normalizers.py

@@ -133,7 +133,9 @@ def _normalize_chunk(
                         schema_contract = self._table_contracts.setdefault(
                             table_name,
                             schema.resolve_contract_settings_for_table(
-                                table_name if table_name in schema.tables else parent_table
+                                table_name
+                                if table_name in schema.tables
+                                else parent_table or table_name
                             ),  # parent_table, if present, exists in the schema
                         )
                         partial_table, filters = schema.apply_schema_contract(

tests/extract/test_decorators.py

@@ -17,13 +17,14 @@
 from dlt.common.pipeline import StateInjectableContext, TPipelineState
 from dlt.common.schema import Schema
 from dlt.common.schema.utils import new_table, new_column
-from dlt.common.schema.typing import TTableSchemaColumns
+from dlt.common.schema.typing import TTableReference, TTableSchemaColumns
 from dlt.common.schema.exceptions import InvalidSchemaName
-from dlt.common.typing import TDataItem
+from dlt.common.typing import TDataItem, TTableNames
 
 from dlt.cli.source_detection import detect_source_configs
 from dlt.common.utils import custom_environ
 from dlt.extract.decorators import _DltSingleSource, DltSourceFactoryWrapper
+from dlt.extract.hints import TResourceNestedHints
 from dlt.extract.reference import SourceReference
 from dlt.extract import DltResource, DltSource
 from dlt.extract.exceptions import (
@@ -320,7 +321,7 @@ def get_users():
 
 
 def test_apply_hints_reference() -> None:
-    example_reference = {
+    example_reference: TTableReference = {
         "columns": ["User ID", "user_name"],
         "referenced_table": "users",
         "referenced_columns": ["id", "name"],
@@ -342,7 +343,7 @@ def campaign_details(campaign):
 
 
 def test_nested_hints_decorator() -> None:
-    nested_hints = {
+    nested_hints: Dict[TTableNames, TResourceNestedHints] = {
         "steps": dlt.mark.make_nested_hints(
             columns=[{"name": "timestamp", "data_type": "timestamp"}]
         ),

tests/extract/test_extract.py

@@ -5,6 +5,7 @@
 import dlt
 from dlt.common import json
 from dlt.common.data_types.typing import TDataType
+from dlt.common.schema.utils import is_nested_table, may_be_nested
 from dlt.common.storages import (
     SchemaStorage,
     SchemaStorageConfiguration,
@@ -23,6 +24,19 @@
 from tests.extract.utils import expect_extracted_file
 
 
+NESTED_DATA = [
+    {
+        "id": 1,
+        "outer1": [
+            {"outer1_id": "2", "innerfoo": [{"innerfoo_id": "3"}]},
+        ],
+        "outer2": [
+            {"outer2_id": "4", "innerbar": [{"innerbar_id": "5"}]},
+        ],
+    }
+]
+
+
 @pytest.fixture
 def extract_step() -> Extract:
     clean_test_storage(init_normalize=True)
@@ -279,19 +293,8 @@ def with_table_hints():
 
 
 def test_extract_nested_hints(extract_step: Extract) -> None:
-    data = [
-        {
-            "id": 1,
-            "outer1": [
-                {"outer1_id": "2", "innerfoo": [{"innerfoo_id": "3"}]},
-            ],
-            "outer2": [
-                {"outer2_id": "4", "innerbar": [{"innerbar_id": "5"}]},
-            ],
-        }
-    ]
     resource_name = "with_nested_hints"
-    nested_resource = DltResource.from_data(data, name=resource_name)
+    nested_resource = DltResource.from_data(NESTED_DATA, name=resource_name)
 
     # Check 1: apply nested hints
     outer1_id_new_type: TDataType = "double"
@@ -318,21 +321,21 @@ def test_extract_nested_hints(extract_step: Extract) -> None:
 
     # root table exists even though there are no explicit hints
     assert pre_extract_schema.get_table(resource_name)
-    assert (
-        pre_extract_schema.get_table("with_nested_hints__outer1")["parent"] == "with_nested_hints"
-    )
-    assert (
-        pre_extract_schema.get_table("with_nested_hints__outer1")["columns"]
-        == nested_hints["outer1"]["columns"]
-    )
-    assert (
-        pre_extract_schema.get_table("with_nested_hints__outer2__innerbar")["parent"]
-        == "with_nested_hints__outer2"
-    )
-    assert (
-        pre_extract_schema.get_table("with_nested_hints__outer2__innerbar")["columns"]
-        == nested_hints[("outer2", "innerbar")]["columns"]
-    )
+    outer1_tab = pre_extract_schema.get_table("with_nested_hints__outer1")
+    assert outer1_tab["parent"] == "with_nested_hints"
+    assert outer1_tab["columns"] == nested_hints["outer1"]["columns"]
+    # no resource on nested table
+    assert "resource" not in outer1_tab
+    assert is_nested_table(outer1_tab) is True
+    assert may_be_nested(outer1_tab) is True
+
+    outer2_innerbar_tab = pre_extract_schema.get_table("with_nested_hints__outer2__innerbar")
+    assert outer2_innerbar_tab["parent"] == "with_nested_hints__outer2"
+    assert outer2_innerbar_tab["columns"] == nested_hints[("outer2", "innerbar")]["columns"]
+    assert "resource" not in outer2_innerbar_tab
+    assert is_nested_table(outer2_innerbar_tab) is True
+    assert may_be_nested(outer2_innerbar_tab) is True
+
     # this table is generated to ensure `innerbar` has a parent that links it to the root table
     # NOTE: nested tables do not have parent set
     assert pre_extract_schema.get_table(implicit_parent) == {
@@ -341,19 +344,88 @@ def test_extract_nested_hints(extract_step: Extract) -> None:
         "columns": {},
     }
 
-    source = DltSource(dlt.Schema("hintable"), "module", [nested_resource])
     extract_step.extract(source, 20, 1)
     # schema after extractions must be same as discovered schema
     assert source.schema._schema_tables == pre_extract_schema._schema_tables
 
 
-def test_break_nesting_with_primary_key() -> None:
+def test_break_nesting_with_primary_key(extract_step: Extract) -> None:
+    resource_name = "with_nested_hints"
+    nested_resource = DltResource.from_data(NESTED_DATA, name=resource_name)
+    nested_hints: Dict[TTableNames, TResourceNestedHints] = {
+        "outer1": {"columns": {"outer1_id": {"name": "outer1_id", "data_type": "bigint"}}},
+        ("outer1", "innerbar"): {"primary_key": "innerfoo_id"},
+    }
+    nested_resource.apply_hints(nested_hints=nested_hints)
+    assert nested_resource.nested_hints == nested_hints
+
+    source = DltSource(dlt.Schema("hintable"), "module", [nested_resource])
+    pre_extract_schema = source.discover_schema()
     # primary key will break nesting
+    # print(pre_extract_schema.to_pretty_yaml())
+    innerfoo_tab = pre_extract_schema.tables["with_nested_hints__outer1__innerbar"]
+    assert innerfoo_tab["columns"]["innerfoo_id"]["primary_key"] is True
     # resource must be present
+    assert innerfoo_tab["resource"] == "with_nested_hints"
     # parent cannot be present
+    assert "parent" not in innerfoo_tab["columns"]["innerfoo_id"]
     # is_nested_table must be false
+    assert is_nested_table(innerfoo_tab) is False
+    assert may_be_nested(innerfoo_tab) is False
+    extract_step.extract(source, 20, 1)
+    # schema after extractions must be same as discovered schema
+    assert source.schema._schema_tables == pre_extract_schema._schema_tables
+
+
+def test_nested_hints_dynamic_table_names(extract_step: Extract) -> None:
+    data = [
+        {"Event": "issue", "DataBlob": [{"ID": 1, "Name": "first", "Date": "2024-01-01"}]},
+        {"Event": "purchase", "DataBlob": [{"PID": "20-1", "Name": "first", "Date": "2024-01-01"}]},
+    ]
+    events = DltResource.from_data(
+        data,
+        name="events",
+        hints=dlt.mark.make_hints(
+            table_name=lambda e: e["Event"],
+            nested_hints={
+                "DataBlob": dlt.mark.make_nested_hints(
+                    columns=[{"name": "Date", "data_type": "date"}]
+                )
+            },
+        ),
+    )
+
+    source = DltSource(dlt.Schema("hintable"), "module", [events])
+    extract_step.extract(source, 20, 1)
+    # make sure that tables exist and types are applies
+    assert "issue" in source.schema.tables
+    assert "purchase" in source.schema.tables
+    assert source.schema.tables["issue__data_blob"]["columns"]["date"]["data_type"] == "date"
+    assert source.schema.tables["purchase__data_blob"]["columns"]["date"]["data_type"] == "date"
+
+
+def test_nested_hints_table_name(extract_step: Extract) -> None:
+    data = [
+        {"Event": "issue", "DataBlob": [{"ID": 1, "Name": "first", "Date": "2024-01-01"}]},
+        {"Event": "purchase", "DataBlob": [{"PID": "20-1", "Name": "first", "Date": "2024-01-01"}]},
+    ]
+    events = DltResource.from_data(
+        data,
+        name="events",
+        hints=dlt.mark.make_hints(
+            table_name="events_table",
+            nested_hints={
+                "DataBlob": dlt.mark.make_nested_hints(
+                    columns=[{"name": "Date", "data_type": "date"}]
+                )
+            },
+        ),
+    )
 
-    pass
+    source = DltSource(dlt.Schema("hintable"), "module", [events])
+    extract_step.extract(source, 20, 1)
+    assert "events_table" in source.schema.tables
+    assert source.schema.tables["events_table__data_blob"]["columns"]["date"]["data_type"] == "date"
 
 
 def test_extract_metrics_on_exception_no_flush(extract_step: Extract) -> None:

tests/extract/test_sources.py

@@ -1848,20 +1848,66 @@ def test_apply_nested_hints():
     ]
 
     nested_resource.apply_hints(nested_hints=nested_hints)
-    print(nested_resource.nested_hints)
     assert nested_resource.nested_hints == nested_hints
 
     nested_schemas = nested_resource.compute_nested_table_schemas(
         resource_name, naming=SnakeCaseNamingConvention()
     )
     assert nested_schemas == expected_nested_schemas
 
-    # TODO: repeat apply_hints with different nested hints. we have a different code path
-
-    # TODO: also test nested hints that declare primary or merge keys. that should
-    # - generate 'resource' name
-    # - drop the `parent`
-    # effectively breaking the nesting chain`
+    # apply hints again
+    nested_hints = {
+        ("outer1",): {
+            "columns": {
+                "outer1_id": {"name": "outer1_id", "data_type": "decimal", "precision": 18}
+            },
+            "references": [
+                {
+                    "columns": ["outer1_id"],
+                    "referenced_columns": ["outer1_fk_1"],
+                    "referenced_table": "external_table",
+                }
+            ],
+            "file_format": "parquet",
+            "table_format": "delta",
+            "schema_contract": "discard_value",
+        },
+        ("outer2", "innerbar"): {"merge_key": "innerfoo_id"},
+    }
+    nested_resource.apply_hints(nested_hints=nested_hints)
+    nested_schemas = nested_resource.compute_nested_table_schemas(
+        resource_name, naming=SnakeCaseNamingConvention()
+    )
+    # currently new hints overwrite fully old hints. nothing gets merged
+    assert nested_schemas == [
+        {
+            "columns": {
+                "outer1_id": {"name": "outer1_id", "data_type": "decimal", "precision": 18}
+            },
+            "references": [
+                {
+                    "columns": ["outer1_id"],
+                    "referenced_columns": ["outer1_fk_1"],
+                    "referenced_table": "external_table",
+                }
+            ],
+            "file_format": "parquet",
+            "table_format": "delta",
+            "schema_contract": "discard_value",
+            "name": "with_nested_hints__outer1",
+            "parent": "with_nested_hints",
+        },
+        {
+            "columns": {
+                "innerfoo_id": {"name": "innerfoo_id", "nullable": False, "merge_key": True}
+            },
+            "name": "with_nested_hints__outer2__innerbar",
+            "resource": "with_nested_hints",
+        },
+    ]
+    # with_nested_hints__outer2__innerbar not a nested table
+    assert utils.is_nested_table(nested_schemas[1]) is False
+    assert utils.may_be_nested(nested_schemas[1]) is False
 
 
 def test_resource_no_template() -> None: