VStream: mysql protocol

go/vt/discovery/healthcheck.go

@@ -572,6 +572,9 @@ func (hc *HealthCheckImpl) GetHealthyTabletStats(target *query.Target) []*Tablet
  var result []*TabletHealth
  hc.mu.Lock()
  defer hc.mu.Unlock()
+ if target.Shard == "" {
+ target.Shard = "0"
+ }
  return append(result, hc.healthy[hc.keyFromTarget(target)]...)
 }
 

go/vt/sqlparser/analyzer.go

@@ -56,6 +56,7 @@ const (
  StmtSavepoint
  StmtSRollback
  StmtRelease
+ StmtVStream
 )
 
 //ASTToStatementType returns a StatementType from an AST stmt
@@ -138,6 +139,8 @@ func Preview(sql string) StatementType {
  return StmtSelect
  case "stream":
  return StmtStream
+ case "vstream":
+ return StmtVStream
  case "insert":
  return StmtInsert
  case "replace":
@@ -192,6 +195,8 @@ func (s StatementType) String() string {
  return "SELECT"
  case StmtStream:
  return "STREAM"
+ case StmtVStream:
+ return "VSTREAM"
  case StmtInsert:
  return "INSERT"
  case StmtReplace:

go/vt/sqlparser/ast.go

@@ -86,6 +86,15 @@ type (
  Lock string
  }
 
+ // VStream represents a VSTREAM statement.
+ VStream struct {
+ Comments Comments
+ SelectExpr SelectExpr
+ Table TableName
+ Where *Where
+ Limit *Limit
+ }
+
  // Stream represents a SELECT statement.
  Stream struct {
  Comments Comments
@@ -270,6 +279,7 @@ type (
 func (*Union) iStatement() {}
 func (*Select) iStatement() {}
 func (*Stream) iStatement() {}
+func (*VStream) iStatement() {}
 func (*Insert) iStatement() {}
 func (*Update) iStatement() {}
 func (*Delete) iStatement() {}
@@ -921,6 +931,12 @@ func (node *UnionSelect) Format(buf *TrackedBuffer) {
  buf.astPrintf(node, " %s %v", node.Type, node.Statement)
 }
 
+// Format formats the node.
+func (node *VStream) Format(buf *TrackedBuffer) {
+ buf.astPrintf(node, "vstream %v%v from %v",
+ node.Comments, node.SelectExpr, node.Table)
+}
+
 // Format formats the node.
 func (node *Stream) Format(buf *TrackedBuffer) {
  buf.astPrintf(node, "stream %v%v from %v",

go/vt/sqlparser/parse_test.go

@@ -1588,6 +1588,8 @@ var (
  input: "delete from t partition (p0) where a = 1",
  }, {
  input: "stream * from t",
+ }, {
+ input: "vstream * from t",
  }, {
  input: "stream /* comment */ * from t",
  }, {

go/vt/sqlparser/parser.go

@@ -20,6 +20,7 @@ import (
  "errors"
  "fmt"
  "io"
+ "runtime/debug"
  "sync"
 
  "vitess.io/vitess/go/vt/log"
@@ -93,6 +94,7 @@ func Parse(sql string) (Statement, error) {
  return nil, vterrors.New(vtrpcpb.Code_INVALID_ARGUMENT, tokenizer.LastError.Error())
  }
  if tokenizer.ParseTree == nil {
+ log.Infof("Empty Statement: %s", debug.Stack())
  return nil, ErrEmpty
  }
  return tokenizer.ParseTree, nil
@@ -106,6 +108,8 @@ func ParseStrictDDL(sql string) (Statement, error) {
  return nil, tokenizer.LastError
  }
  if tokenizer.ParseTree == nil {
+ log.Infof("Empty Statement DDL: %s", debug.Stack())
+
  return nil, ErrEmpty
  }
  return tokenizer.ParseTree, nil