diff --git a/field.go b/field.go index dee36745d..5f147d735 100644 --- a/field.go +++ b/field.go @@ -935,7 +935,7 @@ func (f *Field) RowTime(tx Tx, rowID uint64, time time.Time, quantum string) (*R if !TimeQuantum(quantum).Valid() { return nil, ErrInvalidTimeQuantum } - viewname := viewsByTime(viewStandard, time, TimeQuantum(quantum[len(quantum)-1:]))[0] + viewname := viewByTimeUnit(viewStandard, time, rune(quantum[len(quantum)-1])) view := f.view(viewname) if view == nil { return nil, errors.Errorf("view with quantum %v not found.", quantum) @@ -1492,20 +1492,38 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, fieldType := f.Type() // Split import data by fragment. - views := make(map[string]int) - var allData []importData - see := func(name string, columnID uint64, rowID uint64) { + views := make(map[string]*importData) + var timeStringBuf []byte + var timeViews [][]byte + if len(q) > 0 { + // We're supporting time quantums, so we need to store bits in a + // number of views for every entry with a timestamp. We want to compute + // time quantum view names for whatever combination of YMDH views + // we have. But we don't want to allocate four strings per entry, or + // recompute and recreate the entire string. We know that only the + // YYYYMMDDHH part of the string changes over time. + timeStringBuf = make([]byte, len(viewStandard) + 11) + copy(timeStringBuf, []byte(viewStandard)) + copy(timeStringBuf[len(viewStandard):], []byte("_YYYYMMDDHH")) + // Now we have a buffer that contains + // `standard_YYYYMMDDHH`. We also need storage space to hold several + // slice headers, one per entry in q. These will hold the view names + // corresponding to each letter in q. + timeViews = make([][]byte, len(q)) + } + // This helper function records that a given column/row pair is relevant + // to a specific view. We use a map lookup for the strings, but do the + // actual operations using a slice so we're only writing each map entry + // once, not once on every update. + see := func(name []byte, columnID uint64, rowID uint64) { var ok bool - var idx int - if idx, ok = views[name]; !ok { - allData = append(allData, importData{}) - idx = len(allData) - views[name] = idx + var data *importData + if data, ok = views[string(name)]; !ok { + data = &importData{} + views[string(name)] = data } - data := allData[idx] data.RowIDs = append(data.RowIDs, rowID) data.ColumnIDs = append(data.ColumnIDs, columnID) - allData[idx] = data } for i := range rowIDs { rowID, columnID := rowIDs[i], columnIDs[i] @@ -1520,13 +1538,15 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, // attach bit to standard view unless we have a timestamp and // have the NoStandardView option set if !hasTime || !f.options.NoStandardView { - see(viewStandard, columnID, rowID) + see([]byte(viewStandard), columnID, rowID) } if hasTime { - // attach bit to all the views for this timestamp - views := viewsByTime(viewStandard, time.Unix(0, timestamps[i]).UTC(), q) - for _, view := range views { - see(view, columnID, rowID) + // attach bit to all the views for this timestamp. note that the + // `timeViews` slice gets resliced and reused by this process, so + // we don't have to allocate millions of tiny slices of slice headers. + timeViews = viewsByTimeInto(timeStringBuf, timeViews, time.Unix(0, timestamps[i]).UTC(), q) + for _, v := range timeViews { + see(v, columnID, rowID) } } } @@ -1536,8 +1556,7 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, } var err1 error defer finisher(&err1) - for viewName, idx := range views { - data := allData[idx] + for viewName, data := range views { view, err := f.createViewIfNotExists(viewName) if err != nil { return errors.Wrapf(err, "creating view %s", viewName) diff --git a/time.go b/time.go index a35ba7dd4..a17d37726 100644 --- a/time.go +++ b/time.go @@ -89,15 +89,69 @@ func viewByTimeUnit(name string, t time.Time, unit rune) string { } } +// YYYYMMDDHH lengths. Note that this is a []int, not a map[byte]int, so +// the lookups can be cheaper. +var lengthsByQuantum = []int{ + 'Y': 4, + 'M': 6, + 'D': 8, + 'H': 10, +} + +// viewsByTimeInto computes the list of views for a given time. It expects +// to be given an initial buffer of the form `name_YYYYMMDDHH`, and a slice +// of []bytes. This allows us to reuse the buffer for all the sub-buffers, +// and also to reuse the slice of slices, to eliminate all those allocations. +// This might seem crazy, but even including the JSON parsing and all the +// disk activity, the straightforward viewsByTime implementation was 25% +// of runtime in an ingest test. +func viewsByTimeInto(fullBuf []byte, into [][]byte, t time.Time, q TimeQuantum) [][]byte { + l := len(fullBuf) - 10 + date := fullBuf[l : l+10] + y, m, d := t.Date() + h := t.Hour() + // Did you know that Sprintf, Printf, and other things like that all + // do allocations, and that doing allocations in a tight loop like this + // is stunningly expensive? viewsByTime was 25% of an ingest test's + // total CPU, not counting the garbage collector overhead. This is about + // 3%. No, I'm not totally sure that justifies it. + if y < 1000 { + ys := fmt.Sprintf("%04d", y) + copy(date[0:4], []byte(ys)) + } else if y >= 10000 { + // This is probably a bad answer but there isn't really a + // good answer. + ys := fmt.Sprintf("%04d", y%1000) + copy(date[0:4], []byte(ys)) + } else { + strconv.AppendInt(date[:0], int64(y), 10) + } + date[4] = '0' + byte(m/10) + date[5] = '0' + byte(m%10) + date[6] = '0' + byte(d/10) + date[7] = '0' + byte(d%10) + date[8] = '0' + byte(h/10) + date[9] = '0' + byte(h%10) + into = into[:0] + for _, unit := range q { + if int(unit) < len(lengthsByQuantum) && lengthsByQuantum[unit] != 0 { + into = append(into, fullBuf[:l+lengthsByQuantum[unit]]) + } + } + return into +} + // viewsByTime returns a list of views for a given timestamp. func viewsByTime(name string, t time.Time, q TimeQuantum) []string { // nolint: unparam + y, m, d := t.Date() + h := t.Hour() + full := fmt.Sprintf("%s_%04d%02d%02d%02d", name, y, m, d, h) + l := len(name) + 1 a := make([]string, 0, len(q)) for _, unit := range q { - view := viewByTimeUnit(name, t, unit) - if view == "" { - continue + if int(unit) < len(lengthsByQuantum) && lengthsByQuantum[unit] != 0 { + a = append(a, full[:l+lengthsByQuantum[unit]]) } - a = append(a, view) } return a } diff --git a/time_internal_test.go b/time_internal_test.go index 8356d5dbe..2627ec65d 100644 --- a/time_internal_test.go +++ b/time_internal_test.go @@ -83,6 +83,38 @@ func TestViewsByTime(t *testing.T) { }) } +func TestViewsByTimeInto(t *testing.T) { + ts := time.Date(2000, time.January, 2, 3, 4, 5, 6, time.UTC) + s := []byte("F_YYYYMMDDHH") + var timeViews [][]byte + + t.Run("YMDH", func(t *testing.T) { + a := viewsByTime("F", ts, mustParseTimeQuantum("YMDH")) + b := viewsByTimeInto(s, timeViews, ts, mustParseTimeQuantum("YMDH")) + if len(a) != len(b) { + t.Fatalf("mismatch: viewsByTime: %q, viewsByTimeInto: %q", a, b) + } + for i := range a { + if a[i] != string(b[i]) { + t.Fatalf("mismatch: viewsByTime: %q, viewsByTimeInto: %q", a, b) + } + } + }) + + t.Run("D", func(t *testing.T) { + a := viewsByTime("F", ts, mustParseTimeQuantum("D")) + b := viewsByTimeInto(s, timeViews, ts, mustParseTimeQuantum("D")) + if len(a) != len(b) { + t.Fatalf("mismatch: viewsByTime: %q, viewsByTimeInto: %q", a, b) + } + for i := range a { + if a[i] != string(b[i]) { + t.Fatalf("mismatch: viewsByTime: %q, viewsByTimeInto: %q", a, b) + } + } + }) +} + // Ensure sets of fields can be returned for a given time range. func TestViewsByTimeRange(t *testing.T) { t.Run("Y", func(t *testing.T) {