always drop sys id in merge (#1424)

Author: shcheklein

src/datachain/delta.py

@@ -150,7 +150,9 @@ def _get_retry_chain(
         error_records = result_dataset.filter(C(delta_retry) != "")
         error_source_records = source_dc.merge(
             error_records, on=on, right_on=right_on, inner=True
-        ).select(*list(source_dc.signals_schema.values))
+        ).select(
+            *list(source_dc.signals_schema.clone_without_sys_signals().values.keys())
+        )
         retry_chain = error_source_records
 
     # Handle missing records if delta_retry is True

src/datachain/lib/dc/datachain.py

@@ -1697,14 +1697,13 @@ def _resolve(
         query.feature_schema = None
         ds = self._evolve(query=query)
 
+        # Note: merge drops sys signals from both sides, make sure to not include it
+        # in the resulting schema
         signals_schema = self.signals_schema.clone_without_sys_signals()
         right_signals_schema = right_ds.signals_schema.clone_without_sys_signals()
 
         ds.signals_schema = signals_schema.merge(right_signals_schema, rname)
 
-        if not full:
-            ds.signals_schema = SignalSchema({"sys": Sys}) | ds.signals_schema
-
         return ds
 
     @delta_disabled

src/datachain/query/dataset.py

@@ -1065,7 +1065,7 @@ def apply(
         q1 = self.get_query(self.query1, temp_tables)
         q2 = self.get_query(self.query2, temp_tables)
 
-        q1_columns = _drop_system_columns(q1.c) if self.full else list(q1.c)
+        q1_columns = _drop_system_columns(q1.c)
         q1_column_names = {c.name for c in q1_columns}
 
         q2_columns = []

tests/unit/lib/test_datachain_merge.py

@@ -140,7 +140,7 @@ def test_merge_similar_objects(test_session):
     rname = "qq"
     ch = ch1.merge(ch2, "emp.person.name", rname=rname)
 
-    assert list(ch.signals_schema.values.keys()) == ["sys", "emp", rname + "emp"]
+    assert list(ch.signals_schema.values.keys()) == ["emp", rname + "emp"]
 
     empl = list(ch.to_list())
     assert len(empl) == 4
@@ -175,7 +175,7 @@ def test_merge_similar_objects_in_memory():
     assert ch.session.catalog.metastore.db.db_file == ":memory:"
     assert ch.session.catalog.warehouse.db.db_file == ":memory:"
 
-    assert list(ch.signals_schema.values.keys()) == ["sys", "emp", rname + "emp"]
+    assert list(ch.signals_schema.values.keys()) == ["emp", rname + "emp"]
 
     empl = list(ch.to_list())
     assert len(empl) == 4
@@ -198,7 +198,6 @@ def test_merge_values(test_session):
     ch = ch1.merge(ch2, "id")
 
     assert list(ch.signals_schema.values.keys()) == [
-        "sys",
         "id",
         "descr",
         "right_id",
@@ -339,3 +338,42 @@ def _get_expr(chain):
         count += 1
 
     assert count == len(team) * len(team)
+
+
+def test_merge_with_drops_sys_columns(test_session):
+    left = dc.read_values(id=[1, 1], lval=[10, 20], session=test_session)
+    right = dc.read_values(id=[1, 1], rval=["a", "b"], session=test_session)
+
+    merged = left.merge(right, on="id")
+
+    assert "sys" not in merged.signals_schema.values
+
+    cols = merged.settings(sys=True).to_pandas(flatten=True).columns
+    assert all(not str(col).startswith("sys") for col in cols)
+
+    ds_name = "merge_left_dups_sys_check_sys"
+    merged.save(ds_name)
+
+    df_with_sys = (
+        dc.read_dataset(ds_name, session=test_session)
+        .settings(sys=True)
+        .to_pandas(flatten=True)
+    )
+
+    sys_cols = [c for c in df_with_sys.columns if str(c).startswith("sys")]
+    assert sys_cols
+
+    def _col(name: str) -> str:
+        for col in df_with_sys.columns:
+            if str(col) == f"sys.{name}":
+                return str(col)
+        raise AssertionError(f"Missing sys column for {name}")
+
+    sys_id_col = _col("id")
+    sys_rand_col = _col("rand")
+
+    sys_ids = list(df_with_sys[sys_id_col])
+    assert len(sys_ids) == len(set(sys_ids))
+
+    sys_rand = list(df_with_sys[sys_rand_col])
+    assert len(sys_rand) == len(set(sys_rand))