finagle/finagle-mysql: Add support for LONG_BLOB data type

Problem

finagle-mysql does not support the LONG_BLOB data type. When it is encountered,
an UnsupportedOperationException is thrown.

Solution

Support it.

Differential Revision: https://phabricator.twitter.biz/D1152247

CHANGELOG.rst

@@ -13,6 +13,12 @@ Runtime Behavior Changes
 * finagle-mysql: (Testing behaviour change only) Updated mysql version expected by integration tests to 8.0.21.
   Added README in integration tests noting that this must exist for integration tests to run. ``PHAB_ID=D1152235``
 
+
+New Features
+~~~~~~~~~~
+
+* finagle-mysql: Added support for LONG_BLOB data type. ``PHAB_ID=D1152247``
+
 24.5.0
 ------
 

finagle-mysql/src/main/scala/com/twitter/finagle/mysql/BinaryEncodedRow.scala

@@ -1,6 +1,7 @@
 package com.twitter.finagle.mysql
 
-import com.twitter.finagle.mysql.transport.{MysqlBuf, MysqlBufReader}
+import com.twitter.finagle.mysql.transport.MysqlBuf
+import com.twitter.finagle.mysql.transport.MysqlBufReader
 import com.twitter.io.Buf
 
 /**
@@ -39,7 +40,7 @@ private class BinaryEncodedRow(
    * Convert the binary representation of each value
    * into an appropriate Value object.
    *
-   * @see [[https://mariadb.com/kb/en/mariadb/resultset-row/]] for details
+   * @see [[https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_binary_resultset.html]] for details
    *     about the binary row format.
    */
   lazy val values: IndexedSeq[Value] = {
@@ -69,14 +70,11 @@ private class BinaryEncodedRow(
             case Type.Year => ShortValue(reader.readShortLE())
             // Nonbinary strings as stored in the CHAR, VARCHAR, and TEXT data types
             case Type.VarChar | Type.String | Type.VarString | Type.TinyBlob | Type.Blob |
-                Type.MediumBlob
+                Type.MediumBlob | Type.LongBlob
                 if !MysqlCharset.isBinary(field.charset) && MysqlCharset.isCompatible(
                   field.charset
                 ) =>
               StringValue(reader.readLengthCodedString(MysqlCharset(field.charset)))
-
-            case Type.LongBlob =>
-              throw new UnsupportedOperationException("LongBlob is not supported!")
             case typ => RawValue(typ, field.charset, isBinary = true, reader.readLengthCodedBytes())
           }
         }

finagle-mysql/src/main/scala/com/twitter/finagle/mysql/StringEncodedRow.scala

@@ -18,7 +18,7 @@ private class StringEncodedRow(
   /**
    * Convert the string representation of each value
    * into an appropriate Value object.
-   * [[https://dev.mysql.com/doc/internals/en/com-query-response.html#packet-ProtocolText::ResultsetRow]]
+   * [[https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_com_query_response_text_resultset.html]]
    */
   lazy val values: IndexedSeq[Value] = {
     val reader = MysqlBuf.reader(rawRow)
@@ -59,14 +59,9 @@ private class StringEncodedRow(
           case Type.Year =>
             ShortValue(bytesToLong(bytes).toShort)
           case Type.VarChar | Type.String | Type.VarString | Type.TinyBlob | Type.Blob |
-              Type.MediumBlob if !MysqlCharset.isBinary(charset) =>
+              Type.MediumBlob | Type.LongBlob if !MysqlCharset.isBinary(charset) =>
             // Nonbinary strings as stored in the CHAR, VARCHAR, and TEXT data types
             StringValue(bytesToString(bytes, charset))
-          case Type.LongBlob =>
-            // LongBlobs indicate a sequence of bytes with length >= 2^24 which
-            // can't fit into a Array[Byte]. This should be streamed and
-            // support for this needs to begin at the transport layer.
-            throw new UnsupportedOperationException("LongBlob is not supported!")
           case typ =>
             RawValue(typ, charset, isBinary = false, bytes)
         }

finagle-mysql/src/test/scala/com/twitter/finagle/mysql/integration/TypeTest.scala

@@ -311,7 +311,7 @@ object BlobTypeTest {
         VALUES (1, 'a', 'b', 'c', 'd', 'e', X'66',
         X'67', X'68', X'6970', X'6A', 'small', '1');"""
 
-  val sqlQuery: String = "SELECT * FROM `blobs`"
+  val selectRowQuery: String = "SELECT * FROM `blobs`"
 }
 
 class BlobTypeTest extends EmbeddedSimpleSuite {
@@ -325,21 +325,40 @@ class BlobTypeTest extends EmbeddedSimpleSuite {
     // Setup temporary table and insert into it
     await(client.query(createTableQuery))
     await(client.query(insertQuery))
-    val textEncoded = getTextEncodedRow(client)
-    val binaryEncoded = getBinaryEncodedRow(client)
+    val textEncoded = getTextEncodedRow(client, selectRowQuery)
+    val binaryEncoded = getBinaryEncodedRow(client, selectRowQuery)
 
     testRow(textEncoded)
     testRow(binaryEncoded)
+
+    test("extract text from longblob") {
+      // Note -- in mysql 8 group_concat will return a long_blob. If we add a long_blob column,
+      // we'll get back a string in binary when we select, which doesn't exercise the long_blob code
+      // path, so don't test that.
+      val textExcodedRow = getTextEncodedRow(client, "select group_concat('1', '2', '3')")
+      assert(textExcodedRow.fields.map(_.fieldType) == Seq(Type.LongBlob))
+      assert(
+        textExcodedRow.values == Seq(
+          StringValue("123")
+        ))
+
+      val binaryEncodedRow = getBinaryEncodedRow(client, "select group_concat('1', '2', '3')")
+      assert(binaryEncodedRow.fields.map(_.fieldType) == Seq(Type.LongBlob))
+      assert(
+        binaryEncodedRow.values == Seq(
+          StringValue("123")
+        ))
+    }
   })
 
-  def getTextEncodedRow(client: Client with Transactions): Row =
-    await(client.query(sqlQuery) map {
+  def getTextEncodedRow(client: Client with Transactions, query: String): Row =
+    await(client.query(query) map {
       case rs: ResultSet if rs.rows.nonEmpty => rs.rows.head
       case v => fail("expected a ResultSet with 1 row but received: %s".format(v))
     })
 
-  def getBinaryEncodedRow(client: Client with Transactions): Row = {
-    val ps = client.prepare(sqlQuery)
+  def getBinaryEncodedRow(client: Client with Transactions, query: String): Row = {
+    val ps = client.prepare(query)
     val binaryrows: Seq[Row] = await(ps.select()(identity))
     assert(binaryrows.size == 1)
     binaryrows.head