func (e *jsonEncoder) DecodeEvent(b []byte) (*types.CommonFormatEvent, error) {
	return e.jsonDecode(b)
}
DecodeEvent decodes JSON encoded array into CommonFormatEvent struct
package encoder
import (
	"bytes"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"strings"
	"time"
	"github.com/uber/storagetapper/log"
	"github.com/uber/storagetapper/state"
	"github.com/uber/storagetapper/types"
)
func init() {
	registerPlugin("json", initJSONEncoder)
}
type jsonEncoder struct {
	Service   string
	DB        string
	Table     string
	Input     string
	Output    string
	Version   int
	inSchema  *types.TableSchema
	filter    []int  
	outSchema *types.CommonFormatEvent
}
type GenTimeFunc func() int64
func genTime() int64 {
	return time.Now().UnixNano()
}
var GenTime GenTimeFunc = genTime
var ZeroTime = time.Time{}.In(time.UTC)
func initJSONEncoder(service, db, table, input string, output string, version int) (Encoder, error) {
	return &jsonEncoder{Service: service, DB: db, Table: table, Input: input, Output: output, Version: version}, nil
}
func (e *jsonEncoder) Type() string {
	return "json"
}
func (e *jsonEncoder) Schema() *types.TableSchema {
	return e.inSchema
}
func (e *jsonEncoder) EncodeSchema(seqno uint64) ([]byte, error) {
	return e.Row(types.Schema, nil, seqno, time.Time{})
}
func (e *jsonEncoder) Row(tp int, row *[]interface{}, seqno uint64, _ time.Time) ([]byte, error) {
	cf := e.convertRowToCommonFormat(tp, row, e.inSchema, seqno, e.filter)
	return e.CommonFormatEncode(cf)
}
func filterCommonFormat(filter []int, cf *types.CommonFormatEvent) *types.CommonFormatEvent {
	if len(filter) == 0 || cf.Fields == nil || len(*cf.Fields) == 0 {
		return cf
	}
	c := &types.CommonFormatEvent{SeqNo: cf.SeqNo, Type: cf.Type, Timestamp: cf.Timestamp, Key: cf.Key, Fields: new([]types.CommonFormatField)}
	for i, j := 0, 0; i < len(*cf.Fields); i++ {
		if filteredField(filter, i, &j) {
			continue
		}
		*c.Fields = append(*c.Fields, (*cf.Fields)[i])
	}
	return c
}
func (e *jsonEncoder) CommonFormat(cf *types.CommonFormatEvent) ([]byte, error) {
	if cf.Type == "schema" {
		err := e.UpdateCodec()
		if err != nil {
			return nil, err
		}
	}
	cf = filterCommonFormat(e.filter, cf)
	return e.CommonFormatEncode(cf)
}
func (e *jsonEncoder) UpdateCodec() error {
	schema, err := state.GetSchema(e.Service, e.DB, e.Table, e.Input, e.Output, e.Version)
	if err != nil {
		return err
	}
	e.inSchema = schema
	n, err := GetOutputSchemaName(e.Service, e.DB, e.Table, e.Input, e.Output, e.Version)
	if err != nil {
		return err
	}
	s := state.GetOutputSchema(n, "json")
	if s != "" {
		c, err := e.schemaDecode([]byte(s))
		if err != nil {
			return err
		}
		if c.Type != "schema" {
			return nil
		}
		e.outSchema = c
		if e.outSchema.Fields != nil && len(e.inSchema.Columns)-len(*e.outSchema.Fields) < 0 {
			err = fmt.Errorf("input schema has less fields than output schema")
			log.E(err)
			return err
		}
	}
	e.prepareFilter()
	return err
}
func fixFieldType(f interface{}, dt string) (interface{}, error) {
	var err error
	switch v := f.(type) {
	case float64:
		switch dt {
		case "bigint", "enum", "set":
			f = int64(v)
		case "int", "integer", "tinyint", "smallint", "mediumint", "year":
			f = int32(v)
		case "float":
			f = float32(v)
		}
	case string:
		switch dt {
		case "blob", "tinyblob", "mediumblob", "longblob", "binary", "varbinary":
			f, err = base64.StdEncoding.DecodeString(v)
			if err != nil {
				return nil, err
			}
		case "timestamp", "datetime":
			t := ZeroTime
			if !strings.HasPrefix(v, "0000-00-00 00:00:00") {
				t, err = time.Parse(time.RFC3339Nano, v)
				if err != nil {
					return nil, err
				}
				if dt == "timestamp" && !t.Equal(time.Time{}) {
					name, _ := time.Now().Zone()  
					if t.Location().String() == "UTC" && name == "UTC" {
						t = t.In(time.Local)
					}
				} else {
					t = t.In(time.UTC)
				}
			}
			f = t
		}
	}
	return f, nil
}
func (e *jsonEncoder) fixFieldTypes(res *types.CommonFormatEvent) (err error) {
	k := 0
	if e.inSchema != nil && res.Type != "schema" {
		var j int
		for i := 0; i < len(e.inSchema.Columns); i++ {
			if filteredField(e.filter, i, &j) {
				continue
			}
			if res.Fields != nil && i-j < len(*res.Fields) {
				f := &(*res.Fields)[i-j]
				f.Value, err = fixFieldType(f.Value, e.inSchema.Columns[i].DataType)
				if err != nil {
					return err
				}
			}
			if e.inSchema.Columns[i].Key == "PRI" && k < len(res.Key) {
				f := &res.Key[k]
				*f, err = fixFieldType(*f, e.inSchema.Columns[i].DataType)
				if err != nil {
					return err
				}
				k++
			}
		}
	}
	return nil
}
func (e *jsonEncoder) jsonDecode(b []byte) (*types.CommonFormatEvent, error) {
	res := &types.CommonFormatEvent{}
	err := json.Unmarshal(b, res)
	if err != nil {
		return nil, err
	}
	return res, e.fixFieldTypes(res)
}
func (e *jsonEncoder) schemaDecode(b []byte) (*types.CommonFormatEvent, error) {
	return e.jsonDecode(b)
}
func (e *jsonEncoder) CommonFormatEncode(c *types.CommonFormatEvent) ([]byte, error) {
	return json.Marshal(c)
}
func fillCommonFormatKey(e *types.CommonFormatEvent, row *[]interface{}, s *types.TableSchema) {
	e.Key = make([]interface{}, 0)
	for i := 0; i < len(s.Columns); i++ {
		if s.Columns[i].Key == "PRI" {
			if row == nil {
				e.Key = append(e.Key, s.Columns[i].Name)
			} else {
				e.Key = append(e.Key, (*row)[i])
			}
		}
	}
}
func fillCommonFormatFields(c *types.CommonFormatEvent, row *[]interface{}, schema *types.TableSchema, filter []int) {
	f := make([]types.CommonFormatField, 0, len(schema.Columns))
	for i, j := 0, 0; i < len(schema.Columns); i++ {
		if filteredField(filter, i, &j) {
			continue
		}
		var v interface{}
		if row == nil {
			v = schema.Columns[i].Type
		} else {
			s := schema.Columns[i].DataType
			if schema.Columns[i].Type == types.MySQLBoolean {
				b, ok := (*row)[i].(int8)
				if ok {
					if b != 0 {
						v = true
					} else {
						v = false
					}
				} else {
					v = (*row)[i]
				}
			} else if s == "text" || s == "tinytext" || s == "mediumtext" || s == "longtext" || s == "json" {
				b, ok := (*row)[i].([]byte)
				if ok {
					v = string(b)
				} else {
					v = (*row)[i]
				}
			} else if s == "binary" || s == "varbinary" {
				b, ok := (*row)[i].(string)
				if ok {
					v = []byte(b)
				} else {
					v = (*row)[i]
				}
			} else {
				v = (*row)[i]
			}
		}
		f = append(f, types.CommonFormatField{Name: schema.Columns[i].Name, Value: v})
	}
	c.Fields = &f
}
func (e *jsonEncoder) convertRowToCommonFormat(tp int, row *[]interface{}, schema *types.TableSchema, seqNo uint64, filter []int) *types.CommonFormatEvent {
	var c types.CommonFormatEvent
	c.SeqNo = seqNo
	c.Timestamp = GenTime()
	switch tp {
	case types.Insert:
		c.Type = "insert"
		fillCommonFormatKey(&c, row, schema)
		fillCommonFormatFields(&c, row, schema, filter)
	case types.Delete:
		c.Type = "delete"
		fillCommonFormatKey(&c, row, schema)
	case types.Schema:
		c.Type = "schema"
		fillCommonFormatKey(&c, nil, schema)
		fillCommonFormatFields(&c, nil, schema, filter)
	default:
		panic("unknown event type")
	}
	return &c
}
func (e *jsonEncoder) prepareFilter() {
	if e.outSchema == nil {
		return
	}
	nfiltered := len(e.inSchema.Columns)
	if e.outSchema.Fields != nil {
		nfiltered = nfiltered - len(*e.outSchema.Fields)
	}
	if nfiltered == 0 {
		return
	}
	f := e.outSchema.Fields
	e.filter = make([]int, 0)
	var j int
	for i := 0; i < len(e.inSchema.Columns); i++ {
		if f == nil || len(*f) == 0 || (i-j) >= len(*f) || e.inSchema.Columns[i].Name != (*f)[i-j].Name {
			if e.inSchema.Columns[i].Key != "PRI" {
				log.Debugf("Field %v will be filtered", e.inSchema.Columns[i].Name)
				e.filter = append(e.filter, i)
			}
			j++
		}
	}
	log.Debugf("len=%v, filtered fields (%v)", len(e.filter), e.filter)
}
func (e *jsonEncoder) UnwrapEvent(data []byte, cfEvent *types.CommonFormatEvent) (payload []byte, err error) {
	buf := bytes.NewBuffer(data)
	dec := json.NewDecoder(buf)
	err = dec.Decode(cfEvent)
	if err != nil {
		return
	}
	s, ok := cfEvent.Key[0].(string)
	if len(cfEvent.Key) > 1 || !ok || cfEvent.Type == "insert" || cfEvent.Type == "delete" {
		if err = e.fixFieldTypes(cfEvent); err != nil {
			return
		}
	} else if b, err := base64.StdEncoding.DecodeString(s); err == nil {
		cfEvent.Key[0] = string(b)
	} else {
		cfEvent.Key[0] = s
	}
	if e.inSchema != nil && cfEvent.Type == "schema" {
		if err = e.UpdateCodec(); err != nil {
			return
		}
	}
	var buf1 bytes.Buffer
	_, err = buf1.ReadFrom(dec.Buffered())
	if err != nil {
		return
	}
	_, err = buf1.ReadFrom(buf)
	if err != nil {
		return
	}
	return buf1.Bytes(), nil
}