IGNITE-14057 Support big-endian systems

Fix primitives decoding on big-endian architectures

Author: ptupitsyn

pyignite/api/binary.py

@@ -86,7 +86,7 @@ def get_binary_type(
     if result.status != 0:
         return result
     result.value = {
-        'type_exists': response.type_exists
+        'type_exists': Bool.to_python(response.type_exists)
     }
     if hasattr(response, 'body'):
         result.value.update(body_struct.to_python(response.body))

pyignite/datatypes/internal.py

@@ -204,7 +204,7 @@ class Struct:
 
     def parse(
         self, client: 'Client'
-    ) -> Tuple[ctypes.BigEndianStructure, bytes]:
+    ) -> Tuple[ctypes.LittleEndianStructure, bytes]:
         buffer = b''
         fields = []
         values = {}
@@ -219,7 +219,7 @@ def parse(
             values[name] = buffer_fragment
 
         data_class = type(
-            'Struct',
+            'StructLE',
             (ctypes.LittleEndianStructure,),
             {
                 '_pack_': 1,

pyignite/datatypes/primitive.py

@@ -14,6 +14,7 @@
 # limitations under the License.
 
 import ctypes
+import sys
 
 from pyignite.constants import *
 from .base import IgniteDataType
@@ -48,13 +49,20 @@ class Primitive(IgniteDataType):
     def parse(cls, client: 'Client'):
         return cls.c_type, client.recv(ctypes.sizeof(cls.c_type))
 
-    @staticmethod
-    def to_python(ctype_object, *args, **kwargs):
+    @classmethod
+    def to_python(cls, ctype_object, *args, **kwargs):
         return ctype_object
 
     @classmethod
     def from_python(cls, value):
-        return bytes(cls.c_type(value))
+        return Primitive.fix_endianness(bytes(cls.c_type(value)))
+
+    @staticmethod
+    def fix_endianness(buf):
+        if len(buf) > 1 and sys.byteorder != PROTOCOL_BYTE_ORDER:
+            buf = buf[::-1]
+
+        return buf
 
 
 class Byte(Primitive):
@@ -122,4 +130,8 @@ def from_python(cls, value):
 class Bool(Primitive):
     _type_name = NAME_BOOLEAN
     _type_id = TYPE_BOOLEAN
-    c_type = ctypes.c_bool
+    c_type = ctypes.c_byte  # Use c_byte because c_bool throws endianness conversion error on BE systems.
+
+    @classmethod
+    def to_python(cls, ctype_object, *args, **kwargs):
+        return ctype_object != 0

pyignite/datatypes/primitive_objects.py

@@ -207,11 +207,16 @@ def from_python(cls, value):
 class BoolObject(DataObject):
     _type_name = NAME_BOOLEAN
     _type_id = TYPE_BOOLEAN
-    c_type = ctypes.c_bool
+    c_type = ctypes.c_byte  # Use c_byte because c_bool throws endianness conversion error on BE systems.
     type_code = TC_BOOL
     pythonic = bool
     default = False
 
     @staticmethod
     def hashcode(value: bool, *args, **kwargs) -> int:
         return 1231 if value else 1237
+
+    @classmethod
+    def to_python(cls, ctype_object, *args, **kwargs):
+        return ctype_object.value != 0
+

pyignite/queries/response.py

@@ -179,7 +179,7 @@ def _parse_success(self, conn: Connection, buffer: bytearray, fields: list):
         )
         fields += body_class._fields_ + [
             ('data', data_class),
-            ('more', ctypes.c_bool),
+            ('more', ctypes.c_byte),
         ]
 
     def _create_parse_result(self, conn: Connection, header_class, fields: list, buffer: bytearray):
@@ -226,3 +226,193 @@ def to_python(self, ctype_object, *args, **kwargs):
                     )
                 result['data'].append(row)
             return result
+
+
+@attr.s
+class Response130:
+    following = attr.ib(type=list, factory=list)
+    _response_header = None
+
+    def __attrs_post_init__(self):
+        # replace None with empty list
+        self.following = self.following or []
+
+    @classmethod
+    def build_header(cls):
+        if cls._response_header is None:
+            cls._response_header = type(
+                'ResponseHeader',
+                (ctypes.LittleEndianStructure,),
+                {
+                    '_pack_': 1,
+                    '_fields_': [
+                        ('length', ctypes.c_int),
+                        ('query_id', ctypes.c_longlong),
+                        ('status_code', ctypes.c_int),
+                    ],
+                },
+            )
+        return cls._response_header
+
+    def parse(self, client: 'Client'):
+        header_class = self.build_header()
+        buffer = client.recv(ctypes.sizeof(header_class))
+        header = header_class.from_buffer_copy(buffer)
+        fields = []
+
+        if header.status_code == OP_SUCCESS:
+            for name, ignite_type in self.following:
+                c_type, buffer_fragment = ignite_type.parse(client)
+                buffer += buffer_fragment
+                fields.append((name, c_type))
+        else:
+            c_type, buffer_fragment = String.parse(client)
+            buffer += buffer_fragment
+            fields.append(('error_message', c_type))
+
+        response_class = type(
+            'Response',
+            (header_class,),
+            {
+                '_pack_': 1,
+                '_fields_': fields,
+            }
+        )
+        return response_class, buffer
+
+    def to_python(self, ctype_object, *args, **kwargs):
+        result = OrderedDict()
+
+        for name, c_type in self.following:
+            result[name] = c_type.to_python(
+                getattr(ctype_object, name),
+                *args, **kwargs
+            )
+
+        return result if result else None
+
+
+@attr.s
+class SQLResponse130(Response130):
+    """
+    The response class of SQL functions is special in the way the row-column
+    data is counted in it. Basically, Ignite thin client API is following a
+    “counter right before the counted objects” rule in most of its parts.
+    SQL ops are breaking this rule.
+    """
+    include_field_names = attr.ib(type=bool, default=False)
+    has_cursor = attr.ib(type=bool, default=False)
+
+    def fields_or_field_count(self):
+        if self.include_field_names:
+            return 'fields', StringArray
+        return 'field_count', Int
+
+    def parse(self, client: 'Client'):
+        header_class = self.build_header()
+        buffer = client.recv(ctypes.sizeof(header_class))
+        header = header_class.from_buffer_copy(buffer)
+        fields = []
+
+        if header.status_code == OP_SUCCESS:
+            following = [
+                self.fields_or_field_count(),
+                ('row_count', Int),
+            ]
+            if self.has_cursor:
+                following.insert(0, ('cursor', Long))
+            body_struct = Struct(following)
+            body_class, body_buffer = body_struct.parse(client)
+            body = body_class.from_buffer_copy(body_buffer)
+
+            if self.include_field_names:
+                field_count = body.fields.length
+            else:
+                field_count = body.field_count
+
+            data_fields = []
+            data_buffer = b''
+            for i in range(body.row_count):
+                row_fields = []
+                row_buffer = b''
+                for j in range(field_count):
+                    field_class, field_buffer = AnyDataObject.parse(client)
+                    row_fields.append(('column_{}'.format(j), field_class))
+                    row_buffer += field_buffer
+
+                row_class = type(
+                    'SQLResponseRow',
+                    (ctypes.LittleEndianStructure,),
+                    {
+                        '_pack_': 1,
+                        '_fields_': row_fields,
+                    }
+                )
+                data_fields.append(('row_{}'.format(i), row_class))
+                data_buffer += row_buffer
+
+            data_class = type(
+                'SQLResponseData',
+                (ctypes.LittleEndianStructure,),
+                {
+                    '_pack_': 1,
+                    '_fields_': data_fields,
+                }
+            )
+            fields += body_class._fields_ + [
+                ('data', data_class),
+                ('more', ctypes.c_byte),
+            ]
+            buffer += body_buffer + data_buffer
+        else:
+            c_type, buffer_fragment = String.parse(client)
+            buffer += buffer_fragment
+            fields.append(('error_message', c_type))
+
+        final_class = type(
+            'SQLResponse',
+            (header_class,),
+            {
+                '_pack_': 1,
+                '_fields_': fields,
+            }
+        )
+        buffer += client.recv(ctypes.sizeof(final_class) - len(buffer))
+        return final_class, buffer
+
+    def to_python(self, ctype_object, *args, **kwargs):
+        if ctype_object.status_code == 0:
+            result = {
+                'more': Bool.to_python(
+                    ctype_object.more, *args, **kwargs
+                ),
+                'data': [],
+            }
+            if hasattr(ctype_object, 'fields'):
+                result['fields'] = StringArray.to_python(
+                    ctype_object.fields, *args, **kwargs
+                )
+            else:
+                result['field_count'] = Int.to_python(
+                    ctype_object.field_count, *args, **kwargs
+                )
+            if hasattr(ctype_object, 'cursor'):
+                result['cursor'] = Long.to_python(
+                    ctype_object.cursor, *args, **kwargs
+                )
+            for row_item in ctype_object.data._fields_:
+                row_name = row_item[0]
+                row_object = getattr(ctype_object.data, row_name)
+                row = []
+                for col_item in row_object._fields_:
+                    col_name = col_item[0]
+                    col_object = getattr(row_object, col_name)
+                    row.append(
+                        AnyDataObject.to_python(col_object, *args, **kwargs)
+                    )
+                result['data'].append(row)
+            return result
+
+
+Response120 = Response130
+SQLResponse120 = SQLResponse130

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+log_cli = True

tests/test_binary.py

@@ -225,7 +225,7 @@ class OuterType(
 
 def test_add_schema_to_binary_object(client):
 
-    migrate_cache = client.create_cache('migrate_binary')
+    migrate_cache = client.get_or_create_cache('migrate_binary')
 
     class MyBinaryType(
         metaclass=GenericObjectMeta,