fix: address Oracle review findings for UPSERT compiler

- Fix column qualification: only qualify excluded.* in SET clause, not
  target-table columns (visit_column checks is_upsert_set + table name)
- Remove _has_join toggle from visit_on_conflict_do_update
- Reject unsupported where/index_where params with ValueError
- Require index_elements (excel-dbapi parser needs ON CONFLICT (...))
- Fix ColumnCollection set_ to map column→column (not column→string)
- Remove dead code: index_where/where compilation paths
- Add 6 regression tests for all edge cases

Author: yeongseon

src/sqlalchemy_excel/compiler.py

@@ -280,15 +280,15 @@ def visit_column(
         ambiguous_table_name_map: MutableMapping[str, str] | None = None,
         **kwargs: Any,
     ) -> str:
-        """Override to conditionally include table prefix in column references.
-
-        For single-table queries: SELECT id, name FROM users
-        For JOIN queries: SELECT users.id, orders.amount FROM users JOIN orders ...
-        """
+        if kwargs.get("is_upsert_set"):
+            table_name = getattr(getattr(column, "table", None), "name", None)
+            use_table = table_name == "excluded"
+        else:
+            use_table = self._has_join
         return super().visit_column(
             column,
             add_to_result_map=add_to_result_map,
-            include_table=self._has_join,
+            include_table=use_table,
             result_map_targets=result_map_targets,
             ambiguous_table_name_map=ambiguous_table_name_map,
             **kwargs,
@@ -910,17 +910,6 @@ def _on_conflict_target(self, clause: Any, **kw: Any) -> str:
                 )
                 for c in clause.inferred_target_elements
             )
-            if clause.inferred_target_whereclause is not None:
-                whereclause_kw = dict(kw)
-                whereclause_kw.update(
-                    include_table=False,
-                    use_schema=False,
-                    literal_execute=True,
-                )
-                target_text += " WHERE %s" % self.process(
-                    clause.inferred_target_whereclause,
-                    **whereclause_kw,
-                )
         else:
             target_text = ""
 
@@ -943,81 +932,69 @@ def visit_on_conflict_do_update(self, on_conflict: Any, **kw: Any) -> str:
 
         set_parameters = dict(clause.update_values_to_set)
 
-        original_has_join = self._has_join
-        self._has_join = True
-        try:
-            insert_statement = cast(Any, self.stack[-1]["selectable"])
-            cols = insert_statement.table.c
-            set_kw = dict(kw)
-            set_kw.update(use_schema=False)
-            for c in cols:
-                col_key = c.key
-
-                if col_key in set_parameters:
-                    value = set_parameters.pop(col_key)
-                elif c in set_parameters:
-                    value = set_parameters.pop(c)
-                else:
-                    continue
+        insert_statement = cast(Any, self.stack[-1]["selectable"])
+        cols = insert_statement.table.c
+        set_kw = dict(kw)
+        set_kw.update(include_table=False, use_schema=False)
+        for c in cols:
+            col_key = c.key
+
+            if col_key in set_parameters:
+                value = set_parameters.pop(col_key)
+            elif c in set_parameters:
+                value = set_parameters.pop(c)
+            else:
+                continue
 
-                if coercions._is_literal(value):
-                    value = elements.BindParameter(None, value, type_=c.type)
+            if coercions._is_literal(value):
+                value = elements.BindParameter(None, value, type_=c.type)
 
-                else:
-                    if isinstance(value, elements.BindParameter) and value.type._isnull:
-                        value = value._clone()
-                        value.type = c.type
-                value_text = self.process(
-                    value.self_group(),
-                    is_upsert_set=True,
-                    **set_kw,
-                )
+            else:
+                if isinstance(value, elements.BindParameter) and value.type._isnull:
+                    value = value._clone()
+                    value.type = c.type
+            value_text = self.process(
+                value.self_group(),
+                is_upsert_set=True,
+                **set_kw,
+            )
 
-                key_text = self.preparer.quote(c.name)
-                action_set_ops.append("%s = %s" % (key_text, value_text))
+            key_text = self.preparer.quote(c.name)
+            action_set_ops.append("%s = %s" % (key_text, value_text))
 
-            if set_parameters:
-                from sqlalchemy import util as sa_util
+        if set_parameters:
+            from sqlalchemy import util as sa_util
 
-                table_name = getattr(
-                    getattr(self.current_executable, "table", None),
-                    "name",
-                    "<unknown>",
-                )
+            table_name = getattr(
+                getattr(self.current_executable, "table", None),
+                "name",
+                "<unknown>",
+            )
 
-                sa_util.warn(
-                    "Additional column names not matching "
-                    "any column keys in table '%s': %s"
-                    % (
-                        table_name,
-                        (", ".join("'%s'" % c for c in set_parameters)),
-                    )
+            sa_util.warn(
+                "Additional column names not matching "
+                "any column keys in table '%s': %s"
+                % (
+                    table_name,
+                    (", ".join("'%s'" % c for c in set_parameters)),
                 )
-                for k, v in set_parameters.items():
-                    key_text = (
-                        self.preparer.quote(k)
-                        if isinstance(k, str)
-                        else self.process(k, **set_kw)
-                    )
-                    value_text = self.process(
-                        coercions.expect(roles.ExpressionElementRole, v),
-                        is_upsert_set=True,
-                        **set_kw,
-                    )
-                    action_set_ops.append("%s = %s" % (key_text, value_text))
-
-            action_text = ", ".join(action_set_ops)
-            if clause.update_whereclause is not None:
-                where_kw = dict(kw)
-                where_kw.update(include_table=True, use_schema=False)
-                action_text += " WHERE %s" % self.process(
-                    clause.update_whereclause,
-                    **where_kw,
+            )
+            for k, v in set_parameters.items():
+                key_text = (
+                    self.preparer.quote(k)
+                    if isinstance(k, str)
+                    else self.process(k, **set_kw)
+                )
+                value_text = self.process(
+                    coercions.expect(roles.ExpressionElementRole, v),
+                    is_upsert_set=True,
+                    **set_kw,
                 )
+                action_set_ops.append("%s = %s" % (key_text, value_text))
 
-            return "ON CONFLICT %s DO UPDATE SET %s" % (target_text, action_text)
-        finally:
-            self._has_join = original_has_join
+        action_text = ", ".join(action_set_ops)
+
+        return "ON CONFLICT %s DO UPDATE SET %s" % (target_text, action_text)
 
     def visit_over(
         self,

src/sqlalchemy_excel/dml.py

@@ -69,6 +69,16 @@ def on_conflict_do_update(
         where: _OnConflictWhereT = None,
     ) -> Self:
         """Specifies a DO UPDATE SET action for ON CONFLICT clause."""
+        if not index_elements:
+            raise ValueError(
+                "Excel dialect requires index_elements for ON CONFLICT clause"
+            )
+        if index_where is not None:
+            raise ValueError("Excel dialect does not support index_where in ON CONFLICT")
+        if where is not None:
+            raise ValueError(
+                "Excel dialect does not support WHERE clause in ON CONFLICT DO UPDATE"
+            )
         self._post_values_clause = OnConflictDoUpdate(
             index_elements, index_where, set_, where
         )
@@ -82,6 +92,12 @@ def on_conflict_do_nothing(
         index_where: _OnConflictIndexWhereT = None,
     ) -> Self:
         """Specifies a DO NOTHING action for ON CONFLICT clause."""
+        if not index_elements:
+            raise ValueError(
+                "Excel dialect requires index_elements for ON CONFLICT clause"
+            )
+        if index_where is not None:
+            raise ValueError("Excel dialect does not support index_where in ON CONFLICT")
         self._post_values_clause = OnConflictDoNothing(index_elements, index_where)
         return self
 
@@ -135,7 +151,7 @@ def __init__(
                 raise ValueError("set parameter dictionary must not be empty")
             set_dict = set_
         elif isinstance(set_, ColumnCollection):
-            set_dict = cast(_OnConflictSetDictT, dict(set_))
+            set_dict = {column: column for column in set_}
             if not set_dict:
                 raise ValueError("set parameter dictionary must not be empty")
         else:

tests/test_upsert.py

@@ -124,6 +124,35 @@ def test_excluded_reference(self, populated_engine, items_table):
             rows = conn.execute(select(items_table).order_by(items_table.c.id)).all()
             assert rows == [(1, "Renamed", 30, "A")]
 
+    def test_expression_set_value(self, populated_engine, items_table):
+        with populated_engine.begin() as conn:
+            conn.execute(insert(items_table).values(id=1, name="Alice", age=30, code="A"))
+
+        stmt = insert(items_table).values(id=1, name="New", age=99, code="B")
+        upsert_stmt = stmt.on_conflict_do_update(
+            index_elements=["id"],
+            set_={"age": items_table.c.age + 1},
+        )
+        compiled = " ".join(str(upsert_stmt.compile(dialect=populated_engine.dialect)).split())
+        assert "items.age" not in compiled
+        assert " SET age = (age + " in compiled
+
+        with populated_engine.begin() as conn:
+            conn.execute(upsert_stmt)
+
+        with populated_engine.connect() as conn:
+            rows = conn.execute(select(items_table).order_by(items_table.c.id)).all()
+            assert rows == [(1, "Alice", 31, "A")]
+
+    def test_set_with_table_columns(self, populated_engine, items_table):
+        upsert_stmt = insert(items_table).on_conflict_do_update(
+            index_elements=["id"],
+            set_=items_table.c,
+        )
+        compiled = " ".join(str(upsert_stmt.compile(dialect=populated_engine.dialect)).split())
+        assert "items." not in compiled
+        assert " SET id = id, name = name, age = age, code = code" in compiled
+
     def test_multi_column_conflict_target(self, populated_engine, composite_items_table):
         with populated_engine.begin() as conn:
             conn.execute(
@@ -192,3 +221,38 @@ def test_double_on_conflict_raises(items_table):
 
     with pytest.raises(InvalidRequestError, match="already has an ON CONFLICT clause"):
         stmt.on_conflict_do_update(index_elements=["id"], set_={"age": 40})
+
+
+def test_where_rejected(items_table):
+    stmt = insert(items_table).values(id=1, name="Alice", age=30, code="A")
+
+    with pytest.raises(ValueError, match="does not support WHERE clause"):
+        stmt.on_conflict_do_update(
+            index_elements=["id"],
+            set_={"age": 1},
+            where=items_table.c.age > 10,
+        )
+
+
+def test_index_where_rejected(items_table):
+    stmt = insert(items_table).values(id=1, name="Alice", age=30, code="A")
+
+    with pytest.raises(ValueError, match="does not support index_where"):
+        stmt.on_conflict_do_nothing(
+            index_elements=["id"],
+            index_where=items_table.c.id > 0,
+        )
+
+
+def test_missing_index_elements_do_nothing(items_table):
+    stmt = insert(items_table).values(id=1, name="Alice", age=30, code="A")
+
+    with pytest.raises(ValueError, match="requires index_elements"):
+        stmt.on_conflict_do_nothing()
+
+
+def test_missing_index_elements_do_update(items_table):
+    stmt = insert(items_table).values(id=1, name="Alice", age=30, code="A")
+
+    with pytest.raises(ValueError, match="requires index_elements"):
+        stmt.on_conflict_do_update(set_={"age": 1})