hackathon init

domain/domain.go

@@ -661,7 +661,7 @@ func (do *Domain) Init(ddlLease time.Duration, sysFactory func(*Domain) (pools.R
 	if err != nil {
 		return err
 	}
-	do.info, err = infosync.GlobalInfoSyncerInit(ctx, do.ddl.GetID(), do.etcdClient)
+	do.info, err = infosync.GlobalInfoSyncerInit(ctx, do.ddl.GetID(), do.etcdClient, do.store)
 	if err != nil {
 		return err
 	}

domain/infosync/info.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/pingcap/tidb/meta"
 	"strconv"
 	"time"
 
@@ -42,6 +43,8 @@ const (
 	ServerInformationPath = "/tidb/server/info"
 	// ServerMinStartTSPath store the server min start timestamp.
 	ServerMinStartTSPath = "/tidb/server/minstartts"
+
+	ServerGenIDPath = "/tidb/server/id"
 	// keyOpDefaultRetryCnt is the default retry count for etcd store.
 	keyOpDefaultRetryCnt = 5
 	// keyOpDefaultTimeout is the default time out for etcd store.
@@ -55,6 +58,7 @@ const (
 // InfoSyncer stores server info to etcd when the tidb-server starts and delete when tidb-server shuts down.
 type InfoSyncer struct {
 	etcdCli        *clientv3.Client
+	store          kv.Storage
 	info           *ServerInfo
 	serverInfoPath string
 	minStartTS     uint64
@@ -67,6 +71,7 @@ type InfoSyncer struct {
 // It will not be updated when tidb-server running. So please only put static information in ServerInfo struct.
 type ServerInfo struct {
 	ServerVersionInfo
+	ServerID   int64  `json:"server_id"`
 	ID         string `json:"ddl_id"`
 	IP         string `json:"ip"`
 	Port       uint   `json:"listening_port"`
@@ -81,11 +86,13 @@ type ServerVersionInfo struct {
 }
 
 var globalInfoSyncer *InfoSyncer
+var globalServerID int64
 
 // GlobalInfoSyncerInit return a new InfoSyncer. It is exported for testing.
-func GlobalInfoSyncerInit(ctx context.Context, id string, etcdCli *clientv3.Client) (*InfoSyncer, error) {
+func GlobalInfoSyncerInit(ctx context.Context, id string, etcdCli *clientv3.Client, store kv.Storage) (*InfoSyncer, error) {
 	globalInfoSyncer = &InfoSyncer{
 		etcdCli:        etcdCli,
+		store:          store,
 		info:           getServerInfo(id),
 		serverInfoPath: fmt.Sprintf("%s/%s", ServerInformationPath, id),
 		minStartTSPath: fmt.Sprintf("%s/%s", ServerMinStartTSPath, id),
@@ -161,6 +168,13 @@ func (is *InfoSyncer) storeServerInfo(ctx context.Context) error {
 	if is.etcdCli == nil {
 		return nil
 	}
+	id, err := is.GenGlobalServerID(ctx)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	is.info.ServerID = id
+	globalServerID = id
+
 	infoBuf, err := json.Marshal(is.info)
 	if err != nil {
 		return errors.Trace(err)
@@ -170,6 +184,22 @@ func (is *InfoSyncer) storeServerInfo(ctx context.Context) error {
 	return err
 }
 
+func GetGlobalServerID() int64 {
+	return globalServerID
+}
+
+func (is *InfoSyncer) GenGlobalServerID(ctx context.Context) (int64, error) {
+	var id int64
+	err := kv.RunInNewTxn(is.store, true, func(txn kv.Transaction) error {
+		t := meta.NewMeta(txn)
+		var err error
+		id, err = t.GenGlobalServerID()
+		return errors.Trace(err)
+	})
+	logutil.BgLogger().Info("gen server id", zap.Int64("id", int64(id)), zap.Error(err))
+	return id, err
+}
+
 // RemoveServerInfo remove self server static information from etcd.
 func (is *InfoSyncer) RemoveServerInfo() {
 	if is.etcdCli == nil {

executor/set.go

@@ -16,6 +16,9 @@ package executor
 import (
 	"context"
 	"fmt"
+	"github.com/pingcap/tidb/kv"
+	"github.com/pingcap/tipb/go-tipb"
+	"math"
 	"strings"
 
 	"github.com/pingcap/errors"
@@ -89,6 +92,13 @@ func (e *SetExecutor) Next(ctx context.Context, req *chunk.Chunk) error {
 			continue
 		}
 
+		if v.IsCluster {
+			err := e.setClusterServerVar(ctx, name, v)
+			if err != nil {
+				return err
+			}
+		}
+
 		syns := e.getSynonyms(name)
 		// Set system variable
 		for _, n := range syns {
@@ -101,6 +111,65 @@ func (e *SetExecutor) Next(ctx context.Context, req *chunk.Chunk) error {
 	return nil
 }
 
+func (e *SetExecutor) setClusterServerVar(ctx context.Context, name string, v *expression.VarAssignment) error {
+	sysVar := variable.GetSysVar(name)
+	if sysVar == nil {
+		return variable.UnknownSystemVar.GenWithStackByArgs(name)
+	}
+	if !variable.IsServerVariable(name) {
+		return errors.Errorf("variable `%v` is not a cluster server variable", name)
+	}
+	value, err := e.getVarValue(v, sysVar)
+	if err != nil {
+		return err
+	}
+	if value.IsNull() {
+		value.SetString("")
+	}
+	valStr, err := value.ToString()
+	if err != nil {
+		return err
+	}
+
+	setExec := PhysicalSetClusterVar{
+		name:  name,
+		value: valStr,
+	}.ToPB()
+	dagReq := &tipb.DAGRequest{
+		Executors: []*tipb.Executor{setExec},
+	}
+	dagData, err := dagReq.Marshal()
+	if err != nil {
+		return err
+	}
+	req := &kv.Request{
+		Tp:        kv.ReqTypeDAG,
+		Data:      dagData,
+		StoreType: kv.ClusterMem,
+		StartTs:   math.MaxUint64,
+	}
+
+	resp := e.ctx.GetClient().Send(ctx, req, e.ctx.GetSessionVars().KVVars)
+	if resp == nil {
+		return errors.New("client returns nil response")
+	}
+	_, err = resp.Next(ctx)
+	return err
+}
+
+type PhysicalSetClusterVar struct {
+	name  string
+	value string
+}
+
+func (p PhysicalSetClusterVar) ToPB() *tipb.Executor {
+	setExec := &tipb.SetServerVar{
+		VariableName:  p.name,
+		VariableValue: p.value,
+	}
+	return &tipb.Executor{Tp: tipb.ExecType_TypeSetServerVar, SetServerVar: setExec}
+}
+
 func (e *SetExecutor) getSynonyms(varName string) []string {
 	synonyms, ok := variable.SynonymsSysVariables[varName]
 	if ok {

expression/expression.go

@@ -503,6 +503,7 @@ type VarAssignment struct {
 	IsDefault   bool
 	IsGlobal    bool
 	IsSystem    bool
+	IsCluster   bool
 	ExtendValue *Constant
 }
 

go.mod

@@ -54,6 +54,7 @@ require (
 	github.com/shurcooL/httpfs v0.0.0-20171119174359-809beceb2371 // indirect
 	github.com/shurcooL/vfsgen v0.0.0-20181020040650-a97a25d856ca // indirect
 	github.com/sirupsen/logrus v1.2.0
+	github.com/soheilhy/cmux v0.1.4
 	github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72
 	github.com/struCoder/pidusage v0.1.2
 	github.com/tiancaiamao/appdash v0.0.0-20181126055449-889f96f722a2
@@ -80,3 +81,9 @@ require (
 )
 
 go 1.13
+
+replace github.com/pingcap/kvproto => github.com/crazycs520/kvproto v0.0.0-20191019041535-13904503e947
+
+replace github.com/pingcap/tipb => github.com/crazycs520/tipb v0.0.0-20191025111928-07ee0e9e0f91
+
+replace github.com/pingcap/parser => github.com/crazycs520/parser v0.0.0-20191025083943-997795afdec7

go.sum

@@ -27,6 +27,18 @@ github.com/coreos/go-systemd v0.0.0-20181031085051-9002847aa142/go.mod h1:F5haX7
 github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
 github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f h1:lBNOc5arjvs8E5mO2tbpBpLoyyu8B6e44T7hJy6potg=
 github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
+github.com/crazycs520/kvproto v0.0.0-20191019041535-13904503e947 h1:Boi74qEn74YPXfhnCCPNaOpJ1YfSDojQoQ3vd2YoB/4=
+github.com/crazycs520/kvproto v0.0.0-20191019041535-13904503e947/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w=
+github.com/crazycs520/parser v0.0.0-20191025070558-13bb6db9f318 h1:slqSgATev0xLUxB+yhF22m6GZVY7/nQO7gtiCTHTmNk=
+github.com/crazycs520/parser v0.0.0-20191025070558-13bb6db9f318/go.mod h1:xLjI+gnWYexq011WPMEvCNS8rFM9qe1vdojIEzSKPuc=
+github.com/crazycs520/parser v0.0.0-20191025083943-997795afdec7 h1:HWF2ryQvx8+IUVp2OiBlcCJwqXxwneMav0d4ow9n9gQ=
+github.com/crazycs520/parser v0.0.0-20191025083943-997795afdec7/go.mod h1:xLjI+gnWYexq011WPMEvCNS8rFM9qe1vdojIEzSKPuc=
+github.com/crazycs520/tipb v0.0.0-20191019072218-db0762008467 h1:afsxSYCCaDb7JCldCb43GnACvhF0Oau2/m1x0ZVLz2I=
+github.com/crazycs520/tipb v0.0.0-20191019072218-db0762008467/go.mod h1:J5W0Gvay/AsnE5I7p3JH1RG3HemVvEmp/gSSBCn1NiA=
+github.com/crazycs520/tipb v0.0.0-20191025054918-3710de5b0513 h1:trhYPGvwWvbqhcas+EBwidXiwVS3coFd5pUPtiDrdhk=
+github.com/crazycs520/tipb v0.0.0-20191025054918-3710de5b0513/go.mod h1:J5W0Gvay/AsnE5I7p3JH1RG3HemVvEmp/gSSBCn1NiA=
+github.com/crazycs520/tipb v0.0.0-20191025111928-07ee0e9e0f91 h1:nYjZf93B46Qsad25jabapIg8oYl0t5TxozN5ZPWJDe4=
+github.com/crazycs520/tipb v0.0.0-20191025111928-07ee0e9e0f91/go.mod h1:J5W0Gvay/AsnE5I7p3JH1RG3HemVvEmp/gSSBCn1NiA=
 github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548 h1:iwZdTE0PVqJCos1vaoKsclOGD3ADKpshg3SRtYBbwso=
 github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548/go.mod h1:e6NPNENfs9mPDVNRekM7lKScauxd5kXTr1Mfyig6TDM=
 github.com/cznic/sortutil v0.0.0-20150617083342-4c7342852e65 h1:hxuZop6tSoOi0sxFzoGGYdRqNrPubyaIf9KoBG9tPiE=
@@ -190,6 +202,7 @@ github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d h1:GoAlyOgbOEIFd
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/remyoudompheng/bigfft v0.0.0-20190512091148-babf20351dd7 h1:FUL3b97ZY2EPqg2NbXKuMHs5pXJB9hjj1fDHnF2vl28=
 github.com/remyoudompheng/bigfft v0.0.0-20190512091148-babf20351dd7/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
+github.com/sergi/go-diff v1.0.1-0.20180205163309-da645544ed44 h1:tB9NOR21++IjLyVx3/PCPhWMwqGNCMQEH96A6dMZ/gc=
 github.com/sergi/go-diff v1.0.1-0.20180205163309-da645544ed44/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/shirou/gopsutil v2.18.10+incompatible h1:cy84jW6EVRPa5g9HAHrlbxMSIjBhDSX0OFYyMYminYs=
 github.com/shirou/gopsutil v2.18.10+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=

infoschema/cluster.go

@@ -0,0 +1,126 @@
+package infoschema
+
+import (
+	"github.com/pingcap/parser/mysql"
+	"github.com/pingcap/tidb/domain/infosync"
+	"github.com/pingcap/tidb/sessionctx"
+	"github.com/pingcap/tidb/sessionctx/variable"
+	"github.com/pingcap/tidb/types"
+	"github.com/pingcap/tidb/util/stmtsummary"
+	"strings"
+)
+
+const clusterTableSuffix = "_CLUSTER"
+
+const TableNameEventsStatementsSummaryByDigestUpper = "EVENTS_STATEMENTS_SUMMARY_BY_DIGEST"
+
+// Cluster table list.
+const (
+	clusterTableSlowLog                             = tableSlowLog + clusterTableSuffix
+	clusterTableProcesslist                         = tableProcesslist + clusterTableSuffix
+	clusterTableNameEventsStatementsSummaryByDigest = TableNameEventsStatementsSummaryByDigestUpper + clusterTableSuffix
+	clusterTableServerVariable                      = "SERVER_VARIABLES" + clusterTableSuffix
+)
+
+// cluster table columns
+var (
+	clusterSlowQueryCols   []columnInfo
+	clusterProcesslistCols []columnInfo
+	clusterServerVarCols   []columnInfo
+)
+
+// register for cluster memory tables;
+var clusterTableMap = map[string]struct{}{
+	clusterTableSlowLog:                             {},
+	clusterTableProcesslist:                         {},
+	clusterTableNameEventsStatementsSummaryByDigest: {},
+	clusterTableServerVariable:                      {},
+}
+
+var clusterTableCols = []columnInfo{
+	{"TiDB_ID", mysql.TypeLonglong, 20, mysql.UnsignedFlag, nil, nil},
+}
+
+func init() {
+	// Slow query
+	clusterSlowQueryCols = append(clusterSlowQueryCols, slowQueryCols...)
+	clusterSlowQueryCols = append(clusterSlowQueryCols, clusterTableCols...)
+	// ProcessList
+	clusterProcesslistCols = append(clusterProcesslistCols, tableProcesslistCols...)
+	clusterProcesslistCols = append(clusterProcesslistCols, clusterTableCols...)
+
+	// Register Server_variables
+	clusterServerVarCols = append(clusterServerVarCols, sessionVarCols...)
+	clusterServerVarCols = append(clusterServerVarCols, clusterTableCols...)
+
+	// Register tidb_mem_cluster information_schema tables.
+	tableNameToColumns[clusterTableSlowLog] = clusterSlowQueryCols
+	tableNameToColumns[clusterTableProcesslist] = clusterProcesslistCols
+
+	// Register tikv mem_table to information_schema tables.
+	tableNameToColumns[tableTiKVInfo] = tikvInfoCols
+
+	// Register Server_variables
+	tableNameToColumns[clusterTableServerVariable] = clusterServerVarCols
+}
+
+func IsClusterTable(tableName string) bool {
+	tableName = strings.ToUpper(tableName)
+	_, ok := clusterTableMap[tableName]
+	return ok
+}
+
+func GetClusterMemTableRows(ctx sessionctx.Context, tableName string) (rows [][]types.Datum, err error) {
+	tableName = strings.ToUpper(tableName)
+	switch tableName {
+	case clusterTableSlowLog:
+		rows, err = dataForClusterSlowLog(ctx)
+	case clusterTableProcesslist:
+		rows, err = dataForClusterProcesslist(ctx)
+	case clusterTableNameEventsStatementsSummaryByDigest:
+		rows = dataForClusterTableNameEventsStatementsSummaryByDigest()
+	case clusterTableServerVariable:
+		rows, err = dataForServerVar(ctx)
+	}
+	return rows, err
+}
+
+func dataForClusterSlowLog(ctx sessionctx.Context) ([][]types.Datum, error) {
+	rows, err := dataForSlowLog(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return appendClusterColumnsToRows(rows), nil
+}
+
+func dataForClusterProcesslist(ctx sessionctx.Context) ([][]types.Datum, error) {
+	rows := dataForProcesslist(ctx)
+	return appendClusterColumnsToRows(rows), nil
+}
+
+func dataForClusterTableNameEventsStatementsSummaryByDigest() [][]types.Datum {
+	rows := stmtsummary.StmtSummaryByDigestMap.ToDatum()
+	return appendClusterColumnsToRows(rows)
+}
+
+func dataForServerVar(ctx sessionctx.Context) (rows [][]types.Datum, err error) {
+	sessionVars := ctx.GetSessionVars()
+	for name := range variable.ServerVariableMap {
+		var value string
+		value, err = variable.GetSessionSystemVar(sessionVars, name)
+		if err != nil {
+			return nil, err
+		}
+		row := types.MakeDatums(name, value)
+		rows = append(rows, row)
+	}
+	return appendClusterColumnsToRows(rows), nil
+}
+
+func appendClusterColumnsToRows(rows [][]types.Datum) [][]types.Datum {
+	for i := range rows {
+		rows[i] = append(rows[i], types.NewUintDatum(uint64(infosync.GetGlobalServerID())))
+	}
+	return rows
+}

infoschema/infoschema.go

@@ -386,8 +386,8 @@ func initInfoSchemaDB() {
 		tableInfo := buildTableMeta(name, cols)
 		infoSchemaTables = append(infoSchemaTables, tableInfo)
 		tableInfo.ID = autoid.GenLocalSchemaID()
-		for _, c := range tableInfo.Columns {
-			c.ID = autoid.GenLocalSchemaID()
+		for i, c := range tableInfo.Columns {
+			c.ID = int64(i) + 1
 		}
 	}
 	infoSchemaDB = &model.DBInfo{