diff --git a/lib/logstorage/pipe_copy_test.go b/lib/logstorage/pipe_copy_test.go
index c5dc5ed0f..c9e30ae22 100644
--- a/lib/logstorage/pipe_copy_test.go
+++ b/lib/logstorage/pipe_copy_test.go
@@ -6,20 +6,9 @@ import (
 )
 
 func TestPipeCopyUpdateNeededFields(t *testing.T) {
-	f := func(s string, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
+	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeCopy(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
@@ -53,6 +42,22 @@ func TestPipeCopyUpdateNeededFields(t *testing.T) {
 	f("copy s1 d1, s2 d2", "s2,d1,f1,f2", "", "s1,s2,f1,f2", "")
 }
 
+func expectPipeNeededFields(t *testing.T, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
+	t.Helper()
+
+	nfs := newTestFieldsSet(neededFields)
+	unfs := newTestFieldsSet(unneededFields)
+
+	lex := newLexer(s)
+	p, err := parsePipe(lex)
+	if err != nil {
+		t.Fatalf("cannot parse %s: %s", s, err)
+	}
+	p.updateNeededFields(nfs, unfs)
+
+	assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+}
+
 func assertNeededFields(t *testing.T, nfs, unfs fieldsSet, neededFieldsExpected, unneededFieldsExpected string) {
 	t.Helper()
 
diff --git a/lib/logstorage/pipe_delete_test.go b/lib/logstorage/pipe_delete_test.go
index f3db305f5..05c718dea 100644
--- a/lib/logstorage/pipe_delete_test.go
+++ b/lib/logstorage/pipe_delete_test.go
@@ -7,18 +7,7 @@ import (
 func TestPipeDeleteUpdateNeededFields(t *testing.T) {
 	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeDelete(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_extract_test.go b/lib/logstorage/pipe_extract_test.go
index ad6f0f3ea..b8c21ff0a 100644
--- a/lib/logstorage/pipe_extract_test.go
+++ b/lib/logstorage/pipe_extract_test.go
@@ -181,18 +181,7 @@ func TestParseExtractFormatStepFailure(t *testing.T) {
 func TestPipeExtractUpdateNeededFields(t *testing.T) {
 	f := func(s string, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeExtract(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_field_names_test.go b/lib/logstorage/pipe_field_names_test.go
index df8d68c48..4b19f2e34 100644
--- a/lib/logstorage/pipe_field_names_test.go
+++ b/lib/logstorage/pipe_field_names_test.go
@@ -7,18 +7,7 @@ import (
 func TestPipeFieldNamesUpdateNeededFields(t *testing.T) {
 	f := func(s string, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeFieldNames(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_fields_test.go b/lib/logstorage/pipe_fields_test.go
index 0dcc7bf39..a7b1e792b 100644
--- a/lib/logstorage/pipe_fields_test.go
+++ b/lib/logstorage/pipe_fields_test.go
@@ -7,18 +7,7 @@ import (
 func TestPipeFieldsUpdateNeededFields(t *testing.T) {
 	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeFields(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_filter_test.go b/lib/logstorage/pipe_filter_test.go
index d4d5e3ff9..73da2c8bb 100644
--- a/lib/logstorage/pipe_filter_test.go
+++ b/lib/logstorage/pipe_filter_test.go
@@ -7,18 +7,7 @@ import (
 func TestPipeFilterUpdateNeededFields(t *testing.T) {
 	f := func(s string, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeFilter(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_limit_test.go b/lib/logstorage/pipe_limit_test.go
index b19a567e0..49e79d5d2 100644
--- a/lib/logstorage/pipe_limit_test.go
+++ b/lib/logstorage/pipe_limit_test.go
@@ -7,18 +7,7 @@ import (
 func TestPipeLimitUpdateNeededFields(t *testing.T) {
 	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeLimit(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_offset_test.go b/lib/logstorage/pipe_offset_test.go
index 63eba3407..53b22af61 100644
--- a/lib/logstorage/pipe_offset_test.go
+++ b/lib/logstorage/pipe_offset_test.go
@@ -7,18 +7,7 @@ import (
 func TestPipeOffsetUpdateNeededFields(t *testing.T) {
 	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeOffset(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_rename_test.go b/lib/logstorage/pipe_rename_test.go
index 5011e3333..506a2f5fa 100644
--- a/lib/logstorage/pipe_rename_test.go
+++ b/lib/logstorage/pipe_rename_test.go
@@ -7,18 +7,7 @@ import (
 func TestPipeRenameUpdateNeededFields(t *testing.T) {
 	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeRename(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_sort_test.go b/lib/logstorage/pipe_sort_test.go
index 2111409f0..b2bb0aef8 100644
--- a/lib/logstorage/pipe_sort_test.go
+++ b/lib/logstorage/pipe_sort_test.go
@@ -7,18 +7,7 @@ import (
 func TestPipeSortUpdateNeededFields(t *testing.T) {
 	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeSort(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_stats_test.go b/lib/logstorage/pipe_stats_test.go
index c5a3f5481..51837eba6 100644
--- a/lib/logstorage/pipe_stats_test.go
+++ b/lib/logstorage/pipe_stats_test.go
@@ -7,18 +7,7 @@ import (
 func TestPipeStatsUpdateNeededFields(t *testing.T) {
 	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeStats(lex)
-		if err != nil {
-			t.Fatalf("unexpected error when parsing %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_uniq_test.go b/lib/logstorage/pipe_uniq_test.go
index 6896c3b04..b88a2697e 100644
--- a/lib/logstorage/pipe_uniq_test.go
+++ b/lib/logstorage/pipe_uniq_test.go
@@ -7,18 +7,7 @@ import (
 func TestPipeUniqUpdateNeededFields(t *testing.T) {
 	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeUniq(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields
diff --git a/lib/logstorage/pipe_unpack_json_test.go b/lib/logstorage/pipe_unpack_json_test.go
index 2caf44caf..b3a02da64 100644
--- a/lib/logstorage/pipe_unpack_json_test.go
+++ b/lib/logstorage/pipe_unpack_json_test.go
@@ -326,18 +326,7 @@ func rowToString(row []Field) string {
 func TestPipeUnpackJSONUpdateNeededFields(t *testing.T) {
 	f := func(s string, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
 		t.Helper()
-
-		nfs := newTestFieldsSet(neededFields)
-		unfs := newTestFieldsSet(unneededFields)
-
-		lex := newLexer(s)
-		p, err := parsePipeUnpackJSON(lex)
-		if err != nil {
-			t.Fatalf("cannot parse %s: %s", s, err)
-		}
-		p.updateNeededFields(nfs, unfs)
-
-		assertNeededFields(t, nfs, unfs, neededFieldsExpected, unneededFieldsExpected)
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
 	}
 
 	// all the needed fields