From 4ea690ff7073a5f39943da20bcaa7a65b0463605 Mon Sep 17 00:00:00 2001
From: Matyx <matyx@users.noreply.github.com>
Date: Fri, 12 Jan 2024 09:45:38 +0100
Subject: [PATCH 1/2] Fix JsonBinaryDecoder when there is NULL in data or if
 there is long array with small numbers (#96)

Thank you
---
 .../JsonBinaryDecoderService.php              | 23 +++++++++++++++----
 1 file changed, 19 insertions(+), 4 deletions(-)

diff --git a/src/MySQLReplication/JsonBinaryDecoder/JsonBinaryDecoderService.php b/src/MySQLReplication/JsonBinaryDecoder/JsonBinaryDecoderService.php
index dc54782..a44dee1 100644
--- a/src/MySQLReplication/JsonBinaryDecoder/JsonBinaryDecoderService.php
+++ b/src/MySQLReplication/JsonBinaryDecoder/JsonBinaryDecoderService.php
@@ -74,6 +74,11 @@ public static function makeJsonBinaryDecoder(string $data): self
      */
     public function parseToString(): string
     {
+        // Sometimes, we can insert a NULL JSON even we set the JSON field as NOT NULL.
+        // If we meet this case, we can return a 'null' value.
+        if($this->binaryDataReader->getBinaryDataLength() === 0) {
+            return 'null';
+        }
         $this->parseJson($this->binaryDataReader->readUInt8());
 
         return $this->jsonBinaryDecoderFormatter->getJsonString();
@@ -198,7 +203,7 @@ private function parseArrayOrObject(int $type, int $intSize): array
 
         $entries = [];
         for ($i = 0; $i !== $elementCount; ++$i) {
-            $entries[$i] = $this->getOffsetOrInLinedValue($bytes, $intSize);
+            $entries[$i] = $this->getOffsetOrInLinedValue($bytes, $intSize, $valueEntrySize);
         }
 
         $keys = [];
@@ -238,11 +243,21 @@ private static function valueEntrySize(bool $large): int
      * @throws BinaryDataReaderException
      * @throws JsonBinaryDecoderException
      */
-    private function getOffsetOrInLinedValue(int $bytes, int $intSize): JsonBinaryDecoderValue
+    private function getOffsetOrInLinedValue(int $bytes, int $intSize, int $valueEntrySize): JsonBinaryDecoderValue
     {
         $type = $this->binaryDataReader->readUInt8();
+
         if (self::isInLinedType($type, $intSize)) {
-            return $this->parseScalar($type);
+            $scalar = $this->parseScalar($type);
+
+            // In binlog format, JSON arrays are fixed width elements, even though type value can be smaller.
+            // In order to properly process this case, we need to move cursor to the next element, which is on position 1 + $valueEntrySize (1 is length of type)
+            if($type === self::UINT16 || $type === self::INT16) {
+                $readNextBytes = $valueEntrySize - 2 - 1;
+                $this->binaryDataReader->read($readNextBytes);
+            }
+
+            return $scalar;
         }
 
         $offset = $this->binaryDataReader->readUIntBySize($intSize);
@@ -371,4 +386,4 @@ private function ensureOffset(?int $ensureOffset): void
             $this->binaryDataReader->advance($ensureOffset + 1 - $pos);
         }
     }
-}
\ No newline at end of file
+}

From 95f852a5d6caddbccea5caec0790b70b9040a127 Mon Sep 17 00:00:00 2001
From: Matt <matt@fasterwebsite.com>
Date: Fri, 12 Jan 2024 09:46:31 +0100
Subject: [PATCH 2/2] Preserve threadId in QUERY_EVENT. (#90)

---
 src/MySQLReplication/Event/DTO/QueryDTO.php | 10 +++++++++-
 src/MySQLReplication/Event/QueryEvent.php   |  5 +++--
 2 files changed, 12 insertions(+), 3 deletions(-)

diff --git a/src/MySQLReplication/Event/DTO/QueryDTO.php b/src/MySQLReplication/Event/DTO/QueryDTO.php
index 5c215b2..1ec3604 100644
--- a/src/MySQLReplication/Event/DTO/QueryDTO.php
+++ b/src/MySQLReplication/Event/DTO/QueryDTO.php
@@ -12,18 +12,21 @@ class QueryDTO extends EventDTO
     private $query;
     private $database;
     private $type = ConstEventsNames::QUERY;
+    private $threadId;
 
     public function __construct(
         EventInfo $eventInfo,
         string $database,
         int $executionTime,
-        string $query
+        string $query,
+        int $threadId
     ) {
         parent::__construct($eventInfo);
 
         $this->executionTime = $executionTime;
         $this->query = $query;
         $this->database = $database;
+        $this->threadId = $threadId;
     }
 
     public function getDatabase(): string
@@ -46,6 +49,11 @@ public function getType(): string
         return $this->type;
     }
 
+    public function getThreadId(): int
+    {
+        return $this->threadId;
+    }
+
     public function __toString(): string
     {
         return PHP_EOL .
diff --git a/src/MySQLReplication/Event/QueryEvent.php b/src/MySQLReplication/Event/QueryEvent.php
index 48e2b7d..88f6b31 100644
--- a/src/MySQLReplication/Event/QueryEvent.php
+++ b/src/MySQLReplication/Event/QueryEvent.php
@@ -12,7 +12,7 @@ class QueryEvent extends EventCommon
 {
     public function makeQueryDTO(): QueryDTO
     {
-        $this->binaryDataReader->advance(4);
+        $threadId = $this->binaryDataReader->readUInt32();
         $executionTime = $this->binaryDataReader->readUInt32();
         $schemaLength = $this->binaryDataReader->readUInt8();
         $this->binaryDataReader->advance(2);
@@ -26,7 +26,8 @@ public function makeQueryDTO(): QueryDTO
             $this->eventInfo,
             $schema,
             $executionTime,
-            $query
+            $query,
+            $threadId
         );
     }
 }
\ No newline at end of file