From 54d3a68403e6366f0d1d993db41f8ca3d3d234ff Mon Sep 17 00:00:00 2001
From: Karel Alfonso Sague <kalfonso@squareup.com>
Date: Thu, 29 Aug 2019 16:24:42 +1000
Subject: [PATCH] Delete vitess metadata by setting empty key

Signed-off-by: Karel Alfonso Sague <kalfonso@squareup.com>
---
 go/vt/topo/metadata.go        | 14 ++++++++++++++
 go/vt/vtgate/executor.go      | 10 ++++++++--
 go/vt/vtgate/executor_test.go | 34 ++++++++++++++++++++++++++++++++++
 3 files changed, 56 insertions(+), 2 deletions(-)

diff --git a/go/vt/topo/metadata.go b/go/vt/topo/metadata.go
index 349ea7f8084..ab235da2719 100644
--- a/go/vt/topo/metadata.go
+++ b/go/vt/topo/metadata.go
@@ -19,6 +19,7 @@ package topo
 import (
 	"context"
 	"path"
+
 	"vitess.io/vitess/go/vt/sqlparser"
 
 	"vitess.io/vitess/go/event"
@@ -74,6 +75,19 @@ func (ts *Server) GetMetadata(ctx context.Context, keyFilter string) (map[string
 	return result, nil
 }
 
+// DeleteMetadata deletes the key in the metadata
+func (ts *Server) DeleteMetadata(ctx context.Context, key string) error {
+	keyPath := path.Join(MetadataPath, key)
+
+	// nil version means that it will insert if keyPath does not exist
+	if err := ts.globalCell.Delete(ctx, keyPath, nil); err != nil {
+		return err
+	}
+
+	dispatchEvent(keyPath, "deleted")
+	return nil
+}
+
 func (ts *Server) getMetadata(ctx context.Context, key string) (string, error) {
 	keyPath := path.Join(MetadataPath, key)
 	contents, _, err := ts.globalCell.Get(ctx, keyPath)
diff --git a/go/vt/vtgate/executor.go b/go/vt/vtgate/executor.go
index a21476e6b35..e1bc9ee252c 100644
--- a/go/vt/vtgate/executor.go
+++ b/go/vt/vtgate/executor.go
@@ -667,11 +667,17 @@ func (e *Executor) handleSetVitessMetadata(ctx context.Context, session *SafeSes
 		return nil, err
 	}
 
-	if err := ts.UpsertMetadata(ctx, k.Key, val); err != nil {
+	if val == "" {
+		err = ts.DeleteMetadata(ctx, k.Key)
+	} else {
+		err = ts.UpsertMetadata(ctx, k.Key, val)
+	}
+
+	if err != nil {
 		return nil, err
 	}
 
-	return &sqltypes.Result{}, nil
+	return &sqltypes.Result{RowsAffected: 1}, nil
 }
 
 func (e *Executor) handleShowVitessMetadata(ctx context.Context, session *SafeSession, opt *sqlparser.ShowTablesOpt) (*sqltypes.Result, error) {
diff --git a/go/vt/vtgate/executor_test.go b/go/vt/vtgate/executor_test.go
index 54aa1ac7b8c..bbb158bfebb 100644
--- a/go/vt/vtgate/executor_test.go
+++ b/go/vt/vtgate/executor_test.go
@@ -28,6 +28,8 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/vterrors"
 
 	"context"
@@ -531,6 +533,38 @@ func TestExecutorSetMetadata(t *testing.T) {
 	assert.ElementsMatch(t, wantqr.Rows, gotqr.Rows)
 }
 
+func TestExecutorDeleteMetadata(t *testing.T) {
+	*vschemaacl.AuthorizedDDLUsers = "%"
+	defer func() {
+		*vschemaacl.AuthorizedDDLUsers = ""
+	}()
+
+	executor, _, _, _ := createExecutorEnv()
+	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master", Autocommit: true})
+
+	set := "set @@vitess_metadata.app_v1= '1'"
+	_, err := executor.Execute(context.Background(), "TestExecute", session, set, nil)
+	assert.NoError(t, err, "%s error: %v", set, err)
+
+	show := `show vitess_metadata variables like 'app\\_%'`
+	result, _ := executor.Execute(context.Background(), "TestExecute", session, show, nil)
+	assert.Len(t, result.Rows, 1)
+
+	// Fails if deleting key that doesn't exist
+	delete := "set @@vitess_metadata.doesnt_exist=''"
+	_, err = executor.Execute(context.Background(), "TestExecute", session, delete, nil)
+	assert.True(t, topo.IsErrType(err, topo.NoNode))
+
+	// Delete existing key, show should fail given the node doesn't exist
+	delete = "set @@vitess_metadata.app_v1=''"
+	_, err = executor.Execute(context.Background(), "TestExecute", session, delete, nil)
+	assert.NoError(t, err)
+
+	show = `show vitess_metadata variables like 'app\\_%'`
+	result, err = executor.Execute(context.Background(), "TestExecute", session, show, nil)
+	assert.True(t, topo.IsErrType(err, topo.NoNode))
+}
+
 func TestExecutorAutocommit(t *testing.T) {
 	executor, _, _, sbclookup := createExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master"})