Skip to content

Commit 1b8b54a

Browse files
author
James Cor
committed
implementing but something is wrong
1 parent 206f30e commit 1b8b54a

File tree

15 files changed

+458
-127
lines changed

15 files changed

+458
-127
lines changed

‎server/handler.go‎

Lines changed: 150 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -495,8 +495,8 @@ func (h *Handler) doQuery(
495495
r, err = resultForEmptyIter(sqlCtx, rowIter, resultFields)
496496
} else if analyzer.FlagIsSet(qFlags, sql.QFlagMax1Row) {
497497
r, err = resultForMax1RowIter(sqlCtx, schema, rowIter, resultFields, buf)
498-
} else if ri2, ok := rowIter.(sql.RowIter2); ok && ri2.IsRowIter2(sqlCtx) {
499-
r, processedAtLeastOneBatch, err = h.resultForDefaultIter2(sqlCtx, c, ri2, resultFields, callback, more)
498+
} else if r2, ok := rowIter.(sql.RowIter2); ok && r2.IsRowIter2(sqlCtx) {
499+
r, processedAtLeastOneBatch, err = h.resultForDefaultIter2(sqlCtx, c, r2, resultFields, callback, more)
500500
} else {
501501
r, processedAtLeastOneBatch, err = h.resultForDefaultIter(sqlCtx, c, schema, rowIter, callback, resultFields, more, buf)
502502
}
@@ -805,7 +805,6 @@ func (h *Handler) resultForDefaultIter2(ctx *sql.Context, c *mysql.Conn, iter sq
805805
wg := sync.WaitGroup{}
806806
wg.Add(2)
807807

808-
// TODO: send results instead of rows?
809808
// Read rows from iter and send them off
810809
var rowChan = make(chan sql.Row2, 512)
811810
eg.Go(func() (err error) {
@@ -869,7 +868,154 @@ func (h *Handler) resultForDefaultIter2(ctx *sql.Context, c *mysql.Conn, iter sq
869868
return nil
870869
}
871870
ctx.GetLogger().Tracef("spooling result row %s", row)
872-
res.Rows = append(res.Rows, row)
871+
vals := make([]sqltypes.Value, len(row))
872+
res.Rows = append(res.Rows, vals)
873+
res.RowsAffected++
874+
if !timer.Stop() {
875+
<-timer.C
876+
}
877+
}
878+
timer.Reset(waitTime)
879+
}
880+
})
881+
882+
// Close() kills this PID in the process list,
883+
// wait until all rows have be sent over the wire
884+
eg.Go(func() (err error) {
885+
defer pan2err(&err)
886+
wg.Wait()
887+
return iter.Close(ctx)
888+
})
889+
890+
err := eg.Wait()
891+
if err != nil {
892+
ctx.GetLogger().WithError(err).Warn("error running query")
893+
if verboseErrorLogging {
894+
fmt.Printf("Err: %+v", err)
895+
}
896+
return nil, false, err
897+
}
898+
899+
return res, processedAtLeastOneBatch, nil
900+
}
901+
902+
func (h *Handler) resultForRowFrameIter(ctx *sql.Context, c *mysql.Conn, iter sql.RowFrameIter, resultFields []*querypb.Field, callback func(*sqltypes.Result, bool) error, more bool) (*sqltypes.Result, bool, error) {
903+
defer trace.StartRegion(ctx, "Handler.resultForRowFrameIter").End()
904+
905+
eg, ctx := ctx.NewErrgroup()
906+
pan2err := func(err *error) {
907+
if recoveredPanic := recover(); recoveredPanic != nil {
908+
stack := debug.Stack()
909+
wrappedErr := fmt.Errorf("handler caught panic: %v\n%s", recoveredPanic, stack)
910+
*err = goerrors.Join(*err, wrappedErr)
911+
}
912+
}
913+
914+
// TODO: poll for closed connections should obviously also run even if
915+
// we're doing something with an OK result or a single row result, etc.
916+
// This should be in the caller.
917+
pollCtx, cancelF := ctx.NewSubContext()
918+
eg.Go(func() (err error) {
919+
defer pan2err(&err)
920+
return h.pollForClosedConnection(pollCtx, c)
921+
})
922+
923+
// Default waitTime is one minute if there is no timeout configured, in which case
924+
// it will loop to iterate again unless the socket died by the OS timeout or other problems.
925+
// If there is a timeout, it will be enforced to ensure that Vitess has a chance to
926+
// call Handler.CloseConnection()
927+
waitTime := 1 * time.Minute
928+
if h.readTimeout > 0 {
929+
waitTime = h.readTimeout
930+
}
931+
timer := time.NewTimer(waitTime)
932+
defer timer.Stop()
933+
934+
wg := sync.WaitGroup{}
935+
wg.Add(2)
936+
937+
// TODO: send results instead of rows?
938+
// Read rows from iter and send them off
939+
var rowFrameChan = make(chan sql.Row2, 512)
940+
//var rowFrameChan = make(chan *sql.RowFrame, 512)
941+
eg.Go(func() (err error) {
942+
defer pan2err(&err)
943+
defer wg.Done()
944+
defer close(rowFrameChan)
945+
for {
946+
select {
947+
case <-ctx.Done():
948+
return context.Cause(ctx)
949+
default:
950+
//rowFrame := sql.NewRowFrame()
951+
r2, ok := iter.(sql.RowIter2)
952+
if !ok {
953+
panic("aaaaaaasdfasdgsdfgsdfghsfgd")
954+
}
955+
row, err := r2.Next2(ctx)
956+
if err == io.EOF {
957+
return nil
958+
}
959+
if err != nil {
960+
return err
961+
}
962+
// DEEP COPY HERE IS IMPORTANT!
963+
//row := rowFrame.Row2Copy()
964+
// Should be safe to release memory
965+
//rowFrame.Recycle()
966+
select {
967+
case rowFrameChan <- row:
968+
case <-ctx.Done():
969+
return nil
970+
}
971+
}
972+
}
973+
})
974+
975+
var res *sqltypes.Result
976+
var processedAtLeastOneBatch bool
977+
eg.Go(func() (err error) {
978+
defer pan2err(&err)
979+
defer cancelF()
980+
defer wg.Done()
981+
for {
982+
if res == nil {
983+
res = &sqltypes.Result{
984+
Fields: resultFields,
985+
Rows: make([][]sqltypes.Value, 0, rowsBatch),
986+
}
987+
}
988+
if res.RowsAffected == rowsBatch {
989+
if err := callback(res, more); err != nil {
990+
return err
991+
}
992+
res = nil
993+
processedAtLeastOneBatch = true
994+
continue
995+
}
996+
997+
select {
998+
case <-ctx.Done():
999+
return context.Cause(ctx)
1000+
case <-timer.C:
1001+
if h.readTimeout != 0 {
1002+
// Cancel and return so Vitess can call the CloseConnection callback
1003+
ctx.GetLogger().Tracef("connection timeout")
1004+
return ErrRowTimeout.New()
1005+
}
1006+
case rowFrame, ok := <-rowFrameChan:
1007+
if !ok {
1008+
return nil
1009+
}
1010+
//panic(fmt.Sprintf("TESTING: %v", rowFrame.Types))
1011+
row := rowFrame
1012+
resRow := make([]sqltypes.Value, len(row))
1013+
for i, val := range row {
1014+
resRow[i] = sqltypes.MakeTrusted(val.Typ, val.Val)
1015+
}
1016+
panic("received?")
1017+
ctx.GetLogger().Tracef("spooling result row %s", resRow)
1018+
res.Rows = append(res.Rows, resRow)
8731019
res.RowsAffected++
8741020
if !timer.Stop() {
8751021
<-timer.C

‎sql/convert_value.go‎

Lines changed: 62 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -4,45 +4,88 @@ import (
44
"fmt"
55

66
"github.com/dolthub/go-mysql-server/sql/values"
7-
8-
"github.com/dolthub/vitess/go/sqltypes"
97
"github.com/dolthub/vitess/go/vt/proto/query"
108
)
119

1210
// ConvertToValue converts the interface to a sql value.
13-
func ConvertToValue(v interface{}) (sqltypes.Value, error) {
11+
func ConvertToValue(v interface{}) (Value, error) {
1412
switch v := v.(type) {
1513
case nil:
16-
return sqltypes.MakeTrusted(query.Type_NULL_TYPE, nil), nil
14+
return Value{
15+
Typ: query.Type_NULL_TYPE,
16+
Val: nil,
17+
}, nil
1718
case int:
18-
return sqltypes.MakeTrusted(query.Type_INT64, values.WriteInt64(make([]byte, values.Int64Size), int64(v))), nil
19+
return Value{
20+
Typ: query.Type_INT64,
21+
Val: values.WriteInt64(make([]byte, values.Int64Size), int64(v)),
22+
}, nil
1923
case int8:
20-
return sqltypes.MakeTrusted(query.Type_INT8, values.WriteInt8(make([]byte, values.Int8Size), v)), nil
24+
return Value{
25+
Typ: query.Type_INT8,
26+
Val: values.WriteInt8(make([]byte, values.Int8Size), v),
27+
}, nil
2128
case int16:
22-
return sqltypes.MakeTrusted(query.Type_INT16, values.WriteInt16(make([]byte, values.Int16Size), v)), nil
29+
return Value{
30+
Typ: query.Type_INT16,
31+
Val: values.WriteInt16(make([]byte, values.Int16Size), v),
32+
}, nil
2333
case int32:
24-
return sqltypes.MakeTrusted(query.Type_INT32, values.WriteInt32(make([]byte, values.Int32Size), v)), nil
34+
return Value{
35+
Typ: query.Type_INT32,
36+
Val: values.WriteInt32(make([]byte, values.Int32Size), v),
37+
}, nil
2538
case int64:
26-
return sqltypes.MakeTrusted(query.Type_INT64, values.WriteInt64(make([]byte, values.Int64Size), v)), nil
39+
return Value{
40+
Typ: query.Type_INT64,
41+
Val: values.WriteInt64(make([]byte, values.Int64Size), v),
42+
}, nil
2743
case uint:
28-
return sqltypes.MakeTrusted(query.Type_UINT64, values.WriteUint64(make([]byte, values.Uint64Size), uint64(v))), nil
44+
return Value{
45+
Typ: query.Type_UINT64,
46+
Val: values.WriteUint64(make([]byte, values.Uint64Size), uint64(v)),
47+
}, nil
2948
case uint8:
30-
return sqltypes.MakeTrusted(query.Type_UINT8, values.WriteUint8(make([]byte, values.Uint8Size), v)), nil
49+
return Value{
50+
Typ: query.Type_UINT8,
51+
Val: values.WriteUint8(make([]byte, values.Uint8Size), v),
52+
}, nil
3153
case uint16:
32-
return sqltypes.MakeTrusted(query.Type_UINT16, values.WriteUint16(make([]byte, values.Uint16Size), v)), nil
54+
return Value{
55+
Typ: query.Type_UINT16,
56+
Val: values.WriteUint16(make([]byte, values.Uint16Size), v),
57+
}, nil
3358
case uint32:
34-
return sqltypes.MakeTrusted(query.Type_UINT32, values.WriteUint32(make([]byte, values.Uint32Size), v)), nil
59+
return Value{
60+
Typ: query.Type_UINT32,
61+
Val: values.WriteUint32(make([]byte, values.Uint32Size), v),
62+
}, nil
3563
case uint64:
36-
return sqltypes.MakeTrusted(query.Type_UINT64, values.WriteUint64(make([]byte, values.Uint64Size), v)), nil
64+
return Value{
65+
Typ: query.Type_UINT64,
66+
Val: values.WriteUint64(make([]byte, values.Uint64Size), v),
67+
}, nil
3768
case float32:
38-
return sqltypes.MakeTrusted(query.Type_FLOAT32, values.WriteFloat32(make([]byte, values.Float32Size), v)), nil
69+
return Value{
70+
Typ: query.Type_FLOAT32,
71+
Val: values.WriteFloat32(make([]byte, values.Float32Size), v),
72+
}, nil
3973
case float64:
40-
return sqltypes.MakeTrusted(query.Type_FLOAT64, values.WriteFloat64(make([]byte, values.Float64Size), v)), nil
74+
return Value{
75+
Typ: query.Type_FLOAT64,
76+
Val: values.WriteFloat64(make([]byte, values.Float64Size), v),
77+
}, nil
4178
case string:
42-
return sqltypes.MakeTrusted(query.Type_VARCHAR, values.WriteString(make([]byte, len(v)), v, values.ByteOrderCollation)), nil
79+
return Value{
80+
Typ: query.Type_VARCHAR,
81+
Val: values.WriteString(make([]byte, len(v)), v, values.ByteOrderCollation),
82+
}, nil
4383
case []byte:
44-
return sqltypes.MakeTrusted(query.Type_BLOB, values.WriteBytes(make([]byte, len(v)), v, values.ByteOrderCollation)), nil
84+
return Value{
85+
Typ: query.Type_BLOB,
86+
Val: values.WriteBytes(make([]byte, len(v)), v, values.ByteOrderCollation),
87+
}, nil
4588
default:
46-
return sqltypes.Value{}, fmt.Errorf("type %T not implemented", v)
89+
return Value{}, fmt.Errorf("type %T not implemented", v)
4790
}
4891
}

‎sql/core.go‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@ import (
2626
"time"
2727
"unsafe"
2828

29-
"github.com/dolthub/vitess/go/sqltypes"
3029
"github.com/shopspring/decimal"
3130
"gopkg.in/src-d/go-errors.v1"
3231

@@ -465,7 +464,7 @@ func DebugString(nodeOrExpression interface{}) string {
465464
type Expression2 interface {
466465
Expression
467466
// Eval2 evaluates the given row frame and returns a result.
468-
Eval2(ctx *Context, row Row2) (sqltypes.Value, error)
467+
Eval2(ctx *Context, row Row2) (Value, error)
469468
// Type2 returns the expression type.
470469
Type2() Type2
471470
IsExpr2() bool

‎sql/expression/comparison.go‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@ package expression
1717
import (
1818
"fmt"
1919

20-
"github.com/dolthub/vitess/go/sqltypes"
2120
querypb "github.com/dolthub/vitess/go/vt/proto/query"
2221
errors "gopkg.in/src-d/go-errors.v1"
2322

@@ -521,7 +520,7 @@ func (gt *GreaterThan) Eval(ctx *sql.Context, row sql.Row) (interface{}, error)
521520
return result == 1, nil
522521
}
523522

524-
func (gt *GreaterThan) Eval2(ctx *sql.Context, row sql.Row2) (sqltypes.Value, error) {
523+
func (gt *GreaterThan) Eval2(ctx *sql.Context, row sql.Row2) (sql.Value, error) {
525524
l, ok := gt.Left().(sql.Expression2)
526525
if !ok {
527526
panic(fmt.Sprintf("%T does not implement sql.Expression2", gt.Left()))
@@ -533,28 +532,30 @@ func (gt *GreaterThan) Eval2(ctx *sql.Context, row sql.Row2) (sqltypes.Value, er
533532

534533
lv, err := l.Eval2(ctx, row)
535534
if err != nil {
536-
return sqltypes.Value{}, err
535+
return sql.Value{}, err
537536
}
538537
rv, err := r.Eval2(ctx, row)
539538
if err != nil {
540-
return sqltypes.Value{}, err
539+
return sql.Value{}, err
541540
}
542541

543542
// TODO: just assume they are int64
544543
l64, err := types.ConvertValueToInt64(types.NumberTypeImpl_{}, lv)
545544
if err != nil {
546-
return sqltypes.Value{}, err
545+
return sql.Value{}, err
547546
}
548547
r64, err := types.ConvertValueToInt64(types.NumberTypeImpl_{}, rv)
549548
if err != nil {
550-
return sqltypes.Value{}, err
549+
return sql.Value{}, err
551550
}
552551
var rb byte
553552
if l64 > r64 {
554553
rb = 1
555554
}
556-
557-
ret := sqltypes.MakeTrusted(querypb.Type_INT8, []byte{rb})
555+
ret := sql.Value{
556+
Val: []byte{rb},
557+
Typ: querypb.Type_INT8,
558+
}
558559
return ret, nil
559560
}
560561

‎sql/expression/get_field.go‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@ import (
1818
"fmt"
1919
"strings"
2020

21-
"github.com/dolthub/vitess/go/sqltypes"
2221
errors "gopkg.in/src-d/go-errors.v1"
2322

2423
"github.com/dolthub/go-mysql-server/sql"
@@ -150,9 +149,9 @@ func (p *GetField) Eval(ctx *sql.Context, row sql.Row) (interface{}, error) {
150149
return row[p.fieldIndex], nil
151150
}
152151

153-
func (p *GetField) Eval2(ctx *sql.Context, row sql.Row2) (sqltypes.Value, error) {
152+
func (p *GetField) Eval2(ctx *sql.Context, row sql.Row2) (sql.Value, error) {
154153
if p.fieldIndex < 0 || p.fieldIndex >= row.Len() {
155-
return sqltypes.Value{}, ErrIndexOutOfBounds.New(p.fieldIndex, row.Len())
154+
return sql.Value{}, ErrIndexOutOfBounds.New(p.fieldIndex, row.Len())
156155
}
157156
return row[p.fieldIndex], nil
158157
}

‎sql/expression/literal.go‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@ import (
1818
"fmt"
1919
"strings"
2020

21-
"github.com/dolthub/vitess/go/sqltypes"
2221
"github.com/dolthub/vitess/go/vt/proto/query"
2322
"github.com/dolthub/vitess/go/vt/sqlparser"
2423
"github.com/shopspring/decimal"
@@ -31,7 +30,7 @@ import (
3130
type Literal struct {
3231
Val interface{}
3332
Typ sql.Type
34-
val2 sqltypes.Value
33+
val2 sql.Value
3534
}
3635

3736
var _ sql.Expression = &Literal{}
@@ -137,7 +136,7 @@ func (*Literal) Children() []sql.Expression {
137136
return nil
138137
}
139138

140-
func (lit *Literal) Eval2(ctx *sql.Context, row sql.Row2) (sqltypes.Value, error) {
139+
func (lit *Literal) Eval2(ctx *sql.Context, row sql.Row2) (sql.Value, error) {
141140
return lit.val2, nil
142141
}
143142

0 commit comments

Comments
 (0)