fix: close oracle round 17 gaps

- Strip quoted literals in DDL metadata sync before constraint keyword
  detection to prevent false PK/NOT NULL reflection
- Reject unsupported subquery shapes at compile time: multi-column,
  compound (UNION/INTERSECT/EXCEPT), and ORDER BY/LIMIT/OFFSET
- Harden _metadata_header_matches with ordered comparison and document
  stale metadata limitation

Author: yeongseon

src/sqlalchemy_excel/compiler.py

@@ -49,6 +49,7 @@
     visitors,
 )
 from sqlalchemy.sql.expression import Join
+from sqlalchemy.sql.selectable import CompoundSelect, Select
 
 if TYPE_CHECKING:
     from collections.abc import Callable, MutableMapping
@@ -288,6 +289,28 @@ def _reject_correlated_subquery(self, inner: Any) -> None:
                     "Excel dialect does not support correlated subqueries"
                 )
 
+    @staticmethod
+    def _guard_subquery_shape(inner: Select[Any] | CompoundSelect[Any]) -> None:
+        if isinstance(inner, CompoundSelect):
+            raise exc.CompileError(
+                "Excel dialect does not support compound subqueries in WHERE ... IN"
+            )
+
+        if len(inner.selected_columns) != 1:
+            raise exc.CompileError(
+                "Excel dialect only supports single-column subqueries in WHERE ... IN"
+            )
+
+        if inner._order_by_clauses:
+            raise exc.CompileError(
+                "Excel dialect does not support ORDER BY in subqueries"
+            )
+
+        if inner._limit_clause is not None or inner._offset_clause is not None:
+            raise exc.CompileError(
+                "Excel dialect does not support LIMIT/OFFSET in subqueries"
+            )
+
     def _is_unresolved_bindparam(self, bindparam: elements.BindParameter[Any]) -> bool:
         return bool(getattr(bindparam, "required", False)) and (
             getattr(bindparam, "callable", None) is None
@@ -580,13 +603,18 @@ def visit_subquery(self, subquery: Any, **kw: Any) -> str:
 
         inner = getattr(subquery, "element", None)
         if inner is not None:
-            # Reject subqueries that themselves contain a JOIN
-            for from_clause in inner.get_final_froms():
-                if isinstance(from_clause, Join):
-                    raise exc.CompileError(
-                        "Excel dialect does not support JOIN inside subqueries"
-                    )
-            self._reject_correlated_subquery(inner)
+            if isinstance(inner, CompoundSelect):
+                self._guard_subquery_shape(inner)
+            elif getattr(inner, "__visit_name__", None) == "select":
+                select_inner = cast("Select[Any]", inner)
+                self._guard_subquery_shape(select_inner)
+                # Reject subqueries that themselves contain a JOIN
+                for from_clause in select_inner.get_final_froms():
+                    if isinstance(from_clause, Join):
+                        raise exc.CompileError(
+                            "Excel dialect does not support JOIN inside subqueries"
+                        )
+                self._reject_correlated_subquery(select_inner)
 
         self._subquery_depth += 1
         try:
@@ -598,7 +626,8 @@ def visit_subquery(self, subquery: Any, **kw: Any) -> str:
 
     def visit_grouping(self, grouping: Any, asfrom: bool = False, **kwargs: Any) -> str:
         element = getattr(grouping, "element", None)
-        is_subquery_select = getattr(element, "__visit_name__", None) == "select"
+        visit_name = getattr(element, "__visit_name__", None)
+        is_subquery_select = visit_name in {"select", "compound_select"}
         if is_subquery_select:
             if not self._in_in_clause:
                 raise exc.CompileError(
@@ -616,13 +645,18 @@ def visit_grouping(self, grouping: Any, asfrom: bool = False, **kwargs: Any) ->
                 )
 
             assert element is not None  # guaranteed by is_subquery_select check
-            # Reject subqueries that themselves contain a JOIN
-            for from_clause in element.get_final_froms():
-                if isinstance(from_clause, Join):
-                    raise exc.CompileError(
-                        "Excel dialect does not support JOIN inside subqueries"
-                    )
-            self._reject_correlated_subquery(element)
+            if visit_name == "compound_select":
+                self._guard_subquery_shape(cast("CompoundSelect[Any]", element))
+            elif visit_name == "select":
+                select_element = cast("Select[Any]", element)
+                self._guard_subquery_shape(select_element)
+                # Reject subqueries that themselves contain a JOIN
+                for from_clause in select_element.get_final_froms():
+                    if isinstance(from_clause, Join):
+                        raise exc.CompileError(
+                            "Excel dialect does not support JOIN inside subqueries"
+                        )
+                self._reject_correlated_subquery(select_element)
             kwargs["literal_binds"] = True
 
         if is_subquery_select:

src/sqlalchemy_excel/dialect.py

@@ -584,7 +584,7 @@ def _sync_create_table_metadata(self, cursor: Any, statement: str) -> None:
 
             col_name = self._unquote_identifier(col_match.group(1))
             remainder = col_match.group(2)
-            remainder_upper = remainder.upper()
+            remainder_upper = _strip_quoted_literals(remainder).upper()
             extracted = self._extract_declared_type_name(remainder)
             if extracted is None:
                 continue
@@ -672,7 +672,7 @@ def _sync_alter_table_metadata(
             type_map[col_name] = added_type
             # Preserve nullable/PK hints from trailing constraints:
             # ALTER TABLE t ADD COLUMN c TYPE NOT NULL PRIMARY KEY
-            tail = remainder.upper()
+            tail = _strip_quoted_literals(remainder).upper()
             is_pk = "PRIMARY KEY" in tail or "PRIMARY_KEY" in tail
             nullable_map[col_name] = "NOT NULL" not in tail and not is_pk
             pk_map[col_name] = is_pk

src/sqlalchemy_excel/reflection.py

@@ -232,10 +232,9 @@ def _metadata_header_matches(
     ) -> bool:
         if header_names is None:
             return False
-        # Check that metadata column names match live headers
-        # AND metadata column count matches live column count.
-        # This guards against the edge case where a sheet is deleted and
-        # recreated with the same header names but different types/nullability/PKs.
+        # Header validation is intentionally strict on column order and count.
+        # Limitation: if a worksheet is recreated with the exact same headers,
+        # this check alone cannot detect drift in type/nullability/PK metadata.
         return [col["name"] for col in meta] == header_names and len(meta) == len(
             header_names
         )

tests/test_compiler_guards.py

@@ -410,6 +410,60 @@ def test_compiler_allows_subquery_with_inline_literal_predicate(tmp_xlsx: str) -
     engine.dispose()
 
 
+def test_compiler_rejects_multi_column_in_subquery(tmp_xlsx: str) -> None:
+    engine = create_engine(f"excel:///{tmp_xlsx}")
+    metadata = MetaData()
+    users, orders = _build_tables(metadata)
+
+    sub = select(orders.c.user_id, orders.c.id)
+    stmt = select(users).where(users.c.id.in_(sub))
+    with pytest.raises(exc.CompileError, match="single-column subqueries"):
+        stmt.compile(dialect=engine.dialect)
+
+    engine.dispose()
+
+
+def test_compiler_rejects_compound_in_subquery(tmp_xlsx: str) -> None:
+    engine = create_engine(f"excel:///{tmp_xlsx}")
+    metadata = MetaData()
+    users, orders = _build_tables(metadata)
+
+    sub = sa.union(select(orders.c.user_id), select(orders.c.id))
+    stmt = select(users).where(users.c.id.in_(sub))
+    with pytest.raises(exc.CompileError, match="compound subqueries"):
+        stmt.compile(dialect=engine.dialect)
+
+    engine.dispose()
+
+
+def test_compiler_rejects_ordered_limited_in_subquery(tmp_xlsx: str) -> None:
+    engine = create_engine(f"excel:///{tmp_xlsx}")
+    metadata = MetaData()
+    users, orders = _build_tables(metadata)
+
+    sub = select(orders.c.user_id).order_by(orders.c.user_id).limit(1)
+    stmt = select(users).where(users.c.id.in_(sub))
+    with pytest.raises(exc.CompileError, match="ORDER BY"):
+        stmt.compile(dialect=engine.dialect)
+
+    engine.dispose()
+
+
+def test_compiler_allows_simple_single_column_subquery(tmp_xlsx: str) -> None:
+    engine = create_engine(f"excel:///{tmp_xlsx}")
+    metadata = MetaData()
+    users, orders = _build_tables(metadata)
+
+    sub = select(orders.c.user_id).where(orders.c.user_id > 0)
+    stmt = select(users).where(users.c.id.in_(sub))
+    compiled = stmt.compile(dialect=engine.dialect)
+    sql = str(compiled)
+    assert "IN" in sql
+    assert "SELECT" in sql
+
+    engine.dispose()
+
+
 def test_compiler_rejects_unresolved_bindparam_in_subquery(tmp_xlsx: str) -> None:
     engine = create_engine(f"excel:///{tmp_xlsx}")
     metadata = MetaData()

tests/test_ddl.py

@@ -187,3 +187,16 @@ def test_drop_table_removes_worksheet(self, engine, metadata, users_table):
         insp = inspect(engine)
         tables = insp.get_table_names()
         assert "users" not in tables
+
+    def test_create_table_default_primary_key_literal_does_not_mark_pk(
+        self, engine
+    ) -> None:
+        with engine.begin() as conn:
+            conn.exec_driver_sql(
+                "CREATE TABLE users (id INTEGER, note TEXT DEFAULT 'PRIMARY KEY')"
+            )
+
+        inspector = inspect(engine)
+        columns = {column["name"]: column for column in inspector.get_columns("users")}
+        assert columns["note"]["nullable"] is True
+        assert inspector.get_pk_constraint("users")["constrained_columns"] == []

tests/test_reflection_full.py

@@ -161,6 +161,32 @@ def test_reflection_validates_metadata_with_cursor_when_headers_unavailable(
     engine.dispose()
 
 
+def test_reflection_same_headers_cannot_detect_metadata_drift(tmp_path) -> None:
+    workbook_path = tmp_path / "reflection.xlsx"
+    engine = create_engine(f"excel:///{workbook_path}")
+
+    with engine.begin() as conn:
+        conn.execute(text("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)"))
+
+    workbook = load_workbook(workbook_path)
+    stale_sheet = workbook["users"]
+    workbook.remove(stale_sheet)
+    recreated = workbook.create_sheet("users")
+    recreated.cell(row=1, column=1, value="id")
+    recreated.cell(row=1, column=2, value="name")
+    workbook.save(workbook_path)
+
+    engine.dispose()
+    engine = create_engine(f"excel:///{workbook_path}")
+
+    inspector = inspect(engine)
+    columns = inspector.get_columns("users")
+    assert [column["name"] for column in columns] == ["id", "name"]
+    assert inspector.get_pk_constraint("users")["constrained_columns"] == ["id"]
+
+    engine.dispose()
+
+
 def test_table_scoped_reflection_methods_raise_for_missing_table(tmp_path) -> None:
     engine = _engine_for(tmp_path)
     inspector = inspect(engine)