Fix NameError: translator not defined and ensure consistent session handling

- Fix critical NameError in _execute_embedded_async.
- Ensure session_id is passed to all IRISExecutor calls in protocol handler.
- Refactor translator usage to be consistent across execution paths.
- Add configurable connection pooling and LRU query caching.
- Verify transaction visibility with regression tests.

Author: tom-dyar

src/iris_pgwire/__init__.py

@@ -6,7 +6,7 @@
 caretdev/sqlalchemy-iris.
 """
 
-__version__ = "1.2.8"
+__version__ = "1.2.9"
 __author__ = "Thomas Dyar <thomas.dyar@intersystems.com>"
 
 # Don't import server/protocol in __init__ to avoid sys.modules conflicts

src/iris_pgwire/iris_executor.py

@@ -796,9 +796,8 @@ def _sync_execute_many():
 
             try:
                 # Feature 022: Apply PostgreSQL→IRIS transaction verb translation
-                transaction_translator = TransactionTranslator()
-                transaction_translated_sql = transaction_translator.translate_transaction_command(
-                    sql
+                transaction_translated_sql = (
+                    self.transaction_translator.translate_transaction_command(sql)
                 )
 
                 # Feature 021: Apply PostgreSQL→IRIS SQL normalization
@@ -920,12 +919,11 @@ def _sync_execute_many():
 
             try:
                 # Get pooled connection
-                connection = self._get_pooled_connection()
+                connection = self._get_pooled_connection(session_id=session_id)
 
                 # Feature 022: Apply PostgreSQL→IRIS transaction verb translation
-                transaction_translator = TransactionTranslator()
-                transaction_translated_sql = transaction_translator.translate_transaction_command(
-                    sql
+                transaction_translated_sql = (
+                    self.transaction_translator.translate_transaction_command(sql)
                 )
 
                 # Feature 021: Apply PostgreSQL→IRIS SQL normalization
@@ -1080,7 +1078,13 @@ def _split_multi_row_insert(self, sql: str) -> list[str]:
 
         return statements
 
-    def _safe_execute(self, sql: str, params: list | None = None, is_embedded: bool = True) -> Any:
+    def _safe_execute(
+        self,
+        sql: str,
+        params: list | None = None,
+        is_embedded: bool = True,
+        session_id: str | None = None,
+    ) -> Any:
         """Execute SQL with DDL idempotency handling."""
         import iris
 
@@ -1127,7 +1131,7 @@ def __iter__(self):
 
             else:
                 # External mode - use DBAPI cursor
-                connection = self._get_pooled_connection()
+                connection = self._get_pooled_connection(session_id=session_id)
                 cursor = connection.cursor()
                 try:
                     if params:
@@ -2854,9 +2858,8 @@ def _sync_execute():
                 self._get_iris_connection()
 
                 # 1. Transaction Translation
-                transaction_translator = TransactionTranslator()
-                transaction_translated_sql = transaction_translator.translate_transaction_command(
-                    sql
+                transaction_translated_sql = (
+                    self.transaction_translator.translate_transaction_command(sql)
                 )
 
                 # 2. SQL Normalization
@@ -2866,7 +2869,7 @@ def _sync_execute():
                 optimized_sql = normalized_sql
 
                 # Log transaction translation metrics
-                txn_metrics = transaction_translator.get_translation_metrics()
+                txn_metrics = self.transaction_translator.get_translation_metrics()
                 logger.info(
                     "Transaction verb translation applied",
                     total_translations=txn_metrics["total_translations"],
@@ -2878,7 +2881,7 @@ def _sync_execute():
                 )
 
                 # Log normalization metrics
-                norm_metrics = translator.get_normalization_metrics()
+                norm_metrics = self.sql_translator.get_normalization_metrics()
                 logger.info(
                     "SQL normalization applied",
                     identifiers_normalized=norm_metrics["identifier_count"],
@@ -3078,17 +3081,23 @@ def _sync_execute():
                             f"Executing intermediate statement: {stmt[:80]}...",
                             session_id=session_id,
                         )
-                        self._safe_execute(stmt, optimized_params, is_embedded=True)
+                        self._safe_execute(
+                            stmt, optimized_params, is_embedded=True, session_id=session_id
+                        )
 
                     # Execute last statement and capture results
                     last_stmt = statements[-1]
                     logger.debug(
                         f"Executing final statement: {last_stmt[:80]}...", session_id=session_id
                     )
-                    result = self._safe_execute(last_stmt, optimized_params, is_embedded=True)
+                    result = self._safe_execute(
+                        last_stmt, optimized_params, is_embedded=True, session_id=session_id
+                    )
                 else:
                     # Single statement - execute normally
-                    result = self._safe_execute(optimized_sql, optimized_params, is_embedded=True)
+                    result = self._safe_execute(
+                        optimized_sql, optimized_params, is_embedded=True, session_id=session_id
+                    )
 
                 # RETURNING emulation: After INSERT/UPDATE/DELETE, fetch the affected row(s)
                 if returning_operation and returning_columns:
@@ -3684,10 +3693,14 @@ def _sync_external_execute():
 
                 # Execute all statements except the last
                 for stmt in statements[:-1]:
-                    self._safe_execute(stmt, optimized_params, is_embedded=False)
+                    self._safe_execute(
+                        stmt, optimized_params, is_embedded=False, session_id=session_id
+                    )
 
                 # Execute last statement and capture cursor
-                cursor = self._safe_execute(statements[-1], optimized_params, is_embedded=False)
+                cursor = self._safe_execute(
+                    statements[-1], optimized_params, is_embedded=False, session_id=session_id
+                )
 
                 # RETURNING emulation
                 if returning_operation and returning_columns:

src/iris_pgwire/protocol.py

@@ -1299,15 +1299,15 @@ async def _handle_single_statement(self, query: str, send_ready: bool = True):
             # Handle transaction commands first (no IRIS execution needed)
             query_upper = query.upper().strip()
             if query_upper in ("BEGIN", "START TRANSACTION"):
-                await self.iris_executor.begin_transaction()
+                await self.iris_executor.begin_transaction(session_id=self.connection_id)
                 await self.send_transaction_response("BEGIN", send_ready=send_ready)
                 return
             elif query_upper in ("COMMIT", "END"):
-                await self.iris_executor.commit_transaction()
+                await self.iris_executor.commit_transaction(session_id=self.connection_id)
                 await self.send_transaction_response("COMMIT", send_ready=send_ready)
                 return
             elif query_upper == "ROLLBACK":
-                await self.iris_executor.rollback_transaction()
+                await self.iris_executor.rollback_transaction(session_id=self.connection_id)
                 await self.send_transaction_response("ROLLBACK", send_ready=send_ready)
                 return
 
@@ -1373,7 +1373,9 @@ async def _handle_single_statement(self, query: str, send_ready: bool = True):
                 )
 
             # Execute translated SQL against IRIS
-            result = await self.iris_executor.execute_query(final_sql)
+            result = await self.iris_executor.execute_query(
+                final_sql, session_id=self.connection_id
+            )
 
             # Add translation metadata to result for debugging/monitoring
             if translation_result.get("translation_used"):
@@ -2379,7 +2381,7 @@ async def flush_batch(self):
                 return
 
             await self.iris_executor.execute_many(
-                sql_to_exec, params_to_exec, session_id=f"batch_{self.connection_id}"
+                sql_to_exec, params_to_exec, session_id=self.connection_id
             )
         except Exception as e:
             logger.error("Batch flush failed", connection_id=self.connection_id, error=str(e))
@@ -3035,7 +3037,9 @@ async def handle_describe_message(self, body: bytes):
                             param_count=param_count,
                         )
 
-                        result = await self.iris_executor.execute_query(query, params=dummy_params)
+                        result = await self.iris_executor.execute_query(
+                            query, params=dummy_params, session_id=self.connection_id
+                        )
 
                         if result.get("success") and result.get("columns"):
                             await self.send_row_description(result["columns"])
@@ -3105,7 +3109,9 @@ async def handle_describe_message(self, body: bytes):
                     ) or self.iris_executor.sql_parser.is_show_statement(query):
                         try:
                             result = await self.iris_executor.execute_query(
-                                query, params=portal.get("params", [])
+                                query,
+                                params=portal.get("params", []),
+                                session_id=self.connection_id,
                             )
                             if result.get("success") and result.get("columns"):
                                 result_formats = portal.get("result_formats", [])
@@ -3240,13 +3246,13 @@ async def handle_execute_message(self, body: bytes):
                 )
 
                 if transaction_type == "BEGIN":
-                    await self.iris_executor.begin_transaction()
+                    await self.iris_executor.begin_transaction(session_id=self.connection_id)
                     await self.send_transaction_response_extended_protocol("BEGIN")
                 elif transaction_type == "COMMIT":
-                    await self.iris_executor.commit_transaction()
+                    await self.iris_executor.commit_transaction(session_id=self.connection_id)
                     await self.send_transaction_response_extended_protocol("COMMIT")
                 elif transaction_type == "ROLLBACK":
-                    await self.iris_executor.rollback_transaction()
+                    await self.iris_executor.rollback_transaction(session_id=self.connection_id)
                     await self.send_transaction_response_extended_protocol("ROLLBACK")
                 return
 
@@ -3319,7 +3325,7 @@ async def handle_execute_message(self, body: bytes):
             await self.flush_batch()
 
             result = await self.iris_executor.execute_query(
-                query, params=params if params else None
+                query, params=params if params else None, session_id=self.connection_id
             )
 
             if result["success"]: