Re: [PR] GH-38333: [C++][FS][Azure] Implement file writes [arrow]

[via GitHub]

kou commented on code in PR #38780:
URL: https://github.com/apache/arrow/pull/38780#discussion_r1398790905


##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -232,13 +232,11 @@ class AzureFileSystemTest : public ::testing::Test {
 
   void UploadLines(const std::vector<std::string>& lines, const char* 
path_to_file,
                    int total_size) {
-    // TODO(GH-38333): Switch to using Azure filesystem to write once its 
implemented.
-    auto blob_client =
-        
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
-            .GetBlockBlobClient(path_to_file);
+    const auto path = PreexistingContainerPath() + path_to_file;
+    ASSERT_OK_AND_ASSIGN(auto output, fs_->OpenOutputStream(path, {}));
     std::string all_lines = std::accumulate(lines.begin(), lines.end(), 
std::string(""));
-    blob_client.UploadFrom(reinterpret_cast<const uint8_t*>(all_lines.data()),
-                           total_size);
+    ASSERT_OK(output->Write(all_lines.data(), all_lines.size()));

Review Comment:
   ```suggestion
       ASSERT_OK(output->Write(all_lines));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -347,21 +345,26 @@ void 
AzureFileSystemTest::RunGetFileInfoObjectWithNestedStructureTest() {
   // Adds detailed tests to handle cases of different edge cases
   // with directory naming conventions (e.g. with and without slashes).
   constexpr auto kObjectName = 
"test-object-dir/some_other_dir/another_dir/foo";
-  // TODO(GH-38333): Switch to using Azure filesystem to write once its 
implemented.
-  blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
-      .GetBlockBlobClient(kObjectName)
-      .UploadFrom(reinterpret_cast<const uint8_t*>(kLoremIpsum), 
strlen(kLoremIpsum));
+  ASSERT_OK_AND_ASSIGN(
+      auto output,
+      fs_->OpenOutputStream(PreexistingContainerPath() + kObjectName, 
/*metadata=*/{}));
+  const auto data = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(data.data(), data.size()));
+  ASSERT_OK(output->Close());
 
   // 0 is immediately after "/" lexicographically, ensure that this doesn't
   // cause unexpected issues.
-  // TODO(GH-38333): Switch to using Azure filesystem to write once its 
implemented.
-  blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
-      .GetBlockBlobClient("test-object-dir/some_other_dir0")
-      .UploadFrom(reinterpret_cast<const uint8_t*>(kLoremIpsum), 
strlen(kLoremIpsum));
-
-  blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
-      .GetBlockBlobClient(std::string(kObjectName) + "0")
-      .UploadFrom(reinterpret_cast<const uint8_t*>(kLoremIpsum), 
strlen(kLoremIpsum));
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_->OpenOutputStream(
+                           PreexistingContainerPath() + 
"test-object-dir/some_other_dir0",
+                           /*metadata=*/{}));
+  ASSERT_OK(output->Write(data.data(), data.size()));

Review Comment:
   ```suggestion
     ASSERT_OK(output->Write(data));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto fs_with_defaults, 
AzureFileSystem::Make(options_));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -347,21 +345,26 @@ void 
AzureFileSystemTest::RunGetFileInfoObjectWithNestedStructureTest() {
   // Adds detailed tests to handle cases of different edge cases
   // with directory naming conventions (e.g. with and without slashes).
   constexpr auto kObjectName = 
"test-object-dir/some_other_dir/another_dir/foo";
-  // TODO(GH-38333): Switch to using Azure filesystem to write once its 
implemented.
-  blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
-      .GetBlockBlobClient(kObjectName)
-      .UploadFrom(reinterpret_cast<const uint8_t*>(kLoremIpsum), 
strlen(kLoremIpsum));
+  ASSERT_OK_AND_ASSIGN(
+      auto output,
+      fs_->OpenOutputStream(PreexistingContainerPath() + kObjectName, 
/*metadata=*/{}));
+  const auto data = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(data.data(), data.size()));

Review Comment:
   ```suggestion
     const std::string_view data(kLoremIpsum);
     ASSERT_OK(output->Write(data));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));

Review Comment:
   ```suggestion
     const std::string_view expected(kLoremIpsum);
     ASSERT_OK(output->Write(expected));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto input, fs_->OpenInputStream(path));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto output, fs_->OpenOutputStream(path, {}));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));

Review Comment:
   ```suggestion
       ASSERT_OK(output->Write(buffers[i]));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);

Review Comment:
   ```suggestion
     const std::string_view expected1(kLoremIpsum);
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto size, input->Read(inbuf.size(), inbuf.data()));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));

Review Comment:
   ```suggestion
     ASSERT_OK(output->Write(expected1));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto size, input->Read(inbuf.size(), inbuf.data()));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify that the initial content has been overwritten.
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected1);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenAppendStreamDoesNotTruncateExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));

Review Comment:
   ```suggestion
     const std::string_view expected0("Existing blob content");
     ASSERT_OK(output->Write(expected0));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify that the initial content has been overwritten.
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected1);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenAppendStreamDoesNotTruncateExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenAppendStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));

Review Comment:
   ```suggestion
     ASSERT_OK(output->Write(expected1));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -232,13 +232,11 @@ class AzureFileSystemTest : public ::testing::Test {
 
   void UploadLines(const std::vector<std::string>& lines, const char* 
path_to_file,
                    int total_size) {
-    // TODO(GH-38333): Switch to using Azure filesystem to write once its 
implemented.
-    auto blob_client =
-        
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
-            .GetBlockBlobClient(path_to_file);
+    const auto path = PreexistingContainerPath() + path_to_file;
+    ASSERT_OK_AND_ASSIGN(auto output, fs_->OpenOutputStream(path, {}));
     std::string all_lines = std::accumulate(lines.begin(), lines.end(), 
std::string(""));

Review Comment:
   ```suggestion
       const auto all_lines = std::accumulate(lines.begin(), lines.end(), 
std::string(""));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto output,
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");

Review Comment:
   How about checking all values at once? It's easier to debug on error.
   
   ```suggestion
     EXPECT_EQ(Azure::Core::CaseInsensitiveMap{std::make_pair("foo", "bar")},
               blob_metadata);
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));

Review Comment:
   ```suggestion
     const std::string_view expected(kLoremIpsum);
     ASSERT_OK(output->Write(expected));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);

Review Comment:
   ```suggestion
     EXPECT_EQ(expected, std::string_view(inbuf.data(), size));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -347,21 +345,26 @@ void 
AzureFileSystemTest::RunGetFileInfoObjectWithNestedStructureTest() {
   // Adds detailed tests to handle cases of different edge cases
   // with directory naming conventions (e.g. with and without slashes).
   constexpr auto kObjectName = 
"test-object-dir/some_other_dir/another_dir/foo";
-  // TODO(GH-38333): Switch to using Azure filesystem to write once its 
implemented.
-  blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
-      .GetBlockBlobClient(kObjectName)
-      .UploadFrom(reinterpret_cast<const uint8_t*>(kLoremIpsum), 
strlen(kLoremIpsum));
+  ASSERT_OK_AND_ASSIGN(
+      auto output,
+      fs_->OpenOutputStream(PreexistingContainerPath() + kObjectName, 
/*metadata=*/{}));
+  const auto data = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(data.data(), data.size()));
+  ASSERT_OK(output->Close());
 
   // 0 is immediately after "/" lexicographically, ensure that this doesn't
   // cause unexpected issues.
-  // TODO(GH-38333): Switch to using Azure filesystem to write once its 
implemented.
-  blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
-      .GetBlockBlobClient("test-object-dir/some_other_dir0")
-      .UploadFrom(reinterpret_cast<const uint8_t*>(kLoremIpsum), 
strlen(kLoremIpsum));
-
-  blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
-      .GetBlockBlobClient(std::string(kObjectName) + "0")
-      .UploadFrom(reinterpret_cast<const uint8_t*>(kLoremIpsum), 
strlen(kLoremIpsum));
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_->OpenOutputStream(
+                           PreexistingContainerPath() + 
"test-object-dir/some_other_dir0",
+                           /*metadata=*/{}));
+  ASSERT_OK(output->Write(data.data(), data.size()));
+  ASSERT_OK(output->Close());
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_->OpenOutputStream(PreexistingContainerPath() + kObjectName + 
"0",
+                                    /*metadata=*/{}));
+  ASSERT_OK(output->Write(data.data(), data.size()));

Review Comment:
   ```suggestion
     ASSERT_OK(output->Write(data));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));

Review Comment:
   ```suggestion
     ASSERT_OK(output->Write(expected));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());

Review Comment:
   ```suggestion
     // Defaults are overwritten and not merged.
     EXPECT_EQ(Azure::Core::CaseInsensitiveMap{std::make_pair("bar", "foo")},
               blob_metadata);
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify that the initial content has been overwritten.
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected1);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenAppendStreamDoesNotTruncateExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto size, input->Read(inbuf.size(), inbuf.data()));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));

Review Comment:
   ```suggestion
     const std::string_view expected0("Existing blob content");
     ASSERT_OK(output->Write(expected0));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);

Review Comment:
   ```suggestion
       ASSERT_EQ(expected, output->Tell());
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify that the initial content has been overwritten.
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected1);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenAppendStreamDoesNotTruncateExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto output, fs_->OpenOutputStream(path, {}));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto output, fs_->OpenOutputStream(path, {}));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);

Review Comment:
   ```suggestion
     EXPECT_EQ(expected0, std::string_view(inbuf.data(), size));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto input, fs_->OpenInputStream(path));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify that the initial content has been overwritten.
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected1);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenAppendStreamDoesNotTruncateExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);

Review Comment:
   ```suggestion
     EXPECT_EQ(expected0, std::string_view(inbuf.data()));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto input, fs_->OpenInputStream(path));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify that the initial content has been overwritten.
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected1);

Review Comment:
   ```suggestion
     EXPECT_EQ(expected1, std::string_view(inbuf.data(), size));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify that the initial content has been overwritten.
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected1);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenAppendStreamDoesNotTruncateExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto input, fs_->OpenInputStream(path));
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify that the initial content has been overwritten.
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected1);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenAppendStreamDoesNotTruncateExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenAppendStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);

Review Comment:
   ```suggestion
     const std::string_view expected1(kLoremIpsum);
   ```



##########
cpp/src/arrow/filesystem/azurefs_test.cc:
##########
@@ -647,6 +650,171 @@ TEST_F(AzuriteFileSystemTest, OpenInputStreamClosed) {
   ASSERT_RAISES(Invalid, stream->Tell());
 }
 
+TEST_F(AzuriteFileSystemTest, TestWriteMetadata) {
+  options_.default_metadata = arrow::key_value_metadata({{"foo", "bar"}});
+
+  std::shared_ptr<AzureFileSystem> fs_with_defaults;
+  ASSERT_OK_AND_ASSIGN(fs_with_defaults, AzureFileSystem::Make(options_));
+  std::string path = "object_with_defaults";
+  auto location = PreexistingContainerPath() + path;
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output,
+                       fs_with_defaults->OpenOutputStream(location, 
/*metadata=*/{}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify the metadata has been set.
+  auto blob_metadata =
+      blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+          .GetBlockBlobClient(path)
+          .GetProperties()
+          .Value.Metadata;
+  EXPECT_EQ(blob_metadata["foo"], "bar");
+
+  // Check that explicit metadata overrides the defaults.
+  ASSERT_OK_AND_ASSIGN(
+      output, fs_with_defaults->OpenOutputStream(
+                  location, /*metadata=*/arrow::key_value_metadata({{"bar", 
"foo"}})));
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+  blob_metadata = 
blob_service_client_->GetBlobContainerClient(PreexistingContainerName())
+                      .GetBlockBlobClient(path)
+                      .GetProperties()
+                      .Value.Metadata;
+  EXPECT_EQ(blob_metadata["bar"], "foo");
+  // Defaults are overwritten and not merged.
+  EXPECT_EQ(blob_metadata.find("foo"), blob_metadata.end());
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamSmall) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected.data(), expected.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+
+  EXPECT_EQ(std::string(inbuf.data(), size), expected);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamLarge) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  std::array<std::int64_t, 3> sizes{257 * 1024, 258 * 1024, 259 * 1024};
+  std::array<std::string, 3> buffers{
+      std::string(sizes[0], 'A'),
+      std::string(sizes[1], 'B'),
+      std::string(sizes[2], 'C'),
+  };
+  auto expected = std::int64_t{0};
+  for (auto i = 0; i != 3; ++i) {
+    ASSERT_OK(output->Write(buffers[i].data(), buffers[i].size()));
+    expected += sizes[i];
+    ASSERT_EQ(output->Tell(), expected);
+  }
+  ASSERT_OK(output->Close());
+
+  // Verify we can read the object back.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::string contents;
+  std::shared_ptr<Buffer> buffer;
+  do {
+    ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
+    ASSERT_TRUE(buffer);
+    contents.append(buffer->ToString());
+  } while (buffer->size() != 0);
+
+  EXPECT_EQ(contents, buffers[0] + buffers[1] + buffers[2]);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamTruncatesExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify that the initial content has been overwritten.
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected1);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenAppendStreamDoesNotTruncateExistingFile) {
+  const auto path = PreexistingContainerPath() + "test-write-object";
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));
+  const std::string expected0 = "Existing blob content";
+  ASSERT_OK(output->Write(expected0.data(), expected0.size()));
+  ASSERT_OK(output->Close());
+
+  // Check that the initial content has been written - if not this test is not 
achieving
+  // what it's meant to.
+  std::shared_ptr<io::InputStream> input;
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+
+  std::array<char, 1024> inbuf{};
+  std::int64_t size;
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0);
+
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenAppendStream(path, {}));
+  const auto expected1 = std::string(kLoremIpsum);
+  ASSERT_OK(output->Write(expected1.data(), expected1.size()));
+  ASSERT_OK(output->Close());
+
+  // Verify that the initial content has not been overwritten and that the 
block from
+  // the other client was not committed.
+  ASSERT_OK_AND_ASSIGN(input, fs_->OpenInputStream(path));
+  ASSERT_OK_AND_ASSIGN(size, input->Read(inbuf.size(), inbuf.data()));
+  EXPECT_EQ(std::string(inbuf.data(), size), expected0 + expected1);
+}
+
+TEST_F(AzuriteFileSystemTest, OpenOutputStreamClosed) {
+  const auto path = internal::ConcatAbstractPath(PreexistingContainerName(),
+                                                 
"open-output-stream-closed.txt");
+  std::shared_ptr<io::OutputStream> output;
+  ASSERT_OK_AND_ASSIGN(output, fs_->OpenOutputStream(path, {}));

Review Comment:
   ```suggestion
     ASSERT_OK_AND_ASSIGN(auto output, fs_->OpenOutputStream(path, {}));
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscr...@arrow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org