Add T006 select-into-without-typed-fields (T-SQL runtime schema inference)

CHANGELOG.md

@@ -34,6 +34,18 @@ a deprecation window (see `GOVERNANCE.md` § Scope discipline).
   helper from `base.py`. `, LATERAL ...` is recognised as a
   legitimate Snowflake/Postgres lateral join and not flagged.
   Resolves #42.
+- **T006 `select-into-without-typed-fields`** (warning) - flags
+  T-SQL `SELECT * INTO target FROM source` because the destination
+  table's schema is derived from whatever the source produces at
+  execution time. If the source columns change shape (a column added,
+  type widened, an index changed) the destination silently adopts
+  those changes and any code reading from `target` finds the schema
+  has shifted underneath it. Recommended pattern: `CREATE TABLE
+  target (...)` with explicit typed columns, then `INSERT INTO target
+  (col1, ...) SELECT col1, ...`. Fires only on the wildcard form;
+  the explicit-column variant (`SELECT col1, col2 INTO target FROM
+  source`) is allowed through here and covered by the contracts pack
+  at C001/C003 if a column type drifts. Resolves #43.
 
 ### Fixed
 

README.md

@@ -265,6 +265,7 @@ positives on non-T-SQL input.
 | T003 | `cursor-declaration` | `DECLARE c CURSOR FOR ...` -- row-by-row processing |
 | T004 | `deprecated-outer-join` | `WHERE a.x *= b.y` -- removed in SQL Server 2012+ |
 | T005 | `create-index-without-online` | `CREATE INDEX ix ON t (...)` -- locks table; add `WITH (ONLINE = ON)` |
+| T006 | `select-into-without-typed-fields` | `SELECT * INTO target FROM source` -- destination schema is inferred at runtime |
 
 ### Contracts (opt-in via `--contract`)
 

sql_guard/rules/__init__.py

@@ -48,6 +48,7 @@
     CreateIndexWithoutOnline,
     CursorDeclaration,
     DeprecatedOuterJoin,
+    SelectStarInto,
     WithNolock,
     XpCmdshell,
 )
@@ -113,12 +114,13 @@
     ImplicitCrossJoin(),
     DeeplyNestedSubquery(),
     UnusedCTE(),
-    # T-SQL specific (T001-T005)
+    # T-SQL specific (T001-T006)
     WithNolock(),
     XpCmdshell(),
     CursorDeclaration(),
     DeprecatedOuterJoin(),
     CreateIndexWithoutOnline(),
+    SelectStarInto(),
 ]
 
 

sql_guard/rules/tsql.py

@@ -165,3 +165,55 @@ def check_statement(self, statement: str, start_line: int, file: str) -> Finding
                 suggestion="Add WITH (ONLINE = ON) on Enterprise; disable T005 on Standard/Express",
             )
         return None
+
+
+class SelectStarInto(Rule):
+    """T006: ``SELECT * INTO target`` infers column types at runtime.
+
+    T-SQL's ``SELECT * INTO new_table FROM source`` derives the schema of
+    ``new_table`` from whatever the source produces at execution time.
+    If the source columns change shape (a column added, type widened, an
+    index changed) the destination table silently adopts those changes,
+    and any code reading from ``new_table`` finds the schema has shifted
+    underneath it. The data-integrity hit is delayed and hard to trace.
+
+    Recommended pattern: ``CREATE TABLE new_table (...)`` with explicit
+    typed columns, then ``INSERT INTO new_table (col1, col2, ...) SELECT
+    ...``. The destination schema lives in source control and a contract
+    breakage shows up as a compile error rather than silently propagated
+    wrong types.
+
+    The rule fires only on the wildcard form. ``SELECT col1, col2 INTO
+    target FROM source`` still derives types from source columns but at
+    least names what is being copied; it stays a green path and gets
+    caught (if a column type drifts) by the contracts pack at C001/C003.
+
+    Suppress with an inline ``-- noqa: T006`` comment on the same line,
+    or use the project-wide ``-- sql-guard: disable=T006`` directive.
+    """
+
+    id = "T006"
+    name = "select-into-without-typed-fields"
+    severity = "warning"
+    description = "SELECT * INTO derives the destination schema from the source at runtime"
+    multiline = True
+
+    _pattern = Rule._compile(r"\bSELECT\s+\*\s+INTO\s+\S+")
+
+    def check_statement(self, statement: str, start_line: int, file: str) -> Finding | None:
+        if self._pattern.search(statement):
+            return Finding(
+                rule_id=self.id,
+                severity=self.severity,
+                file=file,
+                line=start_line,
+                message=(
+                    "SELECT * INTO derives the destination schema from the source "
+                    "at runtime -- silent breakage when the source changes shape"
+                ),
+                suggestion=(
+                    "CREATE TABLE target (col1 TYPE, ...) explicitly, then "
+                    "INSERT INTO target (col1, ...) SELECT col1, ... FROM source"
+                ),
+            )
+        return None

tests/test_rules.py

@@ -18,18 +18,19 @@
 
 class TestRuleRegistry:
     def test_all_rules_loaded(self) -> None:
-        assert len(ALL_RULES) == 40
+        assert len(ALL_RULES) == 41
 
     def test_11_errors(self) -> None:
         # 9 E-series + 2 T-series (T002 xp-cmdshell, T004 deprecated-outer-join).
         errors = [r for r in ALL_RULES if r.severity == "error"]
         assert len(errors) == 11
 
-    def test_29_warnings(self) -> None:
-        # 23 W-series + 3 S-series + 3 T-series (T001 with-nolock,
-        # T003 cursor-declaration, T005 create-index-without-online).
+    def test_30_warnings(self) -> None:
+        # 23 W-series + 3 S-series + 4 T-series (T001 with-nolock,
+        # T003 cursor-declaration, T005 create-index-without-online,
+        # T006 select-into-without-typed-fields).
         warnings = [r for r in ALL_RULES if r.severity == "warning"]
-        assert len(warnings) == 29
+        assert len(warnings) == 30
 
     def test_unique_ids(self) -> None:
         ids = [r.id for r in ALL_RULES]

tests/test_tsql.py

@@ -1,10 +1,11 @@
-"""Tests for T-SQL-specific rules (T001-T004)."""
+"""Tests for T-SQL-specific rules (T001-T006)."""
 
 from __future__ import annotations
 
 from sql_guard.rules.tsql import (
     CursorDeclaration,
     DeprecatedOuterJoin,
+    SelectStarInto,
     WithNolock,
     XpCmdshell,
 )
@@ -138,3 +139,68 @@ def test_w002_still_accepts_fetch_first():
     rule = MissingLimit()
     sql = "SELECT id FROM orders ORDER BY id FETCH FIRST 10 ROWS ONLY"
     assert _check_statement(rule, sql) is None
+
+
+# T006 select-into-without-typed-fields
+
+
+def test_t006_flags_basic_select_star_into():
+    rule = SelectStarInto()
+    finding = _check_statement(rule, "SELECT * INTO staging_orders FROM orders;")
+    assert finding is not None
+    assert finding.rule_id == "T006"
+    assert finding.severity == "warning"
+
+
+def test_t006_flags_select_star_into_with_where():
+    rule = SelectStarInto()
+    sql = "SELECT * INTO archive_2024 FROM orders WHERE year = 2024;"
+    assert _check_statement(rule, sql) is not None
+
+
+def test_t006_flags_multiline_select_star_into():
+    rule = SelectStarInto()
+    sql = "SELECT *\nINTO staging_orders\nFROM orders;"
+    assert _check_statement(rule, sql) is not None
+
+
+def test_t006_case_insensitive():
+    rule = SelectStarInto()
+    assert _check_statement(rule, "select * into staging from orders;") is not None
+
+
+def test_t006_does_not_flag_typed_columns():
+    # The recommended pass form from the issue.
+    rule = SelectStarInto()
+    sql = "SELECT order_id, customer_id INTO staging_orders FROM orders;"
+    assert _check_statement(rule, sql) is None
+
+
+def test_t006_does_not_flag_single_column_into():
+    rule = SelectStarInto()
+    assert _check_statement(rule, "SELECT id INTO ids FROM orders;") is None
+
+
+def test_t006_does_not_flag_select_star_without_into():
+    rule = SelectStarInto()
+    assert _check_statement(rule, "SELECT * FROM orders WHERE id = 1;") is None
+
+
+def test_t006_does_not_flag_tsql_variable_assignment():
+    # SELECT @x = COUNT(*) FROM ... is a T-SQL local-variable assignment,
+    # not a SELECT * INTO target. No schema is being inferred.
+    rule = SelectStarInto()
+    assert _check_statement(rule, "SELECT @x = COUNT(*) FROM orders;") is None
+
+
+def test_t006_does_not_flag_select_star_inside_cte():
+    rule = SelectStarInto()
+    sql = "WITH s AS (SELECT * FROM orders) SELECT id FROM s;"
+    assert _check_statement(rule, sql) is None
+
+
+def test_t006_message_mentions_runtime_schema():
+    rule = SelectStarInto()
+    finding = _check_statement(rule, "SELECT * INTO staging FROM orders;")
+    assert finding is not None
+    assert "runtime" in finding.message.lower() or "source" in finding.message.lower()