From a46f614e0784f8a1982068f8c5d78219c2d775ce Mon Sep 17 00:00:00 2001
From: Sutou Kouhei <kou@clear-code.com>
Date: Mon, 4 Dec 2023 17:30:08 +0900
Subject: [PATCH 1/3] GH-38705: [C++][FS][Azure] Implement CopyFile()

---
 cpp/src/arrow/filesystem/azurefs.cc      | 48 ++++++++++++++++--
 cpp/src/arrow/filesystem/azurefs_test.cc | 64 ++++++++++++++++++++++++
 2 files changed, 108 insertions(+), 4 deletions(-)

diff --git a/cpp/src/arrow/filesystem/azurefs.cc b/cpp/src/arrow/filesystem/azurefs.cc
index ecc8d06f9756..6a544dc5ed71 100644
--- a/cpp/src/arrow/filesystem/azurefs.cc
+++ b/cpp/src/arrow/filesystem/azurefs.cc
@@ -115,6 +115,10 @@ struct AzureLocation {
     return parent;
   }
 
+  Result<AzureLocation> join(const std::string& stem) const {
+    return FromString(internal::ConcatAbstractPath(all, stem));
+  }
+
   bool has_parent() const { return !path.empty(); }
 
   bool empty() const { return container.empty() && path.empty(); }
@@ -971,7 +975,7 @@ class AzureFileSystem::Impl {
   }
 
  private:
-  Status DeleteDirContentsWihtoutHierarchicalNamespace(const AzureLocation& location,
+  Status DeleteDirContentsWithoutHierarchicalNamespace(const AzureLocation& location,
                                                        bool missing_dir_ok) {
     auto container_client =
         blob_service_client_->GetBlobContainerClient(location.container);
@@ -1092,7 +1096,7 @@ class AzureFileSystem::Impl {
             exception);
       }
     } else {
-      return DeleteDirContentsWihtoutHierarchicalNamespace(location,
+      return DeleteDirContentsWithoutHierarchicalNamespace(location,
                                                            /*missing_dir_ok=*/true);
     }
   }
@@ -1149,7 +1153,41 @@ class AzureFileSystem::Impl {
       }
       return Status::OK();
     } else {
-      return DeleteDirContentsWihtoutHierarchicalNamespace(location, missing_dir_ok);
+      return DeleteDirContentsWithoutHierarchicalNamespace(location, missing_dir_ok);
+    }
+  }
+
+ private:
+  Status CopyFileInternal(const AzureLocation& src, const AzureLocation& dest) {
+    if (src == dest) {
+      return Status::OK();
+    }
+    auto dest_blob_client = blob_service_client_->GetBlobContainerClient(dest.container)
+                                .GetBlobClient(dest.path);
+    auto src_url = blob_service_client_->GetBlobContainerClient(src.container)
+                       .GetBlobClient(src.path)
+                       .GetUrl();
+    try {
+      dest_blob_client.CopyFromUri(src_url);
+    } catch (const Azure::Storage::StorageException& exception) {
+      return internal::ExceptionToStatus(
+          "Failed to copy a blob. (" + src_url + " -> " + dest_blob_client.GetUrl() + ")",
+          exception);
+    }
+    return Status::OK();
+  }
+
+ public:
+  Status CopyFile(const AzureLocation& src, const AzureLocation& dest) {
+    RETURN_NOT_OK(ValidateFileLocation(src));
+    if (dest.container.empty()) {
+      return PathNotFound(dest);
+    }
+    if (dest.path.empty() || internal::HasTrailingSlash(dest.path)) {
+      ARROW_ASSIGN_OR_RAISE(auto real_dest, dest.join(src.path_parts.back()));
+      return CopyFileInternal(src, real_dest);
+    } else {
+      return CopyFileInternal(src, dest);
     }
   }
 };
@@ -1208,7 +1246,9 @@ Status AzureFileSystem::Move(const std::string& src, const std::string& dest) {
 }
 
 Status AzureFileSystem::CopyFile(const std::string& src, const std::string& dest) {
-  return Status::NotImplemented("The Azure FileSystem is not fully implemented");
+  ARROW_ASSIGN_OR_RAISE(auto src_location, AzureLocation::FromString(src));
+  ARROW_ASSIGN_OR_RAISE(auto dest_location, AzureLocation::FromString(dest));
+  return impl_->CopyFile(src_location, dest_location);
 }
 
 Result<std::shared_ptr<io::InputStream>> AzureFileSystem::OpenInputStream(
diff --git a/cpp/src/arrow/filesystem/azurefs_test.cc b/cpp/src/arrow/filesystem/azurefs_test.cc
index b6f75987693c..71d85e10585b 100644
--- a/cpp/src/arrow/filesystem/azurefs_test.cc
+++ b/cpp/src/arrow/filesystem/azurefs_test.cc
@@ -771,6 +771,70 @@ TEST_F(AzureHierarchicalNamespaceFileSystemTest, DeleteDirContentsFailureNonexis
   ASSERT_RAISES(IOError, fs_->DeleteDirContents(directory_path, false));
 }
 
+TEST_F(AzuriteFileSystemTest, CopyFileSuccessDestinationNonexistent) {
+  const auto destination_path =
+      internal::ConcatAbstractPath(PreexistingContainerName(), "copy-destionation");
+  ASSERT_OK(fs_->CopyFile(PreexistingObjectPath(), destination_path));
+  ASSERT_OK_AND_ASSIGN(auto info, fs_->GetFileInfo(destination_path));
+  ASSERT_OK_AND_ASSIGN(auto stream, fs_->OpenInputStream(info));
+  ASSERT_OK_AND_ASSIGN(auto buffer, stream->Read(1024));
+  EXPECT_EQ(kLoremIpsum, buffer->ToString());
+}
+
+TEST_F(AzuriteFileSystemTest, CopyFileSuccessDestinationSame) {
+  ASSERT_OK(fs_->CopyFile(PreexistingObjectPath(), PreexistingObjectPath()));
+  ASSERT_OK_AND_ASSIGN(auto info, fs_->GetFileInfo(PreexistingObjectPath()));
+  ASSERT_OK_AND_ASSIGN(auto stream, fs_->OpenInputStream(info));
+  ASSERT_OK_AND_ASSIGN(auto buffer, stream->Read(1024));
+  EXPECT_EQ(kLoremIpsum, buffer->ToString());
+}
+
+TEST_F(AzuriteFileSystemTest, CopyFileSuccessDestinationContainer) {
+  const auto container_name = RandomContainerName();
+  ASSERT_OK(fs_->CreateDir(container_name));
+  ASSERT_OK(fs_->CopyFile(PreexistingObjectPath(), container_name));
+  const auto destination_path =
+      internal::ConcatAbstractPath(container_name, PreexistingObjectName());
+  ASSERT_OK_AND_ASSIGN(auto info, fs_->GetFileInfo(destination_path));
+  ASSERT_OK_AND_ASSIGN(auto stream, fs_->OpenInputStream(info));
+  ASSERT_OK_AND_ASSIGN(auto buffer, stream->Read(1024));
+  EXPECT_EQ(kLoremIpsum, buffer->ToString());
+}
+
+TEST_F(AzuriteFileSystemTest, CopyFileSuccessDestinationDirectory) {
+  const auto directory_path = internal::EnsureTrailingSlash(
+      internal::ConcatAbstractPath(RandomContainerName(), RandomDirectoryName()));
+  ASSERT_OK(fs_->CreateDir(directory_path, true));
+  ASSERT_OK(fs_->CopyFile(PreexistingObjectPath(), directory_path));
+  const auto destination_path =
+      internal::ConcatAbstractPath(directory_path, PreexistingObjectName());
+  ASSERT_OK_AND_ASSIGN(auto info, fs_->GetFileInfo(destination_path));
+  ASSERT_OK_AND_ASSIGN(auto stream, fs_->OpenInputStream(info));
+  ASSERT_OK_AND_ASSIGN(auto buffer, stream->Read(1024));
+  EXPECT_EQ(kLoremIpsum, buffer->ToString());
+}
+
+TEST_F(AzuriteFileSystemTest, CopyFileFailureSourceNonexistent) {
+  const auto destination_path =
+      internal::ConcatAbstractPath(PreexistingContainerName(), "copy-destionation");
+  ASSERT_RAISES(IOError, fs_->CopyFile(NotFoundObjectPath(), destination_path));
+}
+
+TEST_F(AzuriteFileSystemTest, CopyFileFailureDestinationParentNonexistent) {
+  const auto destination_path =
+      internal::ConcatAbstractPath(RandomContainerName(), "copy-destionation");
+  ASSERT_RAISES(IOError, fs_->CopyFile(PreexistingObjectPath(), destination_path));
+}
+
+TEST_F(AzuriteFileSystemTest, CopyFileUri) {
+  const auto destination_path =
+      internal::ConcatAbstractPath(PreexistingContainerName(), "copy-destionation");
+  ASSERT_RAISES(Invalid,
+                fs_->CopyFile("abfs://" + PreexistingObjectPath(), destination_path));
+  ASSERT_RAISES(Invalid,
+                fs_->CopyFile(PreexistingObjectPath(), "abfs://" + destination_path));
+}
+
 TEST_F(AzuriteFileSystemTest, OpenInputStreamString) {
   std::shared_ptr<io::InputStream> stream;
   ASSERT_OK_AND_ASSIGN(stream, fs_->OpenInputStream(PreexistingObjectPath()));

From f1fd5389e0829e9ad55474b4eddc1d530f4deaac Mon Sep 17 00:00:00 2001
From: Sutou Kouhei <kou@clear-code.com>
Date: Tue, 5 Dec 2023 11:29:21 +0900
Subject: [PATCH 2/3] Destination must be a blob path

---
 cpp/src/arrow/filesystem/azurefs.cc      | 19 +++---------------
 cpp/src/arrow/filesystem/azurefs_test.cc | 25 ------------------------
 2 files changed, 3 insertions(+), 41 deletions(-)

diff --git a/cpp/src/arrow/filesystem/azurefs.cc b/cpp/src/arrow/filesystem/azurefs.cc
index 6a544dc5ed71..b349fa7b57e7 100644
--- a/cpp/src/arrow/filesystem/azurefs.cc
+++ b/cpp/src/arrow/filesystem/azurefs.cc
@@ -1157,8 +1157,9 @@ class AzureFileSystem::Impl {
     }
   }
 
- private:
-  Status CopyFileInternal(const AzureLocation& src, const AzureLocation& dest) {
+  Status CopyFile(const AzureLocation& src, const AzureLocation& dest) {
+    RETURN_NOT_OK(ValidateFileLocation(src));
+    RETURN_NOT_OK(ValidateFileLocation(dest));
     if (src == dest) {
       return Status::OK();
     }
@@ -1176,20 +1177,6 @@ class AzureFileSystem::Impl {
     }
     return Status::OK();
   }
-
- public:
-  Status CopyFile(const AzureLocation& src, const AzureLocation& dest) {
-    RETURN_NOT_OK(ValidateFileLocation(src));
-    if (dest.container.empty()) {
-      return PathNotFound(dest);
-    }
-    if (dest.path.empty() || internal::HasTrailingSlash(dest.path)) {
-      ARROW_ASSIGN_OR_RAISE(auto real_dest, dest.join(src.path_parts.back()));
-      return CopyFileInternal(src, real_dest);
-    } else {
-      return CopyFileInternal(src, dest);
-    }
-  }
 };
 
 const AzureOptions& AzureFileSystem::options() const { return impl_->options(); }
diff --git a/cpp/src/arrow/filesystem/azurefs_test.cc b/cpp/src/arrow/filesystem/azurefs_test.cc
index 71d85e10585b..06f3076652eb 100644
--- a/cpp/src/arrow/filesystem/azurefs_test.cc
+++ b/cpp/src/arrow/filesystem/azurefs_test.cc
@@ -789,31 +789,6 @@ TEST_F(AzuriteFileSystemTest, CopyFileSuccessDestinationSame) {
   EXPECT_EQ(kLoremIpsum, buffer->ToString());
 }
 
-TEST_F(AzuriteFileSystemTest, CopyFileSuccessDestinationContainer) {
-  const auto container_name = RandomContainerName();
-  ASSERT_OK(fs_->CreateDir(container_name));
-  ASSERT_OK(fs_->CopyFile(PreexistingObjectPath(), container_name));
-  const auto destination_path =
-      internal::ConcatAbstractPath(container_name, PreexistingObjectName());
-  ASSERT_OK_AND_ASSIGN(auto info, fs_->GetFileInfo(destination_path));
-  ASSERT_OK_AND_ASSIGN(auto stream, fs_->OpenInputStream(info));
-  ASSERT_OK_AND_ASSIGN(auto buffer, stream->Read(1024));
-  EXPECT_EQ(kLoremIpsum, buffer->ToString());
-}
-
-TEST_F(AzuriteFileSystemTest, CopyFileSuccessDestinationDirectory) {
-  const auto directory_path = internal::EnsureTrailingSlash(
-      internal::ConcatAbstractPath(RandomContainerName(), RandomDirectoryName()));
-  ASSERT_OK(fs_->CreateDir(directory_path, true));
-  ASSERT_OK(fs_->CopyFile(PreexistingObjectPath(), directory_path));
-  const auto destination_path =
-      internal::ConcatAbstractPath(directory_path, PreexistingObjectName());
-  ASSERT_OK_AND_ASSIGN(auto info, fs_->GetFileInfo(destination_path));
-  ASSERT_OK_AND_ASSIGN(auto stream, fs_->OpenInputStream(info));
-  ASSERT_OK_AND_ASSIGN(auto buffer, stream->Read(1024));
-  EXPECT_EQ(kLoremIpsum, buffer->ToString());
-}
-
 TEST_F(AzuriteFileSystemTest, CopyFileFailureSourceNonexistent) {
   const auto destination_path =
       internal::ConcatAbstractPath(PreexistingContainerName(), "copy-destionation");

From 81af80e926570393cb6e1723416d095c2edc8366 Mon Sep 17 00:00:00 2001
From: Sutou Kouhei <kou@clear-code.com>
Date: Thu, 7 Dec 2023 09:02:11 +0900
Subject: [PATCH 3/3] Reject directory blob

---
 cpp/src/arrow/filesystem/azurefs.cc      | 4 +---
 cpp/src/arrow/filesystem/azurefs_test.cc | 6 ++++++
 2 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/cpp/src/arrow/filesystem/azurefs.cc b/cpp/src/arrow/filesystem/azurefs.cc
index b349fa7b57e7..301e2003d1de 100644
--- a/cpp/src/arrow/filesystem/azurefs.cc
+++ b/cpp/src/arrow/filesystem/azurefs.cc
@@ -153,6 +153,7 @@ Status ValidateFileLocation(const AzureLocation& location) {
   if (location.path.empty()) {
     return NotAFile(location);
   }
+  ARROW_RETURN_NOT_OK(internal::AssertNoTrailingSlash(location.path));
   return Status::OK();
 }
 
@@ -822,7 +823,6 @@ class AzureFileSystem::Impl {
   Result<std::shared_ptr<ObjectInputFile>> OpenInputFile(const AzureLocation& location,
                                                          AzureFileSystem* fs) {
     RETURN_NOT_OK(ValidateFileLocation(location));
-    ARROW_RETURN_NOT_OK(internal::AssertNoTrailingSlash(location.path));
     auto blob_client = std::make_shared<Azure::Storage::Blobs::BlobClient>(
         blob_service_client_->GetBlobContainerClient(location.container)
             .GetBlobClient(location.path));
@@ -835,7 +835,6 @@ class AzureFileSystem::Impl {
 
   Result<std::shared_ptr<ObjectInputFile>> OpenInputFile(const FileInfo& info,
                                                          AzureFileSystem* fs) {
-    ARROW_RETURN_NOT_OK(internal::AssertNoTrailingSlash(info.path()));
     if (info.type() == FileType::NotFound) {
       return ::arrow::fs::internal::PathNotFound(info.path());
     }
@@ -955,7 +954,6 @@ class AzureFileSystem::Impl {
       const std::shared_ptr<const KeyValueMetadata>& metadata, const bool truncate,
       AzureFileSystem* fs) {
     RETURN_NOT_OK(ValidateFileLocation(location));
-    ARROW_RETURN_NOT_OK(internal::AssertNoTrailingSlash(location.path));
 
     auto block_blob_client = std::make_shared<Azure::Storage::Blobs::BlockBlobClient>(
         blob_service_client_->GetBlobContainerClient(location.container)
diff --git a/cpp/src/arrow/filesystem/azurefs_test.cc b/cpp/src/arrow/filesystem/azurefs_test.cc
index 06f3076652eb..b0c96d41cfda 100644
--- a/cpp/src/arrow/filesystem/azurefs_test.cc
+++ b/cpp/src/arrow/filesystem/azurefs_test.cc
@@ -789,6 +789,12 @@ TEST_F(AzuriteFileSystemTest, CopyFileSuccessDestinationSame) {
   EXPECT_EQ(kLoremIpsum, buffer->ToString());
 }
 
+TEST_F(AzuriteFileSystemTest, CopyFileFailureDestinationTrailingSlash) {
+  ASSERT_RAISES(IOError,
+                fs_->CopyFile(PreexistingObjectPath(),
+                              internal::EnsureTrailingSlash(PreexistingObjectPath())));
+}
+
 TEST_F(AzuriteFileSystemTest, CopyFileFailureSourceNonexistent) {
   const auto destination_path =
       internal::ConcatAbstractPath(PreexistingContainerName(), "copy-destionation");
