optimize memory in hundreds-of tables scenario

Makefile

@@ -27,6 +27,12 @@ ARCH      := "`uname -s`"
 LINUX     := "Linux"
 MAC       := "Darwin"
 
+RACE_FLAG = 
+ifeq ("$(WITH_RACE)", "1")
+	RACE_FLAG = -race
+	GOBUILD   = GOPATH=$(GOPATH) CGO_ENABLED=1 $(GO) build
+endif
+
 .PHONY: all build parser clean parserlib lightning
 
 default: clean lightning checksuccess
@@ -63,21 +69,11 @@ parser: goyacc
 
 	@awk 'BEGIN{print "// Code generated by goyacc"} {print $0}' $(TIDBDIR)/parser/parser.go > tmp_parser.go && mv tmp_parser.go $(TIDBDIR)/parser/parser.go;
 
-parserlib: parser/parser.go
-
-parser/parser.go: $(TIDBDIR)/parser/parser.y
-	make parser
-
-RACE_FLAG = 
-ifeq ("$(WITH_RACE)", "1")
-	RACE_FLAG = -race
-	GOBUILD   = GOPATH=$(GOPATH) CGO_ENABLED=1 $(GO) build
-endif
 
 lightning: 
 	$(GOBUILD) $(RACE_FLAG) -ldflags '$(LDFLAGS)' -o $(LIGHTNING_BIN) cmd/main.go
 
-update: update_vendor parserlib clean_vendor
+update: update_vendor parser clean_vendor
 update_vendor:
 	@which glide >/dev/null || curl https://glide.sh/get | sh
 	@which glide-vc || go get -v -u github.com/sgotti/glide-vc

glide.yaml

@@ -13,7 +13,7 @@ import:
   repo: https://github.com/natefinch/lumberjack.git
   version: v2.0.0
 - package: github.com/pingcap/tidb
-  version: 1e82cb9632346d35b61e416eae0264447cb3a855
+  version: 8058fd644b57089640502d946466ae84fd490e25
   subpackages:
   - ast
   - ddl
@@ -81,7 +81,7 @@ import:
 - package: github.com/coreos/etcd
   version: v3.2.18
 - package: github.com/pingcap/tipb
-  version: 2d5073e5521a10f8db2adc0eaab2c46ca0a151d9
+  version: 015341a07dc30de16a7cca0fc77231ccacc4b397
 - package: github.com/apache/thrift
   version: b2a4d4ae21c789b689dd162deb819665567f481c
   # for building goyacc

lightning/kv/sql2kv.go

@@ -3,6 +3,7 @@ package kv
 import (
 	"github.com/juju/errors"
 	"github.com/pingcap/tidb-lightning/lightning/common"
+	"github.com/pingcap/tidb-lightning/lightning/metric"
 	sqltool "github.com/pingcap/tidb-lightning/lightning/sql"
 	"github.com/pingcap/tidb/kv"
 	kvec "github.com/pingcap/tidb/util/kvencoder"
@@ -22,11 +23,9 @@ func InitMembufCap(batchSQLLength int64) {
 }
 
 type TableKVEncoder struct {
-	db          string
-	table       string
-	tableID     int64
-	tableSchema string
-	columns     int
+	table   string
+	tableID int64
+	columns int
 
 	stmtIds   []uint32
 	bufValues []interface{}
@@ -36,45 +35,41 @@ type TableKVEncoder struct {
 }
 
 func NewTableKVEncoder(
-	db string, table string, tableID int64,
-	columns int, tableSchema string, sqlMode string, idAlloc *kvec.Allocator) (*TableKVEncoder, error) {
+	dbName string,
+	table string, tableID int64,
+	columns int, sqlMode string) (*TableKVEncoder, error) {
 
-	kvEncoder, err := kvec.New(db, idAlloc)
+	idAlloc := kvec.NewAllocator()
+	encoder, err := kvec.New(dbName, idAlloc)
 	if err != nil {
-		common.AppLogger.Errorf("[sql2kv] kv encoder create failed : %v", err)
+		common.AppLogger.Errorf("err %s", errors.ErrorStack(err))
 		return nil, errors.Trace(err)
 	}
 
-	err = kvEncoder.SetSystemVariable("sql_mode", sqlMode)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-	common.AppLogger.Debugf("set sql_mode=%s", sqlMode)
-
 	kvcodec := &TableKVEncoder{
-		db:          db,
 		table:       table,
 		tableID:     tableID,
-		encoder:     kvEncoder,
+		encoder:     encoder,
 		idAllocator: idAlloc,
-		tableSchema: tableSchema,
 		columns:     columns,
 	}
 
-	if err = kvcodec.init(); err != nil {
+	if err := kvcodec.init(sqlMode); err != nil {
 		kvcodec.Close()
 		return nil, errors.Trace(err)
 	}
 
+	metric.KvEncoderCounter.WithLabelValues("open").Inc()
+
 	return kvcodec, nil
 }
 
-func (kvcodec *TableKVEncoder) init() error {
-	if err := kvcodec.encoder.ExecDDLSQL(kvcodec.tableSchema); err != nil {
-		common.AppLogger.Errorf("[sql2kv] tableSchema execute failed : %v", err)
+func (kvcodec *TableKVEncoder) init(sqlMode string) error {
+	err := kvcodec.encoder.SetSystemVariable("sql_mode", sqlMode)
+	if err != nil {
 		return errors.Trace(err)
 	}
-
+	common.AppLogger.Debugf("set sql_mode=%s", sqlMode)
 	if PrepareStmtMode {
 		reserve := (encodeBatchRows * kvcodec.columns) << 1 // TODO : rows x ( cols + indices )
 		kvcodec.bufValues = make([]interface{}, 0, reserve)
@@ -107,6 +102,7 @@ func (kvcodec *TableKVEncoder) ResetRowID(rowID int64) {
 }
 
 func (kvcodec *TableKVEncoder) Close() error {
+	metric.KvEncoderCounter.WithLabelValues("closed").Inc()
 	return errors.Trace(kvcodec.encoder.Close())
 }
 

lightning/metric/metric.go

@@ -40,12 +40,21 @@ var (
 			Name:      "idle_workers",
 			Help:      "counting idle workers",
 		}, []string{"name"})
+
+	KvEncoderCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "lightning",
+			Name:      "kv_encoder",
+			Help:      "counting kv open and closed kv encoder",
+		}, []string{"type"},
+	)
 )
 
 func init() {
 	prometheus.MustRegister(cpuUsageGauge)
 	prometheus.MustRegister(IdleWorkersGauge)
 	prometheus.MustRegister(EngineCounter)
+	prometheus.MustRegister(KvEncoderCounter)
 }
 
 func CalcCPUUsageBackground(ctx context.Context) {