feat(raw_cursor): 支持位置占位符和查询缓存机制

- 实现 GaussDBRawQuery 支持只使用位置占位符 ($1, $2, ...) ，禁止命名占位符
- 添加查询缓存功能，缓存已解析的查询字节串，提高执行效率
- 在执行时检查查询是否包含命名占位符，若有则抛出明确的 ProgrammingError
- 参数序列化时严格要求参数为序列类型，若传入字典则抛出 TypeError，提示使用普通 Cursor
- 提供 clear_cache 方法支持清理查询缓存
- 补充单元测试，验证命名参数使用异常抛出、查询缓存以及缓存清理功能正常工作

Author: 陈云亮

gaussdb/gaussdb/raw_cursor.py

@@ -8,6 +8,7 @@
 
 from typing import TYPE_CHECKING
 
+from . import errors as e
 from .abc import ConnectionType, Params, Query
 from .sql import Composable
 from .rows import Row
@@ -26,6 +27,19 @@
 
 
 class GaussDBRawQuery(GaussDBQuery):
+    """
+    GaussDB raw query class.
+
+    Only supports positional placeholders ($1, $2, ...), not named placeholders.
+    """
+
+    # Query cache size
+    _CACHE_SIZE = 128
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._query_cache: dict[bytes, bytes] = {}
+
     def convert(self, query: Query, vars: Params | None) -> None:
         if isinstance(query, str):
             bquery = query.encode(self._encoding)
@@ -34,14 +48,43 @@ def convert(self, query: Query, vars: Params | None) -> None:
         else:
             bquery = query
 
-        self.query = bquery
+        # Try to get from cache
+        if bquery in self._query_cache:
+            self.query = self._query_cache[bquery]
+        else:
+            # Validate query doesn't contain named placeholders
+            if b"%(" in bquery:
+                raise e.ProgrammingError(
+                    "RawCursor does not support named placeholders (%(name)s). "
+                    "Use positional placeholders ($1, $2, ...) instead."
+                )
+
+            self.query = bquery
+
+            # Cache result
+            if len(self._query_cache) < self._CACHE_SIZE:
+                self._query_cache[bquery] = bquery
+
         self._want_formats = self._order = None
         self.dump(vars)
 
     def dump(self, vars: Params | None) -> None:
+        """
+        Serialize parameters.
+
+        Args:
+            vars: Parameter sequence (must be sequence, not dict)
+
+        Raises:
+            TypeError: If parameters are not a sequence
+        """
         if vars is not None:
             if not GaussDBQuery.is_params_sequence(vars):
-                raise TypeError("raw queries require a sequence of parameters")
+                raise TypeError(
+                    "RawCursor requires a sequence of parameters (tuple or list), "
+                    f"got {type(vars).__name__}. "
+                    "For named parameters, use regular Cursor instead."
+                )
             self._want_formats = [PyFormat.AUTO] * len(vars)
 
             self.params = self._tx.dump_sequence(vars, self._want_formats)
@@ -52,6 +95,10 @@ def dump(self, vars: Params | None) -> None:
             self.types = ()
             self.formats = None
 
+    def clear_cache(self) -> None:
+        """Clear query cache."""
+        self._query_cache.clear()
+
 
 class RawCursorMixin(BaseCursor[ConnectionType, Row]):
     _query_cls = GaussDBRawQuery

tests/test_cursor_raw.py

@@ -36,6 +36,28 @@ def test_sequence_only(conn):
         cur.execute("select 1", {})
 
 
+def test_named_params_error(conn):
+    """Test named parameter error message."""
+    cur = conn.cursor()
+
+    # Should clearly indicate named placeholders are not supported
+    with pytest.raises((TypeError, e.ProgrammingError)) as excinfo:
+        cur.execute("select %(name)s", {"name": 1})
+
+    error_msg = str(excinfo.value).lower()
+    assert "named" in error_msg or "sequence" in error_msg
+
+
+def test_dict_params_error(conn):
+    """Test dict parameter error."""
+    cur = conn.cursor()
+
+    with pytest.raises(TypeError) as excinfo:
+        cur.execute("select $1", {"a": 1})
+
+    assert "sequence" in str(excinfo.value).lower()
+
+
 def test_execute_many_results_param(conn):
     cur = conn.cursor()
     # GaussDB raises SyntaxError, CRDB raises InvalidPreparedStatementDefinition
@@ -115,3 +137,31 @@ def work():
         gc.collect()
         n.append(gc.count())
     assert n[0] == n[1] == n[2], f"objects leaked: {n[1] - n[0]}, {n[2] - n[1]}"
+
+
+def test_query_cache(conn):
+    """Test query cache."""
+    cur = conn.cursor()
+    query = "select $1::int"
+
+    # Execute same query multiple times
+    for i in range(10):
+        cur.execute(query, (i,))
+        assert cur.fetchone()[0] == i
+
+    # Verify cache is working (no exception is good)
+
+
+def test_clear_cache(conn):
+    """Test clearing cache."""
+    cur = conn.cursor()
+    query = "select $1::int"
+
+    cur.execute(query, (1,))
+
+    # Clearing cache should not affect subsequent queries
+    if hasattr(cur._query, "clear_cache"):
+        cur._query.clear_cache()
+
+    cur.execute(query, (2,))
+    assert cur.fetchone()[0] == 2