wip

lib/logstorage/pipe_copy.go

@@ -32,7 +32,7 @@ func (pc *pipeCopy) String() string {
 }
 
 func (pc *pipeCopy) updateNeededFields(neededFields, unneededFields fieldsSet) {
-	for i := len(pc.srcFields)-1; i >=0 ; i-- {
+	for i := len(pc.srcFields) - 1; i >= 0; i-- {
 		srcField := pc.srcFields[i]
 		dstField := pc.dstFields[i]
 

lib/logstorage/pipe_delete_test.go

@@ -4,6 +4,107 @@ import (
 	"testing"
 )
 
+func TestParsePipeDeleteSuccess(t *testing.T) {
+	f := func(pipeStr string) {
+		t.Helper()
+		expectParsePipeSuccess(t, pipeStr)
+	}
+
+	f(`delete f1`)
+	f(`delete f1, f2`)
+}
+
+func TestParsePipeDeleteFailure(t *testing.T) {
+	f := func(pipeStr string) {
+		t.Helper()
+		expectParsePipeFailure(t, pipeStr)
+	}
+
+	f(`delete`)
+	f(`delete x y`)
+}
+
+func TestPipeDelete(t *testing.T) {
+	f := func(pipeStr string, rows, rowsExpected [][]Field) {
+		t.Helper()
+		expectPipeResults(t, pipeStr, rows, rowsExpected)
+	}
+
+	// single row, drop existing field
+	f("delete _msg", [][]Field{
+		{
+			{"_msg", `{"foo":"bar"}`},
+			{"a", `test`},
+		},
+	}, [][]Field{
+		{
+			{"a", `test`},
+		},
+	})
+
+	// single row, drop existing field multiple times
+	f("delete _msg, _msg", [][]Field{
+		{
+			{"_msg", `{"foo":"bar"}`},
+			{"a", `test`},
+		},
+	}, [][]Field{
+		{
+			{"a", `test`},
+		},
+	})
+
+	// single row, drop all the fields
+	f("delete a, _msg", [][]Field{
+		{
+			{"_msg", `{"foo":"bar"}`},
+			{"a", `test`},
+		},
+	}, [][]Field{})
+
+	// delete non-existing fields
+	f("delete foo, _msg, bar", [][]Field{
+		{
+			{"_msg", `{"foo":"bar"}`},
+			{"a", `test`},
+		},
+	}, [][]Field{
+		{
+			{"a", `test`},
+		},
+	})
+
+	// Multiple rows
+	f("delete _msg, a", [][]Field{
+		{
+			{"_msg", `{"foo":"bar"}`},
+			{"a", `test`},
+		},
+		{
+			{"a", `foobar`},
+		},
+		{
+			{"b", `baz`},
+			{"c", "d"},
+			{"e", "afdf"},
+		},
+		{
+			{"c", "dss"},
+			{"b", "df"},
+		},
+	}, [][]Field{
+		{
+			{"b", `baz`},
+			{"c", "d"},
+			{"e", "afdf"},
+		},
+		{
+			{"c", "dss"},
+			{"b", "df"},
+		},
+	})
+}
+
 func TestPipeDeleteUpdateNeededFields(t *testing.T) {
 	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()

lib/logstorage/pipe_fields_test.go

@@ -56,6 +56,19 @@ func TestPipeFields(t *testing.T) {
 		},
 	})
 
+	// single row, no existing fields
+	f("fields x, y", [][]Field{
+		{
+			{"_msg", `{"foo":"bar"}`},
+			{"a", `test`},
+		},
+	}, [][]Field{
+		{
+			{"x", ``},
+			{"y", ``},
+		},
+	})
+
 	// single row, mention existing field multiple times
 	f("fileds a, a", [][]Field{
 		{

lib/logstorage/pipe_rename.go

@@ -32,7 +32,7 @@ func (pr *pipeRename) String() string {
 }
 
 func (pr *pipeRename) updateNeededFields(neededFields, unneededFields fieldsSet) {
-	for i := len(pr.srcFields)-1; i >=0 ; i-- {
+	for i := len(pr.srcFields) - 1; i >= 0; i-- {
 		srcField := pr.srcFields[i]
 		dstField := pr.dstFields[i]