func (m *Meta) updateDDLJob(index int64, job *model.Job, key []byte, updateRawArgs bool) error {
	b, err := job.Encode(updateRawArgs)
	if err == nil {
		err = m.txn.LSet(key, index, b)
	}
	return errors.Trace(err)
}
updateDDLJob updates the DDL job with index and key.
updateRawArgs is used to determine whether to update the raw args when encode the job.
package meta
import (
	"context"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/pingcap/errors"
	"github.com/pingcap/parser/terror"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/metrics"
	"github.com/pingcap/tidb/util/logutil"
	"github.com/zhihu/zetta/pkg/codec"
	"github.com/zhihu/zetta/pkg/model"
	"github.com/zhihu/zetta/pkg/structure"
	"go.uber.org/zap"
)
var (
	globalIDMutex sync.Mutex
)
var (
	mMetaPrefix       = codec.EncodeInt([]byte{}, int64(0))
	mNextGlobalIDKey  = []byte("NextGlobalID")
	mSchemaVersionKey = []byte("SchemaVersionKey")
	mDBs              = []byte("DBs")
	mDBPrefix         = "DB"
	mTablePrefix      = "Table"
	mTableIDPrefix    = "TID"
	mBootstrapKey     = []byte("BootstrapKey")
	mSchemaDiffPrefix = "Diff"
)
var (
	errInvalidTableKey = terror.ClassMeta.New(codeInvalidTableKey, "invalid table meta key")
	errInvalidDBKey    = terror.ClassMeta.New(codeInvalidDBKey, "invalid db key")
	ErrDBExists = terror.ClassMeta.New(codeDatabaseExists, "database already exists")
	ErrDBNotExists = terror.ClassMeta.New(codeDatabaseNotExists, "database doesn't exist")
	ErrTableExists = terror.ClassMeta.New(codeTableExists, "table already exists")
	ErrTableNotExists = terror.ClassMeta.New(codeTableNotExists, "table doesn't exist")
)
const reservedIDLength = 10  
type Meta struct {
	txn        *structure.TxStructure
	StartTS    uint64  
	jobListKey JobListKeyType
}
func NewMeta(txn kv.Transaction, jobListKeys ...JobListKeyType) *Meta {
	txn.SetOption(kv.Priority, kv.PriorityHigh)
	txn.SetOption(kv.SyncLog, true)
	t := structure.NewStructure(txn, txn, mMetaPrefix)
	listKey := DefaultJobListKey
	if len(jobListKeys) != 0 {
		listKey = jobListKeys[0]
	}
	return &Meta{txn: t,
		StartTS:    txn.StartTS(),
		jobListKey: listKey,
	}
}
func NewSnapshotMeta(snapshot kv.Snapshot) *Meta {
	t := structure.NewStructure(snapshot, nil, mMetaPrefix)
	return &Meta{txn: t}
}
func (m *Meta) GenGlobalID() (int64, error) {
	globalIDMutex.Lock()
	defer globalIDMutex.Unlock()
	id, err := m.txn.Inc(mNextGlobalIDKey, 1)
	return id + reservedIDLength, err
}
func (m *Meta) GetGlobalID() (int64, error) {
	id, err := m.txn.GetInt64(mNextGlobalIDKey)
	return id + reservedIDLength, err
}
func (m *Meta) dbKey(dbID int64) []byte {
	return []byte(fmt.Sprintf("%s:%d", mDBPrefix, dbID))
}
func (m *Meta) autoTableIDKey(tableID int64) []byte {
	return []byte(fmt.Sprintf("%s:%d", mTableIDPrefix, tableID))
}
func (m *Meta) tableKey(tableID int64) []byte {
	return []byte(fmt.Sprintf("%s:%d", mTablePrefix, tableID))
}
func DDLJobHistoryKey(m *Meta, jobID int64) []byte {
	return m.txn.EncodeHashDataKey(mDDLJobHistoryKey, m.jobIDKey(jobID))
}
func (m *Meta) GenAutoTableIDKeyValue(dbID, tableID, autoID int64) (key, value []byte) {
	dbKey := m.dbKey(dbID)
	autoTableIDKey := m.autoTableIDKey(tableID)
	return m.txn.EncodeHashAutoIDKeyValue(dbKey, autoTableIDKey, autoID)
}
func (m *Meta) GenAutoTableID(dbID, tableID, step int64) (int64, error) {
	dbKey := m.dbKey(dbID)
	if err := m.checkDBExists(dbKey); err != nil {
		return 0, errors.Trace(err)
	}
	tableKey := m.tableKey(tableID)
	if err := m.checkTableExists(dbKey, tableKey); err != nil {
		return 0, errors.Trace(err)
	}
	return m.txn.HInc(dbKey, m.autoTableIDKey(tableID), step)
}
func (m *Meta) GetAutoTableID(dbID int64, tableID int64) (int64, error) {
	return m.txn.HGetInt64(m.dbKey(dbID), m.autoTableIDKey(tableID))
}
func (m *Meta) GetSchemaVersion() (int64, error) {
	return m.txn.GetInt64(mSchemaVersionKey)
}
func (m *Meta) GenSchemaVersion() (int64, error) {
	return m.txn.Inc(mSchemaVersionKey, 1)
}
func (m *Meta) checkDBExists(dbKey []byte) error {
	v, err := m.txn.HGet(mDBs, dbKey)
	if err == nil && v == nil {
		err = ErrDBNotExists
	}
	return errors.Trace(err)
}
func (m *Meta) checkDBNotExists(dbKey []byte) error {
	v, err := m.txn.HGet(mDBs, dbKey)
	if err == nil && v != nil {
		err = ErrDBExists
	}
	return errors.Trace(err)
}
func (m *Meta) checkTableExists(dbKey []byte, tableKey []byte) error {
	v, err := m.txn.HGet(dbKey, tableKey)
	if err == nil && v == nil {
		err = ErrTableNotExists
	}
	return errors.Trace(err)
}
func (m *Meta) checkTableNotExists(dbKey []byte, tableKey []byte) error {
	fmt.Println("meta checkTableExists:", string(dbKey), string(tableKey))
	v, err := m.txn.HGet(dbKey, tableKey)
	if err == nil && v != nil {
		err = ErrTableExists
	}
	return errors.Trace(err)
}
func (m *Meta) CreateDatabase(dbMeta *model.DatabaseMeta) error {
	dbKey := m.dbKey(dbMeta.Id)
	if err := m.checkDBNotExists(dbKey); err != nil {
		return errors.Trace(err)
	}
	data, err := json.Marshal(dbMeta)
	if err != nil {
		return errors.Trace(err)
	}
	return m.txn.HSet(mDBs, dbKey, data)
}
func (m *Meta) UpdateDatabase(dbMeta *model.DatabaseMeta) error {
	dbKey := m.dbKey(dbMeta.Id)
	if err := m.checkDBExists(dbKey); err != nil {
		return errors.Trace(err)
	}
	data, err := json.Marshal(dbMeta)
	if err != nil {
		return errors.Trace(err)
	}
	return m.txn.HSet(mDBs, dbKey, data)
}
func (m *Meta) CreateTableOrView(dbID int64, tableMeta *model.TableMeta) error {
	dbKey := m.dbKey(dbID)
	if err := m.checkDBExists(dbKey); err != nil {
		return errors.Trace(err)
	}
	tableKey := m.tableKey(tableMeta.Id)
	if err := m.checkTableNotExists(dbKey, tableKey); err != nil {
		return errors.Trace(err)
	}
	data, err := json.Marshal(tableMeta)
	if err != nil {
		return errors.Trace(err)
	}
	return m.txn.HSet(dbKey, tableKey, data)
}
func (m *Meta) CreateTableAndSetAutoID(dbID int64, tableMeta *model.TableMeta, autoID int64) error {
	err := m.CreateTableOrView(dbID, tableMeta)
	if err != nil {
		return errors.Trace(err)
	}
	_, err = m.txn.HInc(m.dbKey(dbID), m.autoTableIDKey(tableMeta.Id), autoID)
	return errors.Trace(err)
}
func (m *Meta) DropDatabase(dbID int64) error {
	dbKey := m.dbKey(dbID)
	if err := m.txn.HClear(dbKey); err != nil {
		return errors.Trace(err)
	}
	if err := m.txn.HDel(mDBs, dbKey); err != nil {
		return errors.Trace(err)
	}
	return nil
}
func (m *Meta) DropTableOrView(dbID int64, tblID int64, delAutoID bool) error {
	dbKey := m.dbKey(dbID)
	if err := m.checkDBExists(dbKey); err != nil {
		return errors.Trace(err)
	}
	tableKey := m.tableKey(tblID)
	if err := m.checkTableExists(dbKey, tableKey); err != nil {
		return errors.Trace(err)
	}
	if err := m.txn.HDel(dbKey, tableKey); err != nil {
		return errors.Trace(err)
	}
	if delAutoID {
		if err := m.txn.HDel(dbKey, m.autoTableIDKey(tblID)); err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}
func (m *Meta) UpdateTable(dbID int64, tableMeta *model.TableMeta) error {
	dbKey := m.dbKey(dbID)
	if err := m.checkDBExists(dbKey); err != nil {
		return errors.Trace(err)
	}
	tableKey := m.tableKey(tableMeta.Id)
	if err := m.checkTableExists(dbKey, tableKey); err != nil {
		return errors.Trace(err)
	}
	data, err := json.Marshal(tableMeta)
	if err != nil {
		return errors.Trace(err)
	}
	err = m.txn.HSet(dbKey, tableKey, data)
	return errors.Trace(err)
}
func (m *Meta) ListTables(dbID int64) ([]*model.TableMeta, error) {
	dbKey := m.dbKey(dbID)
	if err := m.checkDBExists(dbKey); err != nil {
		return nil, errors.Trace(err)
	}
	res, err := m.txn.HGetAll(dbKey)
	if err != nil {
		return nil, errors.Trace(err)
	}
	tables := make([]*model.TableMeta, 0, len(res)/2)
	for _, r := range res {
		tableKey := string(r.Field)
		if !strings.HasPrefix(tableKey, mTablePrefix) {
			continue
		}
		tbMeta := &model.TableMeta{}
		err = json.Unmarshal(r.Value, tbMeta)
		if err != nil {
			return nil, errors.Trace(err)
		}
		tables = append(tables, tbMeta)
	}
	return tables, nil
}
func (m *Meta) ListDatabases() ([]*model.DatabaseMeta, error) {
	res, err := m.txn.HGetAll(mDBs)
	if err != nil {
		return nil, errors.Trace(err)
	}
	dbs := make([]*model.DatabaseMeta, 0, len(res))
	for _, r := range res {
		dbInfo := &model.DatabaseMeta{}
		err = json.Unmarshal(r.Value, dbInfo)
		if err != nil {
			return nil, errors.Trace(err)
		}
		dbs = append(dbs, dbInfo)
	}
	return dbs, nil
}
func (m *Meta) GetDatabase(dbID int64) (*model.DatabaseMeta, error) {
	dbKey := m.dbKey(dbID)
	value, err := m.txn.HGet(mDBs, dbKey)
	if err != nil || value == nil {
		return nil, errors.Trace(err)
	}
	dbInfo := &model.DatabaseMeta{}
	err = json.Unmarshal(value, dbInfo)
	return dbInfo, errors.Trace(err)
}
func (m *Meta) GetTable(dbID int64, tableID int64) (*model.TableMeta, error) {
	dbKey := m.dbKey(dbID)
	if err := m.checkDBExists(dbKey); err != nil {
		return nil, errors.Trace(err)
	}
	tableKey := m.tableKey(tableID)
	value, err := m.txn.HGet(dbKey, tableKey)
	if err != nil || value == nil {
		return nil, errors.Trace(err)
	}
	tableMeta := &model.TableMeta{}
	err = json.Unmarshal(value, tableMeta)
	return tableMeta, errors.Trace(err)
}
var (
	mDDLJobListKey    = []byte("DDLJobList")
	mDDLJobAddIdxList = []byte("DDLJobAddIdxList")
	mDDLJobHistoryKey = []byte("DDLJobHistory")
	mDDLJobReorgKey   = []byte("DDLJobReorg")
)
type JobListKeyType []byte
var (
	DefaultJobListKey JobListKeyType = mDDLJobListKey
	AddIndexJobListKey JobListKeyType = mDDLJobAddIdxList
)
func (m *Meta) enQueueDDLJob(key []byte, job *model.Job) error {
	b, err := job.Encode(true)
	if err == nil {
		err = m.txn.RPush(key, b)
	}
	return errors.Trace(err)
}
func (m *Meta) EnQueueDDLJob(job *model.Job, jobListKeys ...JobListKeyType) error {
	listKey := m.jobListKey
	if len(jobListKeys) != 0 {
		listKey = jobListKeys[0]
	}
	return m.enQueueDDLJob(listKey, job)
}
func (m *Meta) deQueueDDLJob(key []byte) (*model.Job, error) {
	value, err := m.txn.LPop(key)
	if err != nil || value == nil {
		return nil, errors.Trace(err)
	}
	job := &model.Job{}
	err = job.Decode(value)
	return job, errors.Trace(err)
}
func (m *Meta) DeQueueDDLJob() (*model.Job, error) {
	return m.deQueueDDLJob(m.jobListKey)
}
func (m *Meta) getDDLJob(key []byte, index int64) (*model.Job, error) {
	value, err := m.txn.LIndex(key, index)
	if err != nil || value == nil {
		return nil, errors.Trace(err)
	}
	job := &model.Job{
		Priority: kv.PriorityLow,
	}
	err = job.Decode(value)
	if job.Priority < kv.PriorityNormal || job.Priority > kv.PriorityHigh {
		job.Priority = kv.PriorityLow
	}
	return job, errors.Trace(err)
}
func (m *Meta) GetDDLJobByIdx(index int64, jobListKeys ...JobListKeyType) (*model.Job, error) {
	listKey := m.jobListKey
	if len(jobListKeys) != 0 {
		listKey = jobListKeys[0]
	}
	startTime := time.Now()
	job, err := m.getDDLJob(listKey, index)
	metrics.MetaHistogram.WithLabelValues(metrics.GetDDLJobByIdx, metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
	return job, errors.Trace(err)
}