Cleanup failed segment before flushing. (#2606)

* Cleanup failed segment before flushing a new one.

* Do not create a segment if there's no profile to flush

* Uses a temp dir for the test to avoid left-over.

* remove unrelated change

* Update pkg/phlaredb/profile_store.go

Co-authored-by: Christian Simon <simon@swine.de>

---------

Co-authored-by: Christian Simon <simon@swine.de>

pkg/phlaredb/profile_store.go

@@ -239,7 +239,11 @@ func (s *profileStore) cutRowGroup(count int) (err error) {
 		s.path,
 		fmt.Sprintf("%s.%d%s", s.persister.Name(), s.rowsFlushed, block.ParquetSuffix),
 	)
-
+	// Removes the file if it exists. This can happen if the previous
+	// cut attempt failed.
+	if err := os.Remove(path); err == nil {
+		level.Warn(s.logger).Log("msg", "deleting row group segment of a failed previous attempt", "path", path)
+	}
 	f, err := s.prepareFile(path)
 	if err != nil {
 		return err

pkg/phlaredb/profile_store_test.go

@@ -595,3 +595,15 @@ func TestProfileStore_Querying(t *testing.T) {
 		)
 	})
 }
+
+func TestRemoveFailedSegment(t *testing.T) {
+	store := newProfileStore(testContext(t))
+	dir := t.TempDir()
+	require.NoError(t, store.Init(dir, defaultParquetConfig, contextHeadMetrics(context.Background())))
+	// fake a failed segment
+	_, err := os.Create(dir + "/profiles.0.parquet")
+	require.NoError(t, store.ingest(context.Background(), []schemav1.InMemoryProfile{{}}, phlaremodel.LabelsFromStrings(), "memory"))
+	require.NoError(t, err)
+	err = store.cutRowGroup(1)
+	require.NoError(t, err)
+}