Updating from upstream

data/test/vtgate/dml_cases.txt

@@ -1512,3 +1512,24 @@
     "Query": "delete from unsharded where col = (select id from unsharded_a where id = unsharded.col)"
   }
 }
+
+# update vindex value to null
+"update user set name = null where id = 1"
+{
+  "Original": "update user set name = null where id = 1",
+  "Instructions": {
+    "Opcode": "UpdateEqual",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "Query": "update user set name = null where id = 1",
+    "Vindex": "user_index",
+    "Values": [1],
+    "ChangedVindexValues": {
+      "name_user_map": [null]
+    },
+    "Table": "user",
+    "OwnedVindexQuery": "select Name, Costly from user where id = 1 for update"
+  }
+}

doc/GettingStartedKubernetes.md

@@ -278,7 +278,7 @@ $ export KUBECTL=/example/path/to/google-cloud-sdk/bin/kubectl
 1.  **Access vtctld web UI**
 
     To access vtctld from outside Kubernetes, use [kubectl proxy]
-    (http://kubernetes.io/v1.1/docs/user-guide/kubectl/kubectl_proxy.html)
+    (https://kubernetes.io/docs/tasks/access-kubernetes-api/http-proxy-access-api/)
     to create an authenticated tunnel on your workstation:
 
     **Note:** The proxy command runs in the foreground,
@@ -292,13 +292,13 @@ $ export KUBECTL=/example/path/to/google-cloud-sdk/bin/kubectl
 
     You can then load the vtctld web UI on `localhost`:
 
-    http://localhost:8001/api/v1/proxy/namespaces/default/services/vtctld:web/
+    http://localhost:8001/api/v1/namespaces/default/services/vtctld:web/proxy
 
     You can also use this proxy to access the [Kubernetes Dashboard]
-    (http://kubernetes.io/v1.1/docs/user-guide/ui.html),
+    (https://kubernetes.io/docs/tasks/access-application-cluster/web-ui-dashboard/),
     where you can monitor nodes, pods, and services:
 
-    http://localhost:8001/ui
+    http://localhost:8001/api/v1/namespaces/kube-system/services/https:kubernetes-dashboard:/proxy/.
 
 1.  **Use vtctlclient to send commands to vtctld**
 

examples/kubernetes/vtctld-up.sh

@@ -47,5 +47,4 @@ cat vtctld-controller-template.yaml | sed -e "$sed_script" | $KUBECTL $KUBECTL_O
 echo
 echo "To access vtctld web UI, start kubectl proxy in another terminal:"
 echo "  kubectl proxy --port=8001"
-echo "Then visit http://localhost:8001/api/v1/proxy/namespaces/$VITESS_NAME/services/vtctld:web/"
-
+echo "Then visit http://localhost:8001/api/v1/namespaces/$VITESS_NAME/services/vtctld:web/proxy"

go/bucketpool/bucketpool.go

@@ -0,0 +1,91 @@
+package bucketpool
+
+import (
+	"math"
+	"sync"
+)
+
+type sizedPool struct {
+	size int
+	pool sync.Pool
+}
+
+func newSizedPool(size int) *sizedPool {
+	return &sizedPool{
+		size: size,
+		pool: sync.Pool{
+			New: func() interface{} { return makeSlicePointer(size) },
+		},
+	}
+}
+
+// Pool is actually multiple pools which store buffers of specific size.
+// i.e. it can be three pools which return buffers 32K, 64K and 128K.
+type Pool struct {
+	minSize int
+	maxSize int
+	pools   []*sizedPool
+}
+
+// New returns Pool which has buckets from minSize to maxSize.
+// Buckets increase with the power of two, i.e with multiplier 2: [2b, 4b, 16b, ... , 1024b]
+// Last pool will always be capped to maxSize.
+func New(minSize, maxSize int) *Pool {
+	if maxSize < minSize {
+		panic("maxSize can't be less than minSize")
+	}
+	const multiplier = 2
+	var pools []*sizedPool
+	curSize := minSize
+	for curSize < maxSize {
+		pools = append(pools, newSizedPool(curSize))
+		curSize *= multiplier
+	}
+	pools = append(pools, newSizedPool(maxSize))
+	return &Pool{
+		minSize: minSize,
+		maxSize: maxSize,
+		pools:   pools,
+	}
+}
+
+func (p *Pool) findPool(size int) *sizedPool {
+	if size > p.maxSize {
+		return nil
+	}
+	idx := int(math.Ceil(math.Log2(float64(size) / float64(p.minSize))))
+	if idx < 0 {
+		idx = 0
+	}
+	if idx > len(p.pools)-1 {
+		return nil
+	}
+	return p.pools[idx]
+}
+
+// Get returns pointer to []byte which has len size.
+// If there is no bucket with buffers >= size, slice will be allocated.
+func (p *Pool) Get(size int) *[]byte {
+	sp := p.findPool(size)
+	if sp == nil {
+		return makeSlicePointer(size)
+	}
+	buf := sp.pool.Get().(*[]byte)
+	*buf = (*buf)[:size]
+	return buf
+}
+
+// Put returns pointer to slice to some bucket. Discards slice for which there is no bucket
+func (p *Pool) Put(b *[]byte) {
+	sp := p.findPool(cap(*b))
+	if sp == nil {
+		return
+	}
+	*b = (*b)[:cap(*b)]
+	sp.pool.Put(b)
+}
+
+func makeSlicePointer(size int) *[]byte {
+	data := make([]byte, size)
+	return &data
+}

go/bucketpool/bucketpool_test.go

@@ -0,0 +1,202 @@
+package bucketpool
+
+import (
+	"math/rand"
+	"testing"
+)
+
+func TestPool(t *testing.T) {
+	maxSize := 16384
+	pool := New(1024, maxSize)
+	if pool.maxSize != maxSize {
+		t.Fatalf("Invalid max pool size: %d, expected %d", pool.maxSize, maxSize)
+	}
+	if len(pool.pools) != 5 {
+		t.Fatalf("Invalid number of pools: %d, expected %d", len(pool.pools), 5)
+	}
+
+	buf := pool.Get(64)
+	if len(*buf) != 64 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 1024 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+
+	// get from same pool, check that length is right
+	buf = pool.Get(128)
+	if len(*buf) != 128 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 1024 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+
+	// get boundary size
+	buf = pool.Get(1024)
+	if len(*buf) != 1024 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 1024 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+
+	// get from the middle
+	buf = pool.Get(5000)
+	if len(*buf) != 5000 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 8192 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+
+	// check last pool
+	buf = pool.Get(16383)
+	if len(*buf) != 16383 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 16384 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+
+	// get big buffer
+	buf = pool.Get(16385)
+	if len(*buf) != 16385 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 16385 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+}
+
+func TestPoolOneSize(t *testing.T) {
+	maxSize := 1024
+	pool := New(1024, maxSize)
+	if pool.maxSize != maxSize {
+		t.Fatalf("Invalid max pool size: %d, expected %d", pool.maxSize, maxSize)
+	}
+	buf := pool.Get(64)
+	if len(*buf) != 64 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 1024 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+
+	buf = pool.Get(1025)
+	if len(*buf) != 1025 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 1025 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+}
+
+func TestPoolTwoSizeNotMultiplier(t *testing.T) {
+	maxSize := 2000
+	pool := New(1024, maxSize)
+	if pool.maxSize != maxSize {
+		t.Fatalf("Invalid max pool size: %d, expected %d", pool.maxSize, maxSize)
+	}
+	buf := pool.Get(64)
+	if len(*buf) != 64 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 1024 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+
+	buf = pool.Get(2001)
+	if len(*buf) != 2001 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 2001 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+}
+
+func TestPoolWeirdMaxSize(t *testing.T) {
+	maxSize := 15000
+	pool := New(1024, maxSize)
+	if pool.maxSize != maxSize {
+		t.Fatalf("Invalid max pool size: %d, expected %d", pool.maxSize, maxSize)
+	}
+
+	buf := pool.Get(14000)
+	if len(*buf) != 14000 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 15000 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+
+	buf = pool.Get(16383)
+	if len(*buf) != 16383 {
+		t.Fatalf("unexpected buf length: %d", len(*buf))
+	}
+	if cap(*buf) != 16383 {
+		t.Fatalf("unexepected buf cap: %d", cap(*buf))
+	}
+	pool.Put(buf)
+}
+
+func TestFuzz(t *testing.T) {
+	maxTestSize := 16384
+	for i := 0; i < 20000; i++ {
+		minSize := rand.Intn(maxTestSize)
+		maxSize := rand.Intn(maxTestSize-minSize) + minSize
+		p := New(minSize, maxSize)
+		bufSize := rand.Intn(maxTestSize)
+		buf := p.Get(bufSize)
+		if len(*buf) != bufSize {
+			t.Fatalf("Invalid length %d, expected %d", len(*buf), bufSize)
+		}
+		sPool := p.findPool(bufSize)
+		if sPool == nil {
+			if cap(*buf) != len(*buf) {
+				t.Fatalf("Invalid cap %d, expected %d", cap(*buf), len(*buf))
+			}
+		} else {
+			if cap(*buf) != sPool.size {
+				t.Fatalf("Invalid cap %d, expected %d", cap(*buf), sPool.size)
+			}
+		}
+		p.Put(buf)
+	}
+}
+
+func BenchmarkPool(b *testing.B) {
+	pool := New(2, 16384)
+	b.SetParallelism(16)
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			randomSize := rand.Intn(pool.maxSize)
+			data := pool.Get(randomSize)
+			pool.Put(data)
+		}
+	})
+}
+
+func BenchmarkPoolGet(b *testing.B) {
+	pool := New(2, 16384)
+	b.SetParallelism(16)
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			randomSize := rand.Intn(pool.maxSize)
+			data := pool.Get(randomSize)
+			_ = data
+		}
+	})
+}

go/mysql/client.go

@@ -250,8 +250,7 @@ func (c *Conn) clientHandshake(characterSet uint8, params *ConnParams) error {
 		// Switch to SSL.
 		conn := tls.Client(c.conn, clientConfig)
 		c.conn = conn
-		c.reader.Reset(conn)
-		c.writer.Reset(conn)
+		c.bufferedReader.Reset(conn)
 		c.Capabilities |= CapabilityClientSSL
 	}
 
@@ -508,7 +507,7 @@ func (c *Conn) writeSSLRequest(capabilities uint32, characterSet uint8, params *
 	pos = writeByte(data, pos, characterSet)
 
 	// And send it as is.
-	if err := c.writeEphemeralPacket(true /* direct */); err != nil {
+	if err := c.writeEphemeralPacket(); err != nil {
 		return NewSQLError(CRServerLost, SSUnknownSQLState, "cannot send SSLRequest: %v", err)
 	}
 	return nil
@@ -600,7 +599,7 @@ func (c *Conn) writeHandshakeResponse41(capabilities uint32, scrambledPassword [
 		return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "writeHandshakeResponse41: only packed %v bytes, out of %v allocated", pos, len(data))
 	}
 
-	if err := c.writeEphemeralPacket(true /* direct */); err != nil {
+	if err := c.writeEphemeralPacket(); err != nil {
 		return NewSQLError(CRServerLost, SSUnknownSQLState, "cannot send HandshakeResponse41: %v", err)
 	}
 	return nil
@@ -627,5 +626,5 @@ func (c *Conn) writeClearTextPassword(params *ConnParams) error {
 	if pos != len(data) {
 		return fmt.Errorf("error building ClearTextPassword packet: got %v bytes expected %v", pos, len(data))
 	}
-	return c.writeEphemeralPacket(true)
+	return c.writeEphemeralPacket()
 }