Implement zlib compression (#1487)

Implemented the SQL compression protocol. This new feature is enabled by: * Adding `compress=true` in DSN. * `cfg.Apply(Compress(True))` Co-authored-by: Brigitte Lamarche <[email protected]> Co-authored-by: Julien Schmidt <[email protected]> Co-authored-by: Jeffrey Charles <[email protected]> Co-authored-by: Jeff Hodges <[email protected]> Co-authored-by: Daniel Montoya <[email protected]> Co-authored-by: Justin Li <[email protected]> Co-authored-by: Dave Stubbs <[email protected]> Co-authored-by: Linh Tran Tuan <[email protected]> Co-authored-by: Robert R. Russell <[email protected]> Co-authored-by: INADA Naoki <[email protected]> Co-authored-by: Kieron Woodhouse <[email protected]> Co-authored-by: Alexey Palazhchenko <[email protected]> Co-authored-by: Reed Allman <[email protected]> Co-authored-by: Joe Mann <[email protected]>

Author: 15 people

.github/workflows/test.yml

@@ -83,7 +83,7 @@ jobs:
 my-cnf: |
  innodb_log_file_size=256MB
  innodb_buffer_pool_size=512MB
- max_allowed_packet=16MB
+ max_allowed_packet=48MB
  ; TestConcurrent fails if max_connections is too large
  max_connections=50
  local_infile=1

AUTHORS

@@ -21,6 +21,7 @@ Animesh Ray <mail.rayanimesh at gmail.com>
 Arne Hormann <arnehormann at gmail.com>
 Ariel Mashraki <ariel at mashraki.co.il>
 Asta Xie <xiemengjun at gmail.com>
+B Lamarche <blam413 at gmail.com>
 Brian Hendriks <brian at dolthub.com>
 Bulat Gaifullin <gaifullinbf at gmail.com>
 Caine Jette <jette at alum.mit.edu>
@@ -62,6 +63,7 @@ Jennifer Purevsuren <jennifer at dolthub.com>
 Jerome Meyer <jxmeyer at gmail.com>
 Jiajia Zhong <zhong2plus at gmail.com>
 Jian Zhen <zhenjl at gmail.com>
+Joe Mann <contact at joemann.co.uk>
 Joshua Prunier <joshua.prunier at gmail.com>
 Julien Lefevre <julien.lefevr at gmail.com>
 Julien Schmidt <go-sql-driver at julienschmidt.com>

README.md

@@ -38,6 +38,7 @@ A MySQL-Driver for Go's [database/sql](https://golang.org/pkg/database/sql/) pac
 * Secure `LOAD DATA LOCAL INFILE` support with file allowlisting and `io.Reader` support
 * Optional `time.Time` parsing
 * Optional placeholder interpolation
+* Supports zlib compression.
 
 ## Requirements
 
@@ -267,6 +268,16 @@ SELECT u.id FROM users as u
 
 will return `u.id` instead of just `id` if `columnsWithAlias=true`.
 
+##### `compress`
+
+```
+Type: bool
+Valid Values: true, false
+Default: false
+```
+
+Toggles zlib compression. false by default.
+
 ##### `interpolateParams`
 
 ```

benchmark_test.go

@@ -46,9 +46,13 @@ func (tb *TB) checkStmt(stmt *sql.Stmt, err error) *sql.Stmt{
 returnstmt
 }
 
-funcinitDB(b*testing.B, queries...string) *sql.DB{
+funcinitDB(b*testing.B, useCompressionbool, queries...string) *sql.DB{
 tb:= (*TB)(b)
-db:=tb.checkDB(sql.Open(driverNameTest, dsn))
+comprStr:=""
+ifuseCompression{
+comprStr="&compress=1"
+ }
+db:=tb.checkDB(sql.Open(driverNameTest, dsn+comprStr))
 for_, query:=rangequeries{
 if_, err:=db.Exec(query); err!=nil{
 b.Fatalf("error on %q: %v", query, err)
@@ -60,10 +64,18 @@ func initDB(b *testing.B, queries ...string) *sql.DB{
 constconcurrencyLevel=10
 
 funcBenchmarkQuery(b*testing.B){
+benchmarkQueryHelper(b, false)
+}
+
+funcBenchmarkQueryCompression(b*testing.B){
+benchmarkQueryHelper(b, true)
+}
+
+funcbenchmarkQueryHelper(b*testing.B, comprbool){
 tb:= (*TB)(b)
 b.StopTimer()
 b.ReportAllocs()
-db:=initDB(b,
+db:=initDB(b,compr,
 "DROP TABLE IF EXISTS foo",
 "CREATE TABLE foo (id INT PRIMARY KEY, val CHAR(50))",
 `INSERT INTO foo VALUES (1, "one")`,
@@ -222,7 +234,7 @@ func BenchmarkInterpolation(b *testing.B){
  },
 maxAllowedPacket: maxPacketSize,
 maxWriteSize: maxPacketSize-1,
-buf: newBuffer(nil),
+buf: newBuffer(),
  }
 
 args:= []driver.Value{
@@ -269,7 +281,7 @@ func benchmarkQueryContext(b *testing.B, db *sql.DB, p int){
 }
 
 funcBenchmarkQueryContext(b*testing.B){
-db:=initDB(b,
+db:=initDB(b,false,
 "DROP TABLE IF EXISTS foo",
 "CREATE TABLE foo (id INT PRIMARY KEY, val CHAR(50))",
 `INSERT INTO foo VALUES (1, "one")`,
@@ -305,7 +317,7 @@ func benchmarkExecContext(b *testing.B, db *sql.DB, p int){
 }
 
 funcBenchmarkExecContext(b*testing.B){
-db:=initDB(b,
+db:=initDB(b,false,
 "DROP TABLE IF EXISTS foo",
 "CREATE TABLE foo (id INT PRIMARY KEY, val CHAR(50))",
 `INSERT INTO foo VALUES (1, "one")`,
@@ -323,7 +335,7 @@ func BenchmarkExecContext(b *testing.B){
 // "size=" means size of each blobs.
 funcBenchmarkQueryRawBytes(b*testing.B){
 varsizes []int= []int{100, 1000, 2000, 4000, 8000, 12000, 16000, 32000, 64000, 256000}
-db:=initDB(b,
+db:=initDB(b,false,
 "DROP TABLE IF EXISTS bench_rawbytes",
 "CREATE TABLE bench_rawbytes (id INT PRIMARY KEY, val LONGBLOB)",
  )
@@ -376,7 +388,7 @@ func BenchmarkQueryRawBytes(b *testing.B){
 // BenchmarkReceiveMassiveRows measures performance of receiving large number of rows.
 funcBenchmarkReceiveMassiveRows(b*testing.B){
 // Setup -- prepare 10000 rows.
-db:=initDB(b,
+db:=initDB(b,false,
 "DROP TABLE IF EXISTS foo",
 "CREATE TABLE foo (id INT PRIMARY KEY, val TEXT)")
 deferdb.Close()

buffer.go

@@ -10,13 +10,16 @@ package mysql
 
 import (
 "io"
-"net"
-"time"
 )
 
 constdefaultBufSize=4096
 constmaxCachedBufSize=256*1024
 
+// readerFunc is a function that compatible with io.Reader.
+// We use this function type instead of io.Reader because we want to
+// just pass mc.readWithTimeout.
+typereaderFuncfunc([]byte) (int, error)
+
 // A buffer which is used for both reading and writing.
 // This is possible since communication on each connection is synchronous.
 // In other words, we can't write and read simultaneously on the same connection.
@@ -25,15 +28,12 @@ const maxCachedBufSize = 256 * 1024
 typebufferstruct{
 buf []byte// read buffer.
 cachedBuf []byte// buffer that will be reused. len(cachedBuf) <= maxCachedBufSize.
-nc net.Conn
-timeout time.Duration
 }
 
 // newBuffer allocates and returns a new buffer.
-funcnewBuffer(nc net.Conn) buffer{
+funcnewBuffer() buffer{
 returnbuffer{
 cachedBuf: make([]byte, defaultBufSize),
-nc: nc,
  }
 }
 
@@ -43,7 +43,7 @@ func (b *buffer) busy() bool{
 }
 
 // fill reads into the read buffer until at least _need_ bytes are in it.
-func (b*buffer) fill(needint) error{
+func (b*buffer) fill(needint, rreaderFunc) error{
 // we'll move the contents of the current buffer to dest before filling it.
 dest:=b.cachedBuf
 
@@ -64,13 +64,7 @@ func (b *buffer) fill(need int) error{
 copy(dest[:n], b.buf)
 
 for{
-ifb.timeout>0{
-iferr:=b.nc.SetReadDeadline(time.Now().Add(b.timeout)); err!=nil{
-returnerr
- }
- }
-
-nn, err:=b.nc.Read(dest[n:])
+nn, err:=r(dest[n:])
 n+=nn
 
 iferr==nil&&n<need{
@@ -92,10 +86,10 @@ func (b *buffer) fill(need int) error{
 
 // returns next N bytes from buffer.
 // The returned slice is only guaranteed to be valid until the next read
-func (b*buffer) readNext(needint) ([]byte, error){
+func (b*buffer) readNext(needint, rreaderFunc) ([]byte, error){
 iflen(b.buf) <need{
 // refill
-iferr:=b.fill(need); err!=nil{
+iferr:=b.fill(need, r); err!=nil{
 returnnil, err
  }
  }