Fix for CONPY-98:

If a string has a binary collation/charset it needs to be converted to a Binary instead of Unicode.
mariadb-corporation · Aug 7, 2020 · a749c53 · a749c53
1 parent 4321164
commit a749c53
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 17 deletions.
diff --git a/mariadb/mariadb_codecs.c b/mariadb/mariadb_codecs.c
@@ -458,15 +458,24 @@ field_fetch_fromtext(MrdbCursor *self, char *data, unsigned int column)
         case MYSQL_TYPE_SET:
         case MYSQL_TYPE_ENUM:
         {
-            unsigned long utf8len;
+            unsigned long len;
 
-            self->values[column]=
-                PyUnicode_FromStringAndSize((const char *)data,
-                                            (Py_ssize_t)length[column]);
-            utf8len= (unsigned long)PyUnicode_GET_LENGTH(self->values[column]);
-            if (utf8len > self->fields[column].max_length)
+            if (self->fields[column].flags & BINARY_FLAG ||
+                self->fields[column].charsetnr == CHARSET_BINARY) 
+            {
+                self->values[column]= 
+                        PyBytes_FromStringAndSize((const char *)data, 
+                                                       (Py_ssize_t)length[column]);
+                len= (unsigned long)length[column];
+            } else {
+                self->values[column]=
+                    PyUnicode_FromStringAndSize((const char *)data,
+                                                (Py_ssize_t)length[column]);
+                len= (unsigned long)PyUnicode_GET_LENGTH(self->values[column]);
+            }
+            if (len > self->fields[column].max_length)
             {
-                self->fields[column].max_length= utf8len;
+                self->fields[column].max_length= len;
             }
             break;
         }
@@ -690,20 +699,29 @@ field_fetch_callback(void *data, unsigned int column, unsigned char **row)
         case MYSQL_TYPE_DECIMAL:
         case MYSQL_TYPE_SET:
         case MYSQL_TYPE_ENUM:
-            {
-                unsigned long length;
-                unsigned long utf8len;
-                length= mysql_net_field_length(row);
+        {
+            unsigned long length;
+            unsigned long utf8len;
+            length= mysql_net_field_length(row);
 
+            if (self->fields[column].flags & BINARY_FLAG ||
+                self->fields[column].charsetnr == CHARSET_BINARY) 
+            {
                 self->values[column]= 
-                    PyUnicode_FromStringAndSize((const char *)*row,
-                                                (Py_ssize_t)length);
-                utf8len= 
-                   (unsigned long)PyUnicode_GET_LENGTH(self->values[column]);
-                if (utf8len > self->fields[column].max_length)
+                        PyBytes_FromStringAndSize((const char *)*row, 
+                                                   (Py_ssize_t)length);
+                if (length > self->fields[column].max_length)
+                    self->fields[column].max_length= length;
+            } else {
+                 self->values[column]= 
+                 PyUnicode_FromStringAndSize((const char *)*row,
+                                             (Py_ssize_t)length);
+                 utf8len= (unsigned long)PyUnicode_GET_LENGTH(self->values[column]);
+                 if (utf8len > self->fields[column].max_length)
                     self->fields[column].max_length= utf8len;
-                *row+= length;
             }
+            *row+= length;
+        }
         default:
             break;
     }

diff --git a/testing/test/integration/test_cursor.py b/testing/test/integration/test_cursor.py
@@ -1015,6 +1015,17 @@ def test_conpy94(self):
         self.assertEqual(row[0], 2)
         del cur
 
+    def test_conpy98(self):
+        con= create_connection()
+        cursor=con.cursor()
+        cursor.execute("SELECT CAST('foo' AS BINARY) AS anon_1 WHERE 1=?", (1,))
+        row= cursor.fetchone()
+        self.assertEqual(row[0], b'foo')
+        cursor.execute("SELECT CAST('foo' AS BINARY) AS anon_1")
+        row= cursor.fetchone()
+        self.assertEqual(row[0], b'foo')
+        del cursor
+
     def test_conpy91(self):
         with create_connection() as connection:
             with connection.cursor() as cursor: