overwrite fix

integration/libs/filer/filer_test.go

@@ -924,6 +924,17 @@ func TestDbfsFilerForStreamingUploads(t *testing.T) {
 
 	// Assert contents
 	filerTest{t, f}.assertContents(ctx, "foo.txt", "foobar")
+
+	// Overwrite the file with streaming upload, and fail
+	err = f.Write(ctx, "foo.txt", strings.NewReader("barfoo"))
+	require.ErrorIs(t, err, fs.ErrExist)
+
+	// Overwrite the file with streaming upload, and succeed
+	err = f.Write(ctx, "foo.txt", strings.NewReader("barfoo"), filer.OverwriteIfExists)
+	require.NoError(t, err)
+
+	// Assert contents
+	filerTest{t, f}.assertContents(ctx, "foo.txt", "barfoo")
 }
 
 func TestDbfsFilerForPutUploads(t *testing.T) {
@@ -934,15 +945,36 @@ func TestDbfsFilerForPutUploads(t *testing.T) {
 	tmpDir := t.TempDir()
 	err := os.WriteFile(filepath.Join(tmpDir, "foo.txt"), []byte("foobar"), 0o644)
 	require.NoError(t, err)
+	err = os.WriteFile(filepath.Join(tmpDir, "bar.txt"), []byte("barfoo"), 0o644)
+	require.NoError(t, err)
 
-	fd, err := os.Open(filepath.Join(tmpDir, "foo.txt"))
+	fdFoo, err := os.Open(filepath.Join(tmpDir, "foo.txt"))
 	require.NoError(t, err)
-	defer fd.Close()
+	defer fdFoo.Close()
+
+	fdBar, err := os.Open(filepath.Join(tmpDir, "bar.txt"))
+	require.NoError(t, err)
+	defer fdBar.Close()
 
 	// Write a file with PUT upload
-	err = f.Write(ctx, "foo.txt", fd)
+	err = f.Write(ctx, "foo.txt", fdFoo)
 	require.NoError(t, err)
 
 	// Assert contents
 	filerTest{t, f}.assertContents(ctx, "foo.txt", "foobar")
+
+	// Try to overwrite the file, and fail.
+	err = f.Write(ctx, "foo.txt", fdBar)
+	require.ErrorIs(t, err, fs.ErrExist)
+
+	// Reset the file descriptor.
+	_, err = fdBar.Seek(0, io.SeekStart)
+	require.NoError(t, err)
+
+	// Overwrite the file with OverwriteIfExists flag
+	err = f.Write(ctx, "foo.txt", fdBar, filer.OverwriteIfExists)
+	require.NoError(t, err)
+
+	// Assert contents
+	filerTest{t, f}.assertContents(ctx, "foo.txt", "barfoo")
 }

libs/filer/dbfs_client.go

@@ -131,11 +131,16 @@ func (w *DbfsClient) putFile(ctx context.Context, path string, overwrite bool, f
 		return err
 	}
 
-	// Request bodies of Content-Type multipart/form-data must are not supported by
+	// Request bodies of Content-Type multipart/form-data are not supported by
 	// the Go SDK directly for DBFS. So we use the Do method directly.
-	return w.apiClient.Do(ctx, http.MethodPost, "/api/2.0/dbfs/put", map[string]string{
+	err = w.apiClient.Do(ctx, http.MethodPost, "/api/2.0/dbfs/put", map[string]string{
 		"Content-Type": writer.FormDataContentType(),
 	}, buf.Bytes(), nil)
+	var aerr *apierr.APIError
+	if errors.As(err, &aerr) && aerr.ErrorCode == "RESOURCE_ALREADY_EXISTS" {
+		return FileAlreadyExistsError{path}
+	}
+	return err
 }
 
 func (w *DbfsClient) streamFile(ctx context.Context, path string, overwrite bool, reader io.Reader) error {