Apply string interpolation in mogrify if args is not None

singlestoredb/config.py

@@ -251,6 +251,12 @@
     environ='SINGLESTOREDB_VECTOR_DATA_FORMAT',
 )
 
+register_option(
+    'interpolate_query_with_empty_args', 'bool', check_bool, False,
+    'Should mogrify apply string interpolation when args is an empty tuple/list? ',
+    environ='SINGLESTOREDB_interpolate_query_with_empty_args',
+)
+
 register_option(
     'fusion.enabled', 'bool', check_bool, False,
     'Should Fusion SQL queries be enabled?',

singlestoredb/connection.py

@@ -1118,9 +1118,15 @@ def __init__(self, **kwargs: Any):
     def _convert_params(
         cls, oper: str,
         params: Optional[Union[Sequence[Any], Dict[str, Any], Any]],
+        interpolate_query_with_empty_args: bool = False,
     ) -> Tuple[Any, ...]:
         """Convert query to correct parameter format."""
-        if params:
+        if interpolate_query_with_empty_args:
+            should_convert = params is not None
+        else:
+            should_convert = bool(params)
+
+        if should_convert:
 
             if cls._map_param_converter is None:
                 cls._map_param_converter = sqlparams.SQLParams(
@@ -1333,6 +1339,7 @@ def connect(
     enable_extended_data_types: Optional[bool] = None,
     vector_data_format: Optional[str] = None,
     parse_json: Optional[bool] = None,
+    interpolate_query_with_empty_args: Optional[bool] = None,
 ) -> Connection:
     """
     Return a SingleStoreDB connection.
@@ -1418,6 +1425,9 @@ def connect(
         Should extended data types (BSON, vector) be enabled?
     vector_data_format : str, optional
         Format for vector types: json or binary
+    interpolate_query_with_empty_args : bool, optional
+        Should the connector apply parameter interpolation even when the
+        parameters are empty? This corresponds to pymysql/mysqlclient's handling
 
     Examples
     --------

singlestoredb/http/connection.py

@@ -548,7 +548,12 @@ def _execute(
         if handler is not None:
             return self._execute_fusion_query(oper, params, handler=handler)
 
-        oper, params = self._connection._convert_params(oper, params)
+        interpolate_query_with_empty_args = self._connection.connection_params.get(
+            'interpolate_query_with_empty_args', False,
+        )
+        oper, params = self._connection._convert_params(
+            oper, params, interpolate_query_with_empty_args,
+        )
 
         log_query(oper, params)
 

singlestoredb/mysql/connection.py

@@ -360,6 +360,7 @@ def __init__(  # noqa: C901
         track_env=False,
         enable_extended_data_types=True,
         vector_data_format='binary',
+        interpolate_query_with_empty_args=None,
     ):
         BaseConnection.__init__(**dict(locals()))
 
@@ -614,6 +615,7 @@ def _config(key, arg):
         self._auth_plugin_map = auth_plugin_map or {}
         self._binary_prefix = binary_prefix
         self.server_public_key = server_public_key
+        self.interpolate_query_with_empty_args = interpolate_query_with_empty_args
 
         if self.connection_params['nan_as_null'] or \
                 self.connection_params['inf_as_null']:

singlestoredb/mysql/cursors.py

@@ -181,7 +181,12 @@ def mogrify(self, query, args=None):
         """
         conn = self._get_db()
 
-        if args:
+        if conn.interpolate_query_with_empty_args:
+            should_interpolate = args is not None
+        else:
+            should_interpolate = bool(args)
+
+        if should_interpolate:
             query = query % self._escape_args(args, conn)
 
         return query

singlestoredb/tests/test.sql

@@ -708,5 +708,10 @@ INSERT INTO bool_data_with_nulls SET id='nt', bool_a=NULL, bool_b=TRUE;
 INSERT INTO bool_data_with_nulls SET id='nn', bool_a=NULL, bool_b=NULL;
 INSERT INTO bool_data_with_nulls SET id='ff', bool_a=FALSE, bool_b=FALSE;
 
+CREATE TABLE IF NOT EXISTS test_val_with_percent (
+    i VARCHAR(16)
+);
+-- Double percent sign for execution from python
+INSERT INTO test_val_with_percent VALUES ('a%a');
 
 COMMIT;

singlestoredb/tests/test_connection.py

@@ -3144,6 +3144,30 @@ def test_f16_vectors(self):
             decimal=2,
         )
 
+    def test_mogrify_val_with_percent(self):
+        conn = s2.connect(
+            database=type(self).dbname,
+            interpolate_query_with_empty_args=True,
+        )
+        cur = conn.cursor()
+        val_with_percent = 'a%a'
+        cur.execute(
+            '''SELECT REPLACE(i, "%%", "\\%%")
+            FROM test_val_with_percent''',
+            (),
+        )
+        res1 = cur.fetchall()
+        assert res1[0][0] == 'a\\%a'
+
+        cur.execute(
+            '''SELECT REPLACE(i, "%%", "\\%%")
+            FROM test_val_with_percent
+            WHERE i = %s''',
+            (val_with_percent,),
+        )
+        res2 = cur.fetchall()
+        assert res2[0][0] == 'a\\%a'
+
 
 if __name__ == '__main__':
     import nose2

singlestoredb/utils/mogrify.py

@@ -123,6 +123,7 @@ def mogrify(
     encoders: Optional[Encoders] = None,
     server_status: int = 0,
     binary_prefix: bool = False,
+    interpolate_query_with_empty_args: bool = False,
 ) -> Union[str, bytes]:
     """
     Returns the exact string sent to the database by calling the execute() method.
@@ -135,13 +136,21 @@ def mogrify(
         Query to mogrify.
     args : Sequence[Any] or Dict[str, Any] or Any, optional
         Parameters used with query. (optional)
+    interpolate_query_with_empty_args : bool, optional
+        If True, apply string interpolation even when args is an empty tuple/list.
+        If False, only apply when args is truthy. Defaults to False.
 
     Returns
     -------
     str : The query with argument binding applied.
 
     """
-    if args:
+    if interpolate_query_with_empty_args:
+        should_interpolate = args is not None
+    else:
+        should_interpolate = bool(args)
+
+    if should_interpolate:
         query = query % _escape_args(
             args, charset=charset,
             encoders=encoders,