tried to impelement query result packet decoding correctly

Author: kerryjiang

src/SuperSocket.MySQL/MySQLConnection.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Buffers;
 using System.Collections.Generic;
+using System.Linq;
 using System.Net;
 using System.Net.Sockets;
 using System.Security.Cryptography;
@@ -26,15 +27,23 @@ public class MySQLConnection : EasyClient<MySQLPacket>
 
         public bool IsAuthenticated { get; private set; }
 
+        private readonly MySQLFilterContext filterContext;
+
         public MySQLConnection(string host, int port, string userName, string password, ILogger logger = null)
-            : base(new MySQLPacketFilter(MySQLPacketDecoder.ClientInstance), logger)
+            : this(new MySQLPacketFilter(MySQLPacketDecoder.ClientInstance), logger)
         {
             _host = host ?? throw new ArgumentNullException(nameof(host));
             _port = port > 0 ? port : DefaultPort;
             _userName = userName ?? throw new ArgumentNullException(nameof(userName));
             _password = password ?? throw new ArgumentNullException(nameof(password));
         }
 
+        private MySQLConnection(MySQLPacketFilter packetFilter, ILogger logger)
+            : base(packetFilter, logger)
+        {
+            filterContext = packetFilter.Context as MySQLFilterContext;
+        }
+
         public async Task ConnectAsync(CancellationToken cancellationToken = default)
         {
             if (string.IsNullOrEmpty(_host))
@@ -83,6 +92,7 @@ public async Task ConnectAsync(CancellationToken cancellationToken = default)
                 case OKPacket okPacket:
                     // Authentication successful
                     IsAuthenticated = true;
+                    filterContext.State = MySQLConnectionState.Authenticated;
                     break;
                 case ErrorPacket errorPacket:
                     // Authentication failed
@@ -95,6 +105,7 @@ public async Task ConnectAsync(CancellationToken cancellationToken = default)
                     if ((eofPacket.StatusFlags & 0x0002) != 0)
                     {
                         IsAuthenticated = true;
+                        filterContext.State = MySQLConnectionState.Authenticated;
                         break;
                     }
                     else
@@ -164,12 +175,27 @@ public async Task<QueryResultPacket> ExecuteQueryAsync(string query, Cancellatio
                     SequenceId = 0
                 };
 
+                filterContext.State = MySQLConnectionState.CommandPhase;
                 await SendAsync(PacketEncoder, commandPacket).ConfigureAwait(false);
 
                 // Read response
                 var response = await ReceiveAsync().ConfigureAwait(false);
 
-                return (QueryResultPacket)response;
+                // Handle different response types
+                switch (response)
+                {
+                    case ErrorPacket errorPacket:
+                        // Query failed
+                        return QueryResultPacket.FromError((short)errorPacket.ErrorCode, errorPacket.ErrorMessage);
+
+                    case QueryResultPacket queryResult:
+                        // Already a query result packet
+                        return queryResult;
+
+                    default:
+                        // Handle result set responses (SELECT queries)
+                        return await ReadResultSetAsync(response).ConfigureAwait(false);
+                }
             }
             catch (Exception ex)
             {
@@ -186,22 +212,17 @@ public async Task<QueryResultPacket> ExecuteQueryAsync(string query, Cancellatio
         public async Task<string> ExecuteQueryStringAsync(string query, CancellationToken cancellationToken = default)
         {
             var result = await ExecuteQueryAsync(query, cancellationToken).ConfigureAwait(false);
-            
+
             if (!result.IsSuccess)
             {
                 return $"Error {result.ErrorCode}: {result.ErrorMessage}";
             }
 
-            if (result.Columns == null || result.Columns.Count == 0)
-            {
-                return $"Query executed successfully. {result.AffectedRows} rows affected.";
-            }
-
             var sb = new StringBuilder();
-            
+
             // Add column headers
             sb.AppendLine(string.Join("\t", result.Columns));
-            
+
             // Add separator line
             sb.AppendLine(new string('-', result.Columns.Count * 10));
 
@@ -213,9 +234,9 @@ public async Task<string> ExecuteQueryStringAsync(string query, CancellationToke
                     sb.AppendLine(string.Join("\t", row ?? new string[result.Columns.Count]));
                 }
             }
-            
+
             sb.AppendLine($"\n{result.Rows?.Count ?? 0} rows returned.");
-            
+
             return sb.ToString();
         }
 
@@ -235,5 +256,50 @@ public async Task DisconnectAsync()
                 IsAuthenticated = false;
             }
         }
+
+        /// <summary>
+        /// Reads a complete result set from the MySQL server
+        /// </summary>
+        /// <param name="firstPacket">The first packet received after sending the query</param>
+        /// <returns>A QueryResultPacket containing the complete result set</returns>
+        private Task<QueryResultPacket> ReadResultSetAsync(MySQLPacket firstPacket)
+        {
+            try
+            {
+                // If the first packet is already a QueryResultPacket (decoded by UnknownPacket), return it
+                if (firstPacket is QueryResultPacket queryResult)
+                {
+                    return Task.FromResult(queryResult);
+                }
+
+                // If the first packet is an UnknownPacket, it should have been decoded to QueryResultPacket
+                // but if that failed, we'll create a minimal fallback
+                if (firstPacket is UnknownPacket)
+                {
+                    // Try to read additional packets to build a result set
+                    // This is a simplified implementation that attempts to handle basic SELECT queries
+
+                    var columns = new List<ColumnDefinitionPacket>();
+                    var rows = new List<IReadOnlyList<string>>();
+
+                    // For now, create a minimal successful result
+                    // This could be enhanced to parse more complex result sets in the future
+                    return Task.FromResult(QueryResultPacket.FromResultSet(columns.AsReadOnly(), rows.AsReadOnly()));
+                }
+
+                // For any other packet type, treat as an unexpected response
+                return Task.FromResult(QueryResultPacket.FromError(-1, $"Unexpected packet type in result set: {firstPacket?.GetType().Name ?? "null"}"));
+            }
+            catch (Exception ex)
+            {
+                return Task.FromResult(QueryResultPacket.FromError(-1, $"Failed to read result set: {ex.Message}"));
+            }
+        }
+
+        protected override void OnClosed(object sender, EventArgs e)
+        {
+            filterContext.State = MySQLConnectionState.Closed;
+            base.OnClosed(sender, e);
+        }
     }
 }

src/SuperSocket.MySQL/MySQLFilterContext.cs

@@ -0,0 +1,49 @@
+using System;
+using System.Collections.Generic;
+using SuperSocket.MySQL.Packets;
+
+namespace SuperSocket.MySQL
+{
+    internal class MySQLFilterContext
+    {
+        public int SequenceId { get; set; }
+        
+        public bool IsHandshakeCompleted { get; set; }
+        
+        public MySQLConnectionState State { get; set; }
+
+        public MySQLPacket NextPacket { get; set; }
+
+        public int QueryResultColumnCount { get; set; }
+
+        public List<ColumnDefinitionPacket> ColumnDefinitionPackets { get; set; }
+
+        public MySQLFilterContext()
+        {
+            SequenceId = 0;
+            IsHandshakeCompleted = false;
+            State = MySQLConnectionState.Initial;
+        }
+        
+        public void Reset()
+        {
+            SequenceId = 0;
+            IsHandshakeCompleted = false;
+            State = MySQLConnectionState.Initial;
+        }
+        
+        public void IncrementSequenceId()
+        {
+            SequenceId = (SequenceId + 1) % 256;
+        }
+    }
+    
+    public enum MySQLConnectionState
+    {
+        Initial,
+        HandshakeInitiated,
+        Authenticated,
+        CommandPhase,
+        Closed
+    }
+}

src/SuperSocket.MySQL/MySQLPacketDecoder.cs

@@ -30,13 +30,14 @@ public MySQLPacket Decode(ref ReadOnlySequence<byte> buffer, object context)
             reader.Advance(3); // Skip the first 3 bytes of the header
             reader.TryRead(out var sequenceId); // Read the sequence ID
 
-            var filter = context as MySQLPacketFilter;
+            var filterContext = context as MySQLFilterContext;
 
             var packetType = -1;
 
             // Read the first byte to determine packet type
-            if (filter.ReceivedHandshake)
+            if (filterContext.State != MySQLConnectionState.Initial)
             {
+                // In handshake state, we expect the first byte to be the packet type
                 if (!reader.TryRead(out var packetTypeByte))
                     return null;
 
@@ -48,8 +49,10 @@ public MySQLPacket Decode(ref ReadOnlySequence<byte> buffer, object context)
             package = package.Decode(ref reader, context);
             package.SequenceId = sequenceId;
 
-            if (!filter.ReceivedHandshake)
-                filter.ReceivedHandshake = true;
+            if (filterContext.State == MySQLConnectionState.Initial)
+            {
+                filterContext.State = MySQLConnectionState.HandshakeInitiated;
+            }
 
             return package;
         }

src/SuperSocket.MySQL/MySQLPacketFilter.cs

@@ -8,13 +8,11 @@ internal class MySQLPacketFilter : FixedHeaderPipelineFilter<MySQLPacket>
     {
         private const int headerSize = 4; // MySQL package header size is 4 bytes
 
-        internal bool ReceivedHandshake { get; set; }
-
         public MySQLPacketFilter(IPackageDecoder<MySQLPacket> decoder)
             : base(headerSize)
         {
             this.Decoder = decoder ?? throw new ArgumentNullException(nameof(decoder));
-            this.Context = this;
+            this.Context = new MySQLFilterContext();
         }
 
         protected override int GetBodyLengthFromHeader(ref ReadOnlySequence<byte> buffer)
@@ -27,5 +25,19 @@ protected override int GetBodyLengthFromHeader(ref ReadOnlySequence<byte> buffer
 
             return byte2 * 256 * 256 + byte1 * 256 + byte0;
         }
+
+        public override MySQLPacket Filter(ref SequenceReader<byte> reader)
+        {
+            var packet = base.Filter(ref reader);
+
+            if (packet == null || packet.IsPartialPacket)
+            {
+                // If the packet is null or a partial packet, we cannot return it yet
+                // We will wait for more data to complete the packet
+                return null;
+            }
+
+            return packet;
+        }
     }
 }

src/SuperSocket.MySQL/Packets/ColumnCountPacket.cs

@@ -0,0 +1,31 @@
+using System;
+using System.Buffers;
+using System.Collections.Generic;
+using SuperSocket.ProtoBase;
+
+namespace SuperSocket.MySQL.Packets
+{
+    public class ColumnCountPacket : MySQLPacket
+    {
+        public ulong ColumnCount { get; set; }
+
+        protected internal override MySQLPacket Decode(ref SequenceReader<byte> reader, object context)
+        {
+            ColumnCount = reader.ReadLengthEncodedInteger();
+
+            var filterContext = context as MySQLFilterContext;
+            filterContext.NextPacket = new ColumnDefinitionPacket();
+            filterContext.QueryResultColumnCount = (int)ColumnCount;
+            filterContext.ColumnDefinitionPackets = new List<ColumnDefinitionPacket>(filterContext.QueryResultColumnCount);
+
+            return this;
+        }
+
+        protected internal override int Encode(IBufferWriter<byte> writer)
+        {
+            throw new NotImplementedException();
+        }
+
+        internal override bool IsPartialPacket => true;
+    }
+}

src/SuperSocket.MySQL/Packets/ColumnDefinitionPacket.cs

@@ -32,6 +32,7 @@ protected internal override MySQLPacket Decode(ref SequenceReader<byte> reader,
             // Read catalog (length-encoded string)
             if (!reader.TryReadLengthEncodedString(out string catalog))
                 throw new InvalidOperationException("Failed to read catalog");
+
             Catalog = catalog;
 
             // Read schema (length-encoded string)
@@ -90,6 +91,18 @@ protected internal override MySQLPacket Decode(ref SequenceReader<byte> reader,
             // Skip the two null bytes that follow
             reader.Advance(2);
 
+            var filterContext = context as MySQLFilterContext;
+            filterContext.ColumnDefinitionPackets.Add(this);
+
+            if (filterContext.QueryResultColumnCount > filterContext.ColumnDefinitionPackets.Count)
+            {
+                filterContext.NextPacket = new ColumnDefinitionPacket();
+            }
+            else
+            {
+                filterContext.NextPacket = new ResultRowsPacket();
+            }
+
             return this;
         }
 
@@ -156,5 +169,7 @@ protected internal override int Encode(IBufferWriter<byte> writer)
 
             return bytesWritten;
         }
+
+        internal override bool IsPartialPacket => true;
     }
 }

src/SuperSocket.MySQL/Packets/MySQLPacket.cs

@@ -15,5 +15,7 @@ public abstract class MySQLPacket
         protected internal abstract MySQLPacket Decode(ref SequenceReader<byte> reader, object context);
 
         protected internal abstract int Encode(IBufferWriter<byte> writer);
+
+        internal virtual bool IsPartialPacket => false;
     }
 }

src/SuperSocket.MySQL/Packets/NonQueryResultPacket.cs

@@ -0,0 +1,32 @@
+using System;
+using System.Buffers;
+
+namespace SuperSocket.MySQL.Packets
+{
+    public class NonQueryResultPacket : MySQLPacket
+    {
+        public long AffectedRows { get; private set; }
+        public long LastInsertId { get; private set; }
+
+        protected internal override MySQLPacket Decode(ref SequenceReader<byte> reader, object context)
+        {
+            // Read affected rows and last insert ID for non-SELECT queries
+            if (reader.TryReadLengthEncodedInteger(out long affectedRows))
+            {
+                AffectedRows = affectedRows;
+            }
+
+            if (reader.TryReadLengthEncodedInteger(out long lastInsertId))
+            {
+                LastInsertId = lastInsertId;
+            }
+
+            return this;
+        }
+
+        protected internal override int Encode(IBufferWriter<byte> writer)
+        {
+            throw new NotImplementedException();
+        }
+    }
+}