From c5c96d8016ed4e562cc8e0cebe3d983bd622851e Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@victoriametrics.com>
Date: Mon, 20 May 2024 21:47:30 +0200
Subject: [PATCH] wip

---
 lib/logstorage/pipe_copy.go        |   2 +-
 lib/logstorage/pipe_delete_test.go | 101 +++++++++++++++++++++++++++++
 lib/logstorage/pipe_fields_test.go |  13 ++++
 lib/logstorage/pipe_rename.go      |   2 +-
 4 files changed, 116 insertions(+), 2 deletions(-)

diff --git a/lib/logstorage/pipe_copy.go b/lib/logstorage/pipe_copy.go
index 55882d36a..8868b9c90 100644
--- a/lib/logstorage/pipe_copy.go
+++ b/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]
 
diff --git a/lib/logstorage/pipe_delete_test.go b/lib/logstorage/pipe_delete_test.go
index 05c718dea..63620b691 100644
--- a/lib/logstorage/pipe_delete_test.go
+++ b/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()
diff --git a/lib/logstorage/pipe_fields_test.go b/lib/logstorage/pipe_fields_test.go
index 000072081..7d0e6d720 100644
--- a/lib/logstorage/pipe_fields_test.go
+++ b/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{
 		{
diff --git a/lib/logstorage/pipe_rename.go b/lib/logstorage/pipe_rename.go
index 9d3427e1c..76814c3e3 100644
--- a/lib/logstorage/pipe_rename.go
+++ b/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]