fix: resolve memory leaks in result set parsing

Author: rusher

mariadb/impl/client/async_client.py

@@ -505,7 +505,12 @@ async def _parse_result_set(self, packet: PacketBuffer, config: 'Configuration',
             if not self.context.isEofDeprecated():
                 # skip intermediate EOF packet
                 (await self.read_stream.read_payload()).release()
+                        # Build decoder list once for all rows (performance optimization)
+            decoders = self._build_decoder_list(columns, is_binary)
 
+            # Select appropriate row parser based on protocol
+            row_parser = self._parse_binary_row_data if is_binary else self._parse_text_row_data
+
             # Step 4: If unbuffered, create streaming result
             if not buffered:
                 from ..result import AsyncStreamingResult
@@ -516,7 +521,8 @@ async def _parse_result_set(self, packet: PacketBuffer, config: 'Configuration',
                     column_count=column_count,
                     config=config,
                     is_binary=is_binary,
-                    row_parser=self._parse_row_data  # Pass row parser function
+                    row_parser=row_parser,  # Pass appropriate row parser function
+                    decoders=decoders  # Pass pre-built decoder list
                 )
 
                 # Create completion with streaming result
@@ -527,47 +533,50 @@ async def _parse_result_set(self, packet: PacketBuffer, config: 'Configuration',
             # Step 5: Read row data packets until EOF (buffered mode)
             rows = []
             parser = PayloadParser(None)
-            while True:
-                row_packet = await self.read_stream.read_payload()
-                # Check for EOF/OK packet based on DEPRECATE_EOF capability and packet length
-                # EOF/OK packets start with 0xFE and have specific length constraints
-                if (row_packet[0] == self.EOF_PACKET and 
-                    ((self.context.isEofDeprecated() and len(row_packet) < 16777215) or 
-                        (not self.context.isEofDeprecated() and len(row_packet) < 8))):
-                    
-                    if not self.context.isEofDeprecated():
-                        # Traditional EOF packet
-                        completion = EofPacket.decode(row_packet, self.context)
-                    else:
-                        # OK packet with 0xFE header (DEPRECATE_EOF enabled) - use existing OK packet parser
-                        completion = OkPacket.decode(row_packet, self.context)
-                    
-                    # Apply converters to all rows at once
-                    rows = self._apply_converters_to_rows(rows, columns, config)
+            try:
+                while True:
+                    row_packet = await self.read_stream.read_payload()
+                    # Check for EOF/OK packet based on DEPRECATE_EOF capability and packet length
+                    # EOF/OK packets start with 0xFE and have specific length constraints
+                    if (row_packet[0] == self.EOF_PACKET and 
+                        ((self.context.isEofDeprecated() and len(row_packet) < 16777215) or 
+                            (not self.context.isEofDeprecated() and len(row_packet) < 8))):
+                        
+                        if not self.context.isEofDeprecated():
+                            # Traditional EOF packet
+                            completion = EofPacket.decode(row_packet, self.context)
+                        else:
+                            # OK packet with 0xFE header (DEPRECATE_EOF enabled) - use existing OK packet parser
+                            completion = OkPacket.decode(row_packet, self.context)
+                        
+                        # Apply converters to all rows at once
+                        rows = self._apply_converters_to_rows(rows, columns, config)
 
-                    # Create AsyncCompleteResult with all rows
-                    from ..result import AsyncCompleteResult
-                    complete_result = AsyncCompleteResult(
-                        columns=columns,
-                        column_count=column_count,
-                        config=config,
-                        rows=rows,
-                        is_binary=is_binary
-                    )
-                    
-                    # Store result object in completion
-                    completion.result_set = complete_result
-                                        
-                    return completion
-                elif row_packet[0] == self.ERROR_PACKET:
-                    raise ErrorPacket.decode(row_packet, self.context).toError(self.exception_factory)
+                        # Create AsyncCompleteResult with all rows
+                        from ..result import AsyncCompleteResult
+                        complete_result = AsyncCompleteResult(
+                            columns=columns,
+                            column_count=column_count,
+                            config=config,
+                            rows=rows,
+                            is_binary=is_binary
+                        )
+                        
+                        # Store result object in completion
+                        completion.result_set = complete_result
+                                            
+                        return completion
+                    elif row_packet[0] == self.ERROR_PACKET:
+                        raise ErrorPacket.decode(row_packet, self.context).toError(self.exception_factory)
 
-                else:
-                    # Row data packet
-                    parser.set_buffer(row_packet)
-                    row_data = self._parse_row_data(parser, columns, config, is_binary)
-                    rows.append(row_data)
-            
+                    else:
+                        # Row data packet
+                        # Row data packet - use pre-built decoders
+                        parser.set_buffer(row_packet)
+                        rows.append(row_parser(parser, columns, config, decoders))
+            finally:
+                if parser and parser.packet:
+                    parser.packet.release()
 
         except Exception as e:
             raise OperationalError(f"Failed to parse result set: {e}")

mariadb/impl/client/base_client.py

@@ -569,6 +569,7 @@ def _parse_text_row_data(self, parser: PayloadParser, columns: List[ColumnDefini
         for i, decoder in enumerate(decoders):
             value = decoder(parser, columns[i], config)
             row_values[i] = value
+        parser.packet.release()
         return tuple(row_values)
 
     def _parse_binary_row_data(self, parser: PayloadParser, columns: List[ColumnDefinitionPacket], config: 'Configuration', decoders: List[Callable]) -> tuple:
@@ -590,6 +591,7 @@ def _parse_binary_row_data(self, parser: PayloadParser, columns: List[ColumnDefi
             # Use pre-built decoder for non-NULL values
             value = decoder(parser, columns[i], config)
             row_values[i] = value
+        parser.packet.release()    
         return tuple(row_values)
 
 

mariadb/impl/client/socket/payload_parser.py

@@ -152,7 +152,7 @@ def read_null_terminated_string(self, encoding: str = 'utf-8') -> str:
             if self.packet[i] == 0x00:
                 string_data = bytes(self.packet[self.pos:i]).decode(encoding)
                 self.pos = i + 1
-            return string_data
+                return string_data
         string_data = bytes(self.packet[self.pos:]).decode(encoding)        
         self.pos = len(self.packet)
         return string_data

mariadb/impl/client/socket/read_stream.py

@@ -47,7 +47,7 @@ def __len__(self):
 
     def release(self) -> None:
         """Release buffer back to stream (only needed for views)"""
-        if self._is_view:
+        if self._is_view and self._stream:
             self._stream._release_buffer()
             self._stream = None
 

mariadb/impl/client/sync_client.py

@@ -481,7 +481,7 @@ def _parse_result_set(self, packet: PacketBuffer, config: 'Configuration', is_bi
                         parser.set_buffer(row_packet)
                         rows.append(row_parser(parser, columns, config, decoders))
             finally:
-                if parser.packet:
+                if parser and parser.packet:
                     parser.packet.release()