fix: add schema fingerprint to pointer cache key for in-place mutation safety

Include a schema fingerprint (shape + column names + dtypes for pandas,
field names + types for pyarrow/polars) in the prepared-statement cache
key for pointer-type parameters.  Without this, mutating a DataFrame
in place between execute() calls (e.g. df.drop(columns=[...],
inplace=True)) would reuse the cached prepared statement compiled for
the original schema, causing data corruption.

The cache key tuple expands from 4 to 5 elements:
  ("pointer", module, type_name, gen) -> ("pointer", module, type_name, gen, schema_fp)

The fingerprint is computed fresh on every call; two objects with the
same identity and schema share the cache entry, while schema mutations
produce a different key and trigger a fresh prepare.

Author: adsharma

src_py/connection.py

@@ -128,6 +128,31 @@ def _capi_param_signature(parameters: dict[str, Any]) -> tuple:
 _pybind_pointer_tracker: dict[int, tuple[int, weakref.ref]] = {}
 
 
+def _get_pointer_schema_fingerprint(value: Any) -> int:
+    """
+    Compute a hash of the schema for a pandas/pyarrow/polars object.
+
+    Two objects with the same shape, column names, and column dtypes
+    produce the same fingerprint.  If the object is mutated in place
+    (e.g. columns dropped, renamed, or dtypes changed) the fingerprint
+    changes, causing a cache miss so a fresh prepared statement is
+    compiled for the new schema.
+    """
+    module_name = type(value).__module__
+    if module_name == "pandas.core.frame":
+        cols = list(value.columns)
+        return hash(
+            (value.shape, tuple(cols), tuple(str(value[col].dtype) for col in cols))
+        )
+    if module_name.startswith("pyarrow"):
+        schema = value.schema
+        return hash(tuple((f.name, str(f.type)) for f in schema))
+    if module_name.startswith("polars"):
+        schema = value.schema
+        return hash(tuple(sorted((name, str(dtype)) for name, dtype in schema.items())))
+    return 0
+
+
 def _pybind_int_signature(value: int) -> tuple[str]:
     if -(2**7) <= value <= 2**7 - 1:
         return ("int8",)
@@ -188,19 +213,20 @@ def _pybind_value_signature(value: Any) -> tuple:
     if module_name.startswith(("pandas", "polars", "pyarrow")):
         global _pybind_pointer_gen, _pybind_pointer_tracker
         obj_id = id(value)
+        fp = _get_pointer_schema_fingerprint(value)
         entry = _pybind_pointer_tracker.get(obj_id)
         if entry is not None:
             gen, weak = entry
             if weak() is not None:
-                return ("pointer", module_name, type(value).__name__, gen)
+                return ("pointer", module_name, type(value).__name__, gen, fp)
             # Original object was GC'd and a new object now occupies the
             # same memory address.  Purge the stale entry and fall
             # through to assign a fresh generation.
             del _pybind_pointer_tracker[obj_id]
         _pybind_pointer_gen += 1
         gen = _pybind_pointer_gen
         _pybind_pointer_tracker[obj_id] = (gen, weakref.ref(value))
-        return ("pointer", module_name, type(value).__name__, gen)
+        return ("pointer", module_name, type(value).__name__, gen, fp)
     if isinstance(value, dict):
         items = list(value.items())
         if (