Merge pull request #6732 from influxdata/dn-batch-cq-writes2

batch SELECT INTO writes
pull/6766/head
David Norton 2016-06-01 13:24:24 -04:00
commit 1f390ee290
3 changed files with 213 additions and 7 deletions

View File

@ -52,6 +52,7 @@
- [#6727](https://github.com/influxdata/influxdb/issues/6727): queries with strings that look like dates end up with date types, not string types
- [#6250](https://github.com/influxdata/influxdb/issues/6250): Slow startup time
- [#6753](https://github.com/influxdata/influxdb/issues/6753): Prevent panic if there are no values.
- [#6685](https://github.com/influxdata/influxdb/issues/6685): Batch SELECT INTO / CQ writes
## v0.13.0 [2016-05-12]

View File

@ -251,6 +251,124 @@ func TestPointsWriter_WritePoints(t *testing.T) {
}
}
type fakePointsWriter struct {
WritePointsIntoFn func(*coordinator.IntoWriteRequest) error
}
func (f *fakePointsWriter) WritePointsInto(req *coordinator.IntoWriteRequest) error {
return f.WritePointsIntoFn(req)
}
func TestBufferedPointsWriter(t *testing.T) {
db := "db0"
rp := "rp0"
capacity := 10000
writePointsIntoCnt := 0
pointsWritten := []models.Point{}
reset := func() {
writePointsIntoCnt = 0
pointsWritten = pointsWritten[:0]
}
fakeWriter := &fakePointsWriter{
WritePointsIntoFn: func(req *coordinator.IntoWriteRequest) error {
writePointsIntoCnt++
pointsWritten = append(pointsWritten, req.Points...)
return nil
},
}
w := coordinator.NewBufferedPointsWriter(fakeWriter, db, rp, capacity)
// Test that capacity and length are correct for new buffered writer.
if w.Cap() != capacity {
t.Fatalf("exp %d, got %d", capacity, w.Cap())
} else if w.Len() != 0 {
t.Fatalf("exp %d, got %d", 0, w.Len())
}
// Test flushing an empty buffer.
if err := w.Flush(); err != nil {
t.Fatal(err)
} else if writePointsIntoCnt > 0 {
t.Fatalf("exp 0, got %d", writePointsIntoCnt)
}
// Test writing zero points.
if err := w.WritePointsInto(&coordinator.IntoWriteRequest{
Database: db,
RetentionPolicy: rp,
Points: []models.Point{},
}); err != nil {
t.Fatal(err)
} else if writePointsIntoCnt > 0 {
t.Fatalf("exp 0, got %d", writePointsIntoCnt)
} else if w.Len() > 0 {
t.Fatalf("exp 0, got %d", w.Len())
}
// Test writing single large bunch of points points.
req := coordinator.WritePointsRequest{
Database: db,
RetentionPolicy: rp,
}
numPoints := int(float64(capacity) * 5.5)
for i := 0; i < numPoints; i++ {
req.AddPoint("cpu", float64(i), time.Unix(0, 0).Add(time.Duration(i)*time.Second), nil)
}
r := coordinator.IntoWriteRequest(req)
if err := w.WritePointsInto(&r); err != nil {
t.Fatal(err)
} else if writePointsIntoCnt != 5 {
t.Fatalf("exp 5, got %d", writePointsIntoCnt)
} else if w.Len() != capacity/2 {
t.Fatalf("exp %d, got %d", capacity/2, w.Len())
} else if len(pointsWritten) != numPoints-capacity/2 {
t.Fatalf("exp %d, got %d", numPoints-capacity/2, len(pointsWritten))
}
if err := w.Flush(); err != nil {
t.Fatal(err)
} else if writePointsIntoCnt != 6 {
t.Fatalf("exp 6, got %d", writePointsIntoCnt)
} else if w.Len() != 0 {
t.Fatalf("exp 0, got %d", w.Len())
} else if len(pointsWritten) != numPoints {
t.Fatalf("exp %d, got %d", numPoints, len(pointsWritten))
} else if !reflect.DeepEqual(r.Points, pointsWritten) {
t.Fatal("points don't match")
}
reset()
// Test writing points one at a time.
for i, _ := range r.Points {
if err := w.WritePointsInto(&coordinator.IntoWriteRequest{
Database: db,
RetentionPolicy: rp,
Points: r.Points[i : i+1],
}); err != nil {
t.Fatal(err)
}
}
if err := w.Flush(); err != nil {
t.Fatal(err)
} else if writePointsIntoCnt != 6 {
t.Fatalf("exp 6, got %d", writePointsIntoCnt)
} else if w.Len() != 0 {
t.Fatalf("exp 0, got %d", w.Len())
} else if len(pointsWritten) != numPoints {
t.Fatalf("exp %d, got %d", numPoints, len(pointsWritten))
} else if !reflect.DeepEqual(r.Points, pointsWritten) {
t.Fatal("points don't match")
}
}
var shardID uint64
type fakeShardWriter struct {

View File

@ -17,6 +17,10 @@ import (
"github.com/influxdata/influxdb/tsdb"
)
type pointsWriter interface {
WritePointsInto(*IntoWriteRequest) error
}
// StatementExecutor executes a statement in the query.
type StatementExecutor struct {
MetaClient MetaClient
@ -28,9 +32,7 @@ type StatementExecutor struct {
Monitor *monitor.Monitor
// Used for rewriting points back into system for SELECT INTO statements.
PointsWriter interface {
WritePointsInto(*IntoWriteRequest) error
}
PointsWriter pointsWriter
// Select statement limits
MaxSelectPointN int
@ -507,6 +509,12 @@ func (e *StatementExecutor) executeSelectStatement(stmt *influxql.SelectStatemen
// Emit rows to the results channel.
var writeN int64
var emitted bool
var pointsWriter *BufferedPointsWriter
if stmt.Target != nil {
pointsWriter = NewBufferedPointsWriter(e.PointsWriter, stmt.Target.Measurement.Database, stmt.Target.Measurement.RetentionPolicy, 10000)
}
for {
row, err := em.Emit()
if err != nil {
@ -523,7 +531,7 @@ func (e *StatementExecutor) executeSelectStatement(stmt *influxql.SelectStatemen
// Write points back into system for INTO statements.
if stmt.Target != nil {
if err := e.writeInto(stmt, row); err != nil {
if err := e.writeInto(pointsWriter, stmt, row); err != nil {
return err
}
writeN += int64(len(row.Values))
@ -545,8 +553,12 @@ func (e *StatementExecutor) executeSelectStatement(stmt *influxql.SelectStatemen
emitted = true
}
// Emit write count if an INTO statement.
// Flush remaing points and emit write count if an INTO statement.
if stmt.Target != nil {
if err := pointsWriter.Flush(); err != nil {
return err
}
var messages []*influxql.Message
if ctx.ReadOnly {
messages = append(messages, influxql.ReadOnlyWarning(stmt.String()))
@ -779,7 +791,82 @@ func (e *StatementExecutor) executeShowUsersStatement(q *influxql.ShowUsersState
return []*models.Row{row}, nil
}
func (e *StatementExecutor) writeInto(stmt *influxql.SelectStatement, row *models.Row) error {
type BufferedPointsWriter struct {
w pointsWriter
buf []models.Point
database string
retentionPolicy string
}
func NewBufferedPointsWriter(w pointsWriter, database, retentionPolicy string, capacity int) *BufferedPointsWriter {
return &BufferedPointsWriter{
w: w,
buf: make([]models.Point, 0, capacity),
database: database,
retentionPolicy: retentionPolicy,
}
}
func (w *BufferedPointsWriter) WritePointsInto(req *IntoWriteRequest) error {
// Make sure we're buffering points only for the expected destination.
if req.Database != w.database || req.RetentionPolicy != w.retentionPolicy {
return fmt.Errorf("writer for %s.%s can't write into %s.%s", w.database, w.retentionPolicy, req.Database, req.RetentionPolicy)
}
for i := 0; i < len(req.Points); {
// Get the available space in the buffer.
avail := cap(w.buf) - len(w.buf)
// Calculate number of points to copy into the buffer.
n := len(req.Points[i:])
if n > avail {
n = avail
}
// Copy points into buffer.
w.buf = append(w.buf, req.Points[i:n+i]...)
// Advance the index by number of points copied.
i += n
// If buffer is full, flush points to underlying writer.
if len(w.buf) == cap(w.buf) {
if err := w.Flush(); err != nil {
return err
}
}
}
return nil
}
// Flush writes all buffered points to the underlying writer.
func (w *BufferedPointsWriter) Flush() error {
if len(w.buf) == 0 {
return nil
}
if err := w.w.WritePointsInto(&IntoWriteRequest{
Database: w.database,
RetentionPolicy: w.retentionPolicy,
Points: w.buf,
}); err != nil {
return err
}
// Clear the buffer.
w.buf = w.buf[:0]
return nil
}
// Len returns the number of points buffered.
func (w *BufferedPointsWriter) Len() int { return len(w.buf) }
// Cap returns the capacity (in points) of the buffer.
func (w *BufferedPointsWriter) Cap() int { return cap(w.buf) }
func (e *StatementExecutor) writeInto(w pointsWriter, stmt *influxql.SelectStatement, row *models.Row) error {
if stmt.Target.Measurement.Database == "" {
return errNoDatabaseInTarget
}
@ -801,7 +888,7 @@ func (e *StatementExecutor) writeInto(stmt *influxql.SelectStatement, row *model
return err
}
if err := e.PointsWriter.WritePointsInto(&IntoWriteRequest{
if err := w.WritePointsInto(&IntoWriteRequest{
Database: stmt.Target.Measurement.Database,
RetentionPolicy: stmt.Target.Measurement.RetentionPolicy,
Points: points,