From b8b5ecd4c98cd83ea05c12fe84a3fbdbf1f43954 Mon Sep 17 00:00:00 2001
From: Joscha Feth <joscha@feth.com>
Date: Thu, 28 Nov 2024 13:45:36 +0000
Subject: [PATCH] test: add test with hint for parent table

---
 dlt/common/schema/utils.py      |  2 +-
 tests/pipeline/test_pipeline.py | 22 ++++++++++++++++++++++
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/dlt/common/schema/utils.py b/dlt/common/schema/utils.py
index e2e1f959dc..f116291e1b 100644
--- a/dlt/common/schema/utils.py
+++ b/dlt/common/schema/utils.py
@@ -511,7 +511,7 @@ def diff_table(
             partial_table[k] = v  # type: ignore
 
     # this should not really happen
-    if is_nested_table(tab_a) and (resource := tab_b.get("resource")):
+    if is_nested_table(tab_a) and (resource := tab_b.get("resource")) and resource != tab_a.get("parent"):
         raise TablePropertiesConflictException(
             schema_name, table_name, "resource", resource, tab_a.get("parent")
         )
diff --git a/tests/pipeline/test_pipeline.py b/tests/pipeline/test_pipeline.py
index 38ba3713bb..8cac5fafde 100644
--- a/tests/pipeline/test_pipeline.py
+++ b/tests/pipeline/test_pipeline.py
@@ -661,6 +661,28 @@ def with_table_hints():
     # check table counts
     assert_data_table_counts(pipeline, {"table_a": 2, "table_b": 2, "table_c": 1})
 
+def test_mark_parent_table() -> None:
+    @dlt.resource()
+    def my_table():
+        yield {"id": 1}
+        yield dlt.mark.with_hints(
+            {"id": 10},
+            dlt.mark.make_hints(
+                table_name="my_other_table",
+                parent_table_name="my_table"
+            ),
+        )
+    pipeline_name = "pipe_" + uniq_id()
+    pipeline = dlt.pipeline(pipeline_name=pipeline_name, destination="duckdb")
+    info = pipeline.run(my_table)
+    assert_load_info(info)
+    assert pipeline.last_trace.last_normalize_info.row_counts == {
+        "_dlt_pipeline_state": 1,
+        "my_table": 1,
+        "my_other_table": 1,
+    }
+    # check table counts
+    assert_data_table_counts(pipeline, {"my_table": 1, "my_other_table": 1 })
 
 def test_restore_state_on_dummy() -> None:
     pipeline_name = "pipe_" + uniq_id()