Merge pull request influxdata#391 from influxdata/nc-issue#390

Remove old mapReduce code

CHANGELOG.md

@@ -36,6 +36,7 @@ format a TICKscript file according to a common standard.
 
 - [#299](https://github.com/influxdata/kapacitor/issues/299): Changes TICKscript chaining method operators and adds `tickfmt` binary.
 - [#389](https://github.com/influxdata/kapacitor/pull/389): Adds benchmarks to Kapacitor for basic use cases.
+- [#390](https://github.com/influxdata/kapacitor/issues/390): BREAKING: Remove old `.mapReduce` functions.
 
 
 ### Bugfixes

edge.go

@@ -33,7 +33,6 @@ type BatchCollector interface {
 type Edge struct {
 	stream chan models.Point
 	batch  chan models.Batch
-	reduce chan *MapResult
 
 	logger     *log.Logger
 	aborted    chan struct{}
@@ -72,8 +71,6 @@ func newEdge(taskName, parentName, childName string, t pipeline.EdgeType, size i
 		e.stream = make(chan models.Point, size)
 	case pipeline.BatchEdge:
 		e.batch = make(chan models.Batch, size)
-	case pipeline.ReduceEdge:
-		e.reduce = make(chan *MapResult, size)
 	}
 	return e
 }
@@ -123,9 +120,6 @@ func (e *Edge) Close() {
 	if e.batch != nil {
 		close(e.batch)
 	}
-	if e.reduce != nil {
-		close(e.reduce)
-	}
 	DeleteStatistics(e.statsKey)
 }
 
@@ -171,17 +165,6 @@ func (e *Edge) NextBatch() (b models.Batch, ok bool) {
 	return
 }
 
-func (e *Edge) NextMaps() (m *MapResult, ok bool) {
-	select {
-	case <-e.aborted:
-	case m, ok = <-e.reduce:
-		if ok {
-			e.emitted.Add(1)
-		}
-	}
-	return
-}
-
 func (e *Edge) CollectPoint(p models.Point) error {
 	e.collected.Add(1)
 	e.incCollected(p.Group, p.Tags, p.Dimensions)
@@ -204,16 +187,6 @@ func (e *Edge) CollectBatch(b models.Batch) error {
 	}
 }
 
-func (e *Edge) CollectMaps(m *MapResult) error {
-	e.collected.Add(1)
-	select {
-	case <-e.aborted:
-		return ErrAborted
-	case e.reduce <- m:
-		return nil
-	}
-}
-
 // Increment the emitted count of the group for this edge.
 func (e *Edge) incEmitted(group models.GroupID, tags models.Tags, dims []string) {
 	e.groupMu.Lock()

integrations/streamer_test.go

@@ -1437,19 +1437,6 @@ func TestStream_InfluxQL(t *testing.T) {
 	}
 
 	var scriptTmpl = `
-stream
-	|from()
-		.measurement('cpu')
-		.where(lambda: "host" == 'serverA')
-	|window()
-		.period(10s)
-		.every(10s)
-	|mapReduce(influxql|{{ .Method }}({{ .Args }}))
-		{{ if .UsePointTimes }}.usePointTimes(){{ end }}
-	|httpOut('TestStream_InfluxQL')
-`
-
-	var newScriptTmpl = `
 stream
 	|from()
 		.measurement('cpu')
@@ -1875,35 +1862,22 @@ stream
 		t.Fatal(err)
 	}
 
-	newTmpl, err := template.New("script").Parse(newScriptTmpl)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	tmpls := []*template.Template{tmpl, newTmpl}
-
 	for _, tc := range testCases {
-		for i, tmpl := range tmpls {
-			if tc.Method == "distinct" && i == 0 {
-				// Skip legacy test for new behavior
-				continue
-			}
-			t.Log("Method:", tc.Method, i)
-			var script bytes.Buffer
-			if tc.Args == "" {
-				tc.Args = "'value'"
-			}
-			tmpl.Execute(&script, tc)
-			testStreamerWithOutput(
-				t,
-				"TestStream_InfluxQL",
-				script.String(),
-				13*time.Second,
-				tc.ER,
-				nil,
-				false,
-			)
-		}
+		t.Log("Method:", tc.Method)
+		var script bytes.Buffer
+		if tc.Args == "" {
+			tc.Args = "'value'"
+		}
+		tmpl.Execute(&script, tc)
+		testStreamerWithOutput(
+			t,
+			"TestStream_InfluxQL",
+			script.String(),
+			13*time.Second,
+			tc.ER,
+			nil,
+			false,
+		)
 	}
 }