lib/protoparser: fix skip csv line when metric can be collect from the line (#4298)

* lib/protoparser: fix skip csv line when metric can be collect from the line

* lib/protoparser: fix comment

docs/CHANGELOG.md

@@ -73,6 +73,7 @@ The following tip changes can be tested by building VictoriaMetrics components f
 * BUGFIX: [vmbackup](https://docs.victoriametrics.com/vmbackup.html): fix compatibility with Windows OS. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/70).
 * BUGFIX: [vmctl](https://docs.victoriametrics.com/vmctl.html): fix performance issue when migrating data from VictoriaMetrics according to [these docs](https://docs.victoriametrics.com/vmctl.html#migrating-data-from-victoriametrics). Add the ability to speed up the data migration via `--vm-native-disable-retries` command-line flag. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4092).
 * BUGFIX: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation.html): fix bug with duplicated labels during stream aggregation via single-node VictoriaMetrics. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4277).
+* BUGFIX: [csvimport](https://docs.victoriametrics.com/#how-to-import-csv-data): properly parse [csv line](https://docs.victoriametrics.com/#how-to-import-csv-data) when value in the last column is empty. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4048).
 
 ## [v1.90.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.90.0)
 

lib/protoparser/csvimport/parser.go

@@ -118,7 +118,8 @@ func parseRows(sc *scanner, dst []Row, tags []Tag, metrics []metric, cds []Colum
 				Value: value,
 			})
 		}
-		if col < uint(len(cds)) && sc.Error == nil {
+
+		if col < uint(len(cds)) && sc.Error == nil && len(metrics) == 0 {
 			sc.Error = fmt.Errorf("missing columns in the csv line %q; got %d columns; want at least %d columns", line, col, len(cds))
 		}
 		if sc.Error != nil {
@@ -126,9 +127,6 @@ func parseRows(sc *scanner, dst []Row, tags []Tag, metrics []metric, cds []Colum
 			invalidLines.Inc()
 			continue
 		}
-		if len(metrics) == 0 {
-			logger.Panicf("BUG: expecting at least a single metric in columnDescriptors=%#v", cds)
-		}
 		r.Metric = metrics[0].Name
 		r.Tags = tags[tagsLen:]
 		r.Value = metrics[0].Value

lib/protoparser/csvimport/parser_test.go

@@ -55,6 +55,42 @@ func TestRowsUnmarshalSuccess(t *testing.T) {
 			t.Fatalf("unexpected rows on the second unmarshal;\ngot\n%v\nwant\n%v", rs.Rows, rowsExpected)
 		}
 	}
+	f("1:label:mytest,2:time:unix_ns,3:metric:metric_1,4:metric:metric_2", "test,1677632461449998000,,", nil)
+	f("1:time:unix_ns,2:metric:metric_1,3:metric:metric_2", "1677632461449998000,,", nil)
+	f("1:time:unix_ns,2:metric:metric_1,3:metric:metric_2", "1677632461449998000,1,", []Row{
+		{
+			Metric:    "metric_1",
+			Value:     1,
+			Timestamp: 1677632461449,
+		},
+	})
+	f("1:time:unix_ns,2:metric:metric_1,3:metric:metric_2",
+		`
+1677632461449998000,1,1
+1677633061449998000,1,
+1677633661449998000,,1`, []Row{
+			{
+				Metric:    "metric_1",
+				Value:     1,
+				Timestamp: 1677632461449,
+			},
+			{
+				Metric:    "metric_2",
+				Value:     1,
+				Timestamp: 1677632461449,
+			},
+			{
+				Metric:    "metric_1",
+				Value:     1,
+				Timestamp: 1677633061449,
+			},
+			{
+				Metric:    "metric_2",
+				Value:     1,
+				Timestamp: 1677633661449,
+			},
+		})
+
 	f("1:metric:foo", "", nil)
 	f("1:metric:foo", `123`, []Row{
 		{