func (krval bvcKeyRange) eval(bv interface{}, op Operator, onMismatch bool) bool {
	switch op {
	case QR_IN:
		switch num, status := getuint64(bv); status {
		case QR_OK:
			k := key.Uint64Key(num).KeyspaceId()
			return key.KeyRange(krval).Contains(k)
		case QR_OUT_OF_RANGE:
			return false
		}
		// Not a number. Check string.
		switch str, status := getstring(bv); status {
		case QR_OK:
			return key.KeyRange(krval).Contains(key.KeyspaceId(str))
		}
	case QR_NOTIN:
		switch num, status := getuint64(bv); status {
		case QR_OK:
			k := key.Uint64Key(num).KeyspaceId()
			return !key.KeyRange(krval).Contains(k)
		case QR_OUT_OF_RANGE:
			return true
		}
		// Not a number. Check string.
		switch str, status := getstring(bv); status {
		case QR_OK:
			return !key.KeyRange(krval).Contains(key.KeyspaceId(str))
		}
	default:
		panic("unexpected:")
	}
	return onMismatch
}

func (c *successClient) GetSrvKeyspace(ctx context.Context, keyspace string) (*topo.SrvKeyspace, error) {
	if keyspace == "big" {
		return &topo.SrvKeyspace{
			Partitions: map[topo.TabletType]*topo.KeyspacePartition{
				topo.TYPE_REPLICA: &topo.KeyspacePartition{
					ShardReferences: []topo.ShardReference{
						topo.ShardReference{
							Name: "shard0",
							KeyRange: key.KeyRange{
								Start: key.Uint64Key(0x4000000000000000).KeyspaceId(),
								End:   key.Uint64Key(0x8000000000000000).KeyspaceId(),
							},
						},
					},
				},
			},
			ShardingColumnName: "sharding_column_name",
			ShardingColumnType: key.KIT_UINT64,
			ServedFrom: map[topo.TabletType]string{
				topo.TYPE_MASTER: "other_keyspace",
			},
			SplitShardCount: 128,
		}, nil
	}
	if keyspace == "small" {
		return &topo.SrvKeyspace{}, nil
	}
	return c.fallback.GetSrvKeyspace(ctx, keyspace)
}

func (krval *bvcKeyRange) eval(bv interface{}, op Operator, onMismatch bool) bool {
	switch op {
	case QRIn:
		switch num, status := getuint64(bv); status {
		case QROK:
			k := key.Uint64Key(num).Bytes()
			return key.KeyRangeContains((*topodatapb.KeyRange)(krval), k)
		case QROutOfRange:
			return false
		}
		// Not a number. Check string.
		switch str, status := getstring(bv); status {
		case QROK:
			return key.KeyRangeContains((*topodatapb.KeyRange)(krval), []byte(str))
		}
	case QRNotIn:
		switch num, status := getuint64(bv); status {
		case QROK:
			k := key.Uint64Key(num).Bytes()
			return !key.KeyRangeContains((*topodatapb.KeyRange)(krval), k)
		case QROutOfRange:
			return true
		}
		// Not a number. Check string.
		switch str, status := getstring(bv); status {
		case QROK:
			return !key.KeyRangeContains((*topodatapb.KeyRange)(krval), []byte(str))
		}
	default:
		panic("unexpected:")
	}
	return onMismatch
}

func numKeyRange(start, end uint64) bvcKeyRange {
	kr := key.KeyRange{
		Start: key.Uint64Key(start).KeyspaceId(),
		End:   key.Uint64Key(end).KeyspaceId(),
	}
	return bvcKeyRange(kr)
}

func numKeyRange(start, end uint64) *bvcKeyRange {
	kr := pb.KeyRange{
		Start: key.Uint64Key(start).Bytes(),
		End:   key.Uint64Key(end).Bytes(),
	}
	b := bvcKeyRange(kr)
	return &b
}

func TestCSVSplitterNumber(t *testing.T) {
	// csvsplitter_mean.csv was generated using "select keyspaced_id,
	// tablename.* into outfile".
	keyspaceIds := readData(t, "csvsplitter_mean.csv", true)

	wantedTable := []pair{
		{key.Uint64Key(1).KeyspaceId(), "\"x\x9c\xf3H\xcd\xc9\xc9W(\xcf/\xcaI\x01\\0\x18\xab\x04=\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\",1,\"ala\\\nhas a cat\\\n\",1\n"},
		{key.Uint64Key(2).KeyspaceId(), "\"x\x9c\xf3\xc8\xcfIT\xc8-\xcdK\xc9\a\\0\x13\xfe\x03\xc8\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\",2,\"ala\\\ntiene un gato\\\\\\\n\r\\\n\",2\n"},
		{key.Uint64Key(3).KeyspaceId(), "\"x\x9cs\xceL\xccW\xc8\xcd\xcfK\xc9\a\\0\x13\x88\x03\xba\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\",3,\"ala\\\nha un gatto\\\\n\\\n\",3\n"},
		{key.Uint64Key(4).KeyspaceId(), "\"x\x9cs\xca\xcf\xcb\xca/-R\xc8\xcd\xcfKI\x05\\0#:\x05\x13\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\",3,\",,,ala \\\"\\\n,a un chat\",4\n"},
	}

	checkWanted(t, keyspaceIds, wantedTable)
}

func getBoundValue(valExpr sqlparser.ValExpr, bindVariables map[string]interface{}) (string, error) {
	switch node := valExpr.(type) {
	case sqlparser.ValTuple:
		if len(node) != 1 {
			return "", fmt.Errorf("tuples not allowed as insert values")
		}
		// TODO: Change parser to create single value tuples into non-tuples.
		return getBoundValue(node[0], bindVariables)
	case sqlparser.StrVal:
		return string(node), nil
	case sqlparser.NumVal:
		val, err := strconv.ParseInt(string(node), 10, 64)
		if err != nil {
			return "", err
		}
		return key.Uint64Key(val).String(), nil
	case sqlparser.ValArg:
		value, err := findBindValue(node, bindVariables)
		if err != nil {
			return "", err
		}
		return key.EncodeValue(value), nil
	}
	panic("Unexpected token")
}

// Split will split the rows into subset for each distribution
func (rs *RowSplitter) Split(result [][][]sqltypes.Value, rows [][]sqltypes.Value) error {
	if rs.Type == key.KIT_UINT64 {
		for _, row := range rows {
			v := sqltypes.MakeNumeric(row[rs.ValueIndex].Raw())
			i, err := v.ParseUint64()
			if err != nil {
				return fmt.Errorf("Non numerical value: %v", err)
			}
			k := key.Uint64Key(i).KeyspaceId()
			for i, kr := range rs.KeyRanges {
				if kr.Contains(k) {
					result[i] = append(result[i], row)
					break
				}
			}
		}
	} else {
		for _, row := range rows {
			k := key.KeyspaceId(row[rs.ValueIndex].Raw())
			for i, kr := range rs.KeyRanges {
				if kr.Contains(k) {
					result[i] = append(result[i], row)
					break
				}
			}
		}
	}
	return nil
}

// Split will split the rows into subset for each distribution
func (rs *RowSplitter) Split(result [][][]sqltypes.Value, rows [][]sqltypes.Value) error {
	if rs.KeyspaceIdType == topodatapb.KeyspaceIdType_UINT64 {
		for _, row := range rows {
			i, err := row[rs.ValueIndex].ParseUint64()
			if err != nil {
				return fmt.Errorf("Non numerical value: %v", err)
			}
			k := key.Uint64Key(i).Bytes()
			for i, kr := range rs.KeyRanges {
				if key.KeyRangeContains(kr, k) {
					result[i] = append(result[i], row)
					break
				}
			}
		}
	} else {
		for _, row := range rows {
			k := row[rs.ValueIndex].Raw()
			for i, kr := range rs.KeyRanges {
				if key.KeyRangeContains(kr, k) {
					result[i] = append(result[i], row)
					break
				}
			}
		}
	}
	return nil
}

func TestCSVSplitter(t *testing.T) {
	// mean.csv was generated using "select keyspaced_id,
	// tablename.* into outfile".
	file, err := os.Open("mean.csv")
	if err != nil {
		t.Fatalf("Cannot open mean.csv: %v", err)
	}
	r := NewKeyspaceCSVReader(file, ',')

	keyspaceIds := make([]pair, 0)

	for {
		kid, line, err := r.ReadRecord()
		if err == io.EOF {
			break
		}
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		keyspaceIds = append(keyspaceIds, pair{kid, string(line)})
	}

	wantedTable := []pair{
		{key.Uint64Key(1).KeyspaceId(), "\"x\x9c\xf3H\xcd\xc9\xc9W(\xcf/\xcaI\x01\\0\x18\xab\x04=\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\",1,\"ala\\\nhas a cat\\\n\",1\n"},
		{key.Uint64Key(2).KeyspaceId(), "\"x\x9c\xf3\xc8\xcfIT\xc8-\xcdK\xc9\a\\0\x13\xfe\x03\xc8\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\",2,\"ala\\\ntiene un gato\\\\\\\n\r\\\n\",2\n"},
		{key.Uint64Key(3).KeyspaceId(), "\"x\x9cs\xceL\xccW\xc8\xcd\xcfK\xc9\a\\0\x13\x88\x03\xba\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\",3,\"ala\\\nha un gatto\\\\n\\\n\",3\n"},
		{key.Uint64Key(4).KeyspaceId(), "\"x\x9cs\xca\xcf\xcb\xca/-R\xc8\xcd\xcfKI\x05\\0#:\x05\x13\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\\0\",3,\",,,ala \\\"\\\n,a un chat\",4\n"},
	}

	for i, wanted := range wantedTable {
		if keyspaceIds[i].kid != key.KeyspaceId(wanted.kid) {
			t.Errorf("Wrong keyspace_id: expected %#v, got %#v", wanted.kid, keyspaceIds[i].kid)
		}
		if keyspaceIds[i].line != wanted.line {
			t.Errorf("Wrong line: expected %q got %q", wanted.line, keyspaceIds[i].line)
		}
	}

	if count := len(keyspaceIds); count != 4 {
		t.Errorf("Wrong number of records: expected 4, got %v", count)
	}

}

// ReadRecord returns a keyspaceId and a line from which it was
// extracted, with the keyspaceId stripped.
func (r KeyspaceCSVReader) ReadRecord() (keyspaceId key.KeyspaceId, line []byte, err error) {
	k, err := r.reader.ReadString(r.delim)
	if err != nil {
		return key.MinKey, nil, err
	}
	if r.numberColumn {
		// the line starts with:
		// NNNN,
		// so remove the comma
		kid, err := strconv.ParseUint(k[:len(k)-1], 10, 64)
		if err != nil {
			return key.MinKey, nil, err
		}
		keyspaceId = key.Uint64Key(kid).KeyspaceId()
	} else {
		// the line starts with:
		// "HHHH",
		// so remove the quotes and comma
		keyspaceId, err = key.HexKeyspaceId(k[1 : len(k)-2]).Unhex()
		if err != nil {
			return key.MinKey, nil, err
		}
	}

	defer r.buf.Reset()

	escaped := false
	inQuote := false
	for {
		b, err := r.reader.ReadByte()
		if err != nil {
			// Assumption: the csv file ends with a
			// newline. Otherwise io.EOF should be treated
			// separately.
			return key.MinKey, nil, err
		}

		r.buf.WriteByte(b)

		if escaped {
			escaped = false
			continue
		}
		switch b {
		case '\\':
			escaped = true
		case '"':
			inQuote = !inQuote
		case '\n':
			if !inQuote {
				return keyspaceId, r.buf.Bytes(), nil
			}
		}
	}
}

// keyspaceID implements the keyspaceIDResolver interface.
func (r *v2Resolver) keyspaceID(row []sqltypes.Value) ([]byte, error) {
	v := row[r.shardingColumnIndex]
	switch r.keyspaceInfo.ShardingColumnType {
	case topodatapb.KeyspaceIdType_BYTES:
		return v.Raw(), nil
	case topodatapb.KeyspaceIdType_UINT64:
		i, err := v.ParseUint64()
		if err != nil {
			return nil, fmt.Errorf("Non numerical value: %v", err)
		}
		return key.Uint64Key(i).Bytes(), nil
	default:
		return nil, fmt.Errorf("unsupported ShardingColumnType: %v", r.keyspaceInfo.ShardingColumnType)
	}
}

// KeyRangeFilterFunc returns a function that calls sendReply only if statements
// in the transaction match the specified keyrange. The resulting function can be
// passed into the BinlogStreamer: bls.Stream(file, pos, sendTransaction) ->
// bls.Stream(file, pos, KeyRangeFilterFunc(sendTransaction))
func KeyRangeFilterFunc(keyrange key.KeyRange, sendReply sendTransactionFunc) sendTransactionFunc {
	return func(reply *proto.BinlogTransaction) error {
		matched := false
		filtered := make([]proto.Statement, 0, len(reply.Statements))
		for _, statement := range reply.Statements {
			switch statement.Category {
			case proto.BL_SET:
				filtered = append(filtered, statement)
			case proto.BL_DDL:
				filtered = append(filtered, statement)
				matched = true
			case proto.BL_DML:
				keyspaceIndex := bytes.LastIndex(statement.Sql, KEYSPACE_ID_COMMENT)
				if keyspaceIndex == -1 {
					updateStreamErrors.Add("KeyRangeStream", 1)
					log.Errorf("Error parsing keyspace id: %s", string(statement.Sql))
					continue
				}
				idstart := keyspaceIndex + len(KEYSPACE_ID_COMMENT)
				idend := bytes.Index(statement.Sql[idstart:], SPACE)
				if idend == -1 {
					updateStreamErrors.Add("KeyRangeStream", 1)
					log.Errorf("Error parsing keyspace id: %s", string(statement.Sql))
					continue
				}
				id, err := strconv.ParseUint(string(statement.Sql[idstart:idstart+idend]), 10, 64)
				if err != nil {
					updateStreamErrors.Add("KeyRangeStream", 1)
					log.Errorf("Error parsing keyspace id: %s", string(statement.Sql))
					continue
				}
				if !keyrange.Contains(key.Uint64Key(id).KeyspaceId()) {
					continue
				}
				filtered = append(filtered, statement)
				matched = true
			}
		}
		if matched {
			reply.Statements = filtered
		} else {
			reply.Statements = nil
		}
		return sendReply(reply)
	}
}

func (node *Node) getBoundValue(bindVariables map[string]interface{}) string {
	switch node.Type {
	case '(':
		return node.NodeAt(0).getBoundValue(bindVariables)
	case STRING:
		return string(node.Value)
	case NUMBER:
		val, err := strconv.ParseInt(string(node.Value), 10, 64)
		if err != nil {
			panic(NewParserError("%s", err.Error()))
		}
		return key.Uint64Key(val).String()
	case VALUE_ARG:
		value := node.findBindValue(bindVariables)
		return key.EncodeValue(value)
	}
	panic("Unexpected token")
}

func (blp *BinlogPlayer) createIndexUpdates(dmlEvent *mysqlctl.BinlogResponse) (indexSql string) {
	keyspaceIdUint, err := strconv.ParseUint(dmlEvent.KeyspaceId, 10, 64)
	if err != nil {
		panic(fmt.Errorf("Invalid keyspaceid '%v', error converting it, %v", dmlEvent.KeyspaceId, err))
	}
	keyspaceId := key.Uint64Key(keyspaceIdUint).KeyspaceId()

	if !blp.keyrange.Contains(keyspaceId) {
		panic(fmt.Errorf("Invalid keyspace id %v for range %v-%v", dmlEvent.KeyspaceId, blp.startPosition.KeyrangeStart, blp.startPosition.KeyrangeEnd))
	}

	if dmlEvent.IndexType != "" {
		indexSql, err = createIndexSql(dmlEvent.SqlType, dmlEvent.IndexType, dmlEvent.IndexId, dmlEvent.UserId)
		if err != nil {
			panic(fmt.Errorf("Error creating index update sql - IndexType %v, IndexId %v, UserId %v Sql '%v', err: '%v'", dmlEvent.IndexType, dmlEvent.IndexId, dmlEvent.UserId, dmlEvent.Sql, err))
		}
	}
	return
}

// ReadRecord returns a keyspaceId and a line from which it was
// extracted, with the keyspaceId stripped.
func (r KeyspaceCSVReader) ReadRecord() (keyspaceId key.KeyspaceId, line []byte, err error) {
	k, err := r.reader.ReadString(r.delim)
	if err != nil {
		return key.MinKey, nil, err
	}
	kid, err := strconv.ParseUint(k[:len(k)-1], 10, 64)
	if err != nil {
		return key.MinKey, nil, err
	}
	keyspaceId = key.Uint64Key(kid).KeyspaceId()

	defer r.buf.Reset()

	escaped := false
	inQuote := false
	for {
		b, err := r.reader.ReadByte()
		if err != nil {
			// Assumption: the csv file ends with a
			// newline. Otherwise io.EOF should be treated
			// separately.
			return key.MinKey, nil, err
		}

		r.buf.WriteByte(b)

		if escaped {
			escaped = false
			continue
		}
		switch b {
		case '\\':
			escaped = true
		case '"':
			inQuote = !inQuote
		case '\n':
			if !inQuote {
				return keyspaceId, r.buf.Bytes(), nil
			}
		}
	}
	panic("unreachable")
}

func parseKeyspaceId(sql []byte, dmlType string) (keyspaceIdStr string, keyspaceId key.KeyspaceId) {
	keyspaceIndex := bytes.Index(sql, KEYSPACE_ID_COMMENT)
	if keyspaceIndex == -1 {
		if controlDbStatement(sql, dmlType) {
			relog.Warning("Ignoring no keyspace id, control db stmt %v", string(sql))
			return
		}
		panic(NewBinlogParseError(fmt.Sprintf("Invalid Sql, doesn't contain keyspace id, sql: %v", string(sql))))
	}
	seekIndex := keyspaceIndex + len(KEYSPACE_ID_COMMENT)
	keyspaceIdComment := sql[seekIndex:]
	keyspaceIdStr = string(bytes.TrimSpace(bytes.SplitN(keyspaceIdComment, USER_ID, 2)[0]))
	if keyspaceIdStr == "" {
		panic(NewBinlogParseError(fmt.Sprintf("Invalid keyspace id, sql %v", string(sql))))
	}
	keyspaceIdUint, err := strconv.ParseUint(keyspaceIdStr, 10, 64)
	if err != nil {
		panic(NewBinlogParseError(fmt.Sprintf("Invalid keyspaceid, error converting it, sql %v", string(sql))))
	}
	keyspaceId = key.Uint64Key(keyspaceIdUint).KeyspaceId()
	return keyspaceIdStr, keyspaceId
}

// KeyRangeFilterFunc returns a function that calls sendReply only if statements
// in the transaction match the specified keyrange. The resulting function can be
// passed into the BinlogStreamer: bls.Stream(file, pos, sendTransaction) ->
// bls.Stream(file, pos, KeyRangeFilterFunc(sendTransaction))
func KeyRangeFilterFunc(kit key.KeyspaceIdType, keyrange *pb.KeyRange, sendReply sendTransactionFunc) sendTransactionFunc {
	isInteger := true
	if kit == key.KIT_BYTES {
		isInteger = false
	}

	return func(reply *proto.BinlogTransaction) error {
		matched := false
		filtered := make([]proto.Statement, 0, len(reply.Statements))
		for _, statement := range reply.Statements {
			switch statement.Category {
			case proto.BL_SET:
				filtered = append(filtered, statement)
			case proto.BL_DDL:
				log.Warningf("Not forwarding DDL: %s", string(statement.Sql))
				continue
			case proto.BL_DML:
				keyspaceIndex := bytes.LastIndex(statement.Sql, KEYSPACE_ID_COMMENT)
				if keyspaceIndex == -1 {
					updateStreamErrors.Add("KeyRangeStream", 1)
					log.Errorf("Error parsing keyspace id: %s", string(statement.Sql))
					continue
				}
				idstart := keyspaceIndex + len(KEYSPACE_ID_COMMENT)
				idend := bytes.Index(statement.Sql[idstart:], SPACE)
				if idend == -1 {
					updateStreamErrors.Add("KeyRangeStream", 1)
					log.Errorf("Error parsing keyspace id: %s", string(statement.Sql))
					continue
				}
				textId := string(statement.Sql[idstart : idstart+idend])
				if isInteger {
					id, err := strconv.ParseUint(textId, 10, 64)
					if err != nil {
						updateStreamErrors.Add("KeyRangeStream", 1)
						log.Errorf("Error parsing keyspace id: %s", string(statement.Sql))
						continue
					}
					if !key.KeyRangeContains(keyrange, key.Uint64Key(id).Bytes()) {
						continue
					}
				} else {
					data, err := base64.StdEncoding.DecodeString(textId)
					if err != nil {
						updateStreamErrors.Add("KeyRangeStream", 1)
						log.Errorf("Error parsing keyspace id: %s", string(statement.Sql))
						continue
					}
					if !key.KeyRangeContains(keyrange, data) {
						continue
					}
				}
				filtered = append(filtered, statement)
				matched = true
			case proto.BL_UNRECOGNIZED:
				updateStreamErrors.Add("KeyRangeStream", 1)
				log.Errorf("Error parsing keyspace id: %s", string(statement.Sql))
				continue
			}
		}
		if matched {
			reply.Statements = filtered
		} else {
			reply.Statements = nil
		}
		return sendReply(reply)
	}
}

// license that can be found in the LICENSE file.

package binlog

import (
	"fmt"
	"testing"

	"github.com/youtube/vitess/go/vt/binlog/proto"
	"github.com/youtube/vitess/go/vt/key"

	pb "github.com/youtube/vitess/go/vt/proto/topodata"
)

var testKeyRange = &pb.KeyRange{
	Start: key.Uint64Key(0).Bytes(),
	End:   key.Uint64Key(10).Bytes(),
}

func TestKeyRangeFilterPass(t *testing.T) {
	input := proto.BinlogTransaction{
		Statements: []proto.Statement{
			{
				Category: proto.BL_SET,
				Sql:      []byte("set1"),
			}, {
				Category: proto.BL_DML,
				Sql:      []byte("dml1 /* EMD keyspace_id:20 */"),
			}, {
				Category: proto.BL_DML,
				Sql:      []byte("dml2 /* EMD keyspace_id:2 */"),

// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package binlog

import (
	"fmt"
	"testing"

	"github.com/youtube/vitess/go/vt/binlog/proto"
	"github.com/youtube/vitess/go/vt/key"
	myproto "github.com/youtube/vitess/go/vt/mysqlctl/proto"
)

var testKeyRange = key.KeyRange{
	Start: key.KeyspaceId(key.Uint64Key(0).String()),
	End:   key.KeyspaceId(key.Uint64Key(10).String()),
}

func TestKeyRangeFilterPass(t *testing.T) {
	input := proto.BinlogTransaction{
		Statements: []proto.Statement{
			{
				Category: proto.BL_SET,
				Sql:      []byte("set1"),
			}, {
				Category: proto.BL_DML,
				Sql:      []byte("dml1 /* EMD keyspace_id:20 */"),
			}, {
				Category: proto.BL_DML,
				Sql:      []byte("dml2 /* EMD keyspace_id:2 */"),