Optimize VFS::touch and fix defects and race conditions. (#5285)

[SC-53188](https://app.shortcut.com/tiledb-inc/story/53188/azure-update-touch-to-use-conditional-requests)

[SC-53189](https://app.shortcut.com/tiledb-inc/story/53189/gcs-update-touch-to-use-conditional-requests)

[SC-54494](https://app.shortcut.com/tiledb-inc/story/54494/win-touch-is-prone-to-race-conditions)

This PR updates the implementations of `VFS::touch` for Azure, GCS and
Windows to improve performance and fix race conditions:

* On Azure we remove checking if the blob exists, and instead use a
single [conditional
request](https://learn.microsoft.com/en-us/rest/api/storageservices/specifying-conditional-headers-for-blob-service-operations)
to upload an empty blob if it does not already exist. This reduces
round-trips and fixes race conditions if the blob was created between
the existence check and the upload.
* GCS did not even have an existence check, which means that the
behavior of `touch` was incorrect and always overwrote the file. This
was fixed by adding a [request
precondition](https://cloud.google.com/storage/docs/request-preconditions).
* On Windows the existence check was removed. We were already doing the
right thing and opened the file with `CREATE_NEW`, but did not mask
failures with `ERROR_FILE_EXISTS` status code. We fix this.

Additionally, the documentation of `tiledb_vfs_touch` was updated to
mention that the file is created only if it does not exist, matching
existing behavior in most VFS implmentations, and the expectations from
the `touch` command.

Testing coverage was added.

---
TYPE: BUG
DESC: Fixed `tiledb_vfs_touch` to not overwrite existing files on GCS.

---
TYPE: BUG
DESC: Fixed `tiledb_vfs_touch` to not overwrite existing files or fail
on Azure and Windows respectively, under race conditions.

---
TYPE: C_API
DESC: Update documentation of `tiledb_vfs_touch` to specify that
existing files are not overwritten, matching the current behavior.

Author: teo-tsirpanis

test/src/unit-vfs.cc

@@ -619,6 +619,12 @@ TEST_CASE("VFS: test ls_with_sizes", "[vfs][ls-with-sizes]") {
   // Directories don't get a size
   REQUIRE(children[1].file_size() == 0);
 
+  // Touch does not overwrite an existing file.
+  require_tiledb_ok(vfs_ls.touch(URI(subdir_file)));
+  uint64_t size;
+  require_tiledb_ok(vfs_ls.file_size(URI(subdir_file), &size));
+  REQUIRE(size == 6);
+
   // Clean up
   require_tiledb_ok(vfs_ls.remove_dir(URI(path)));
 }

tiledb/api/c_api/vfs/vfs_api_external.h

@@ -640,7 +640,9 @@ TILEDB_EXPORT capi_return_t tiledb_vfs_fh_is_closed(
     tiledb_ctx_t* ctx, tiledb_vfs_fh_t* fh, int32_t* is_closed) TILEDB_NOEXCEPT;
 
 /**
- * Touches a file, i.e., creates a new empty file.
+ * Touches a file, i.e., creates a new empty file if it does not already exist.
+ *
+ * The access timestamps of the file are not guaranteed to change.
  *
  * **Example:**
  *

tiledb/sm/cpp_api/vfs.h

@@ -519,7 +519,8 @@ class VFS {
         ctx.ptr().get(), vfs_.get(), old_uri.c_str(), new_uri.c_str()));
   }
 
-  /** Touches a file with the input URI, i.e., creates a new empty file. */
+  /** Touches a file with the input URI, i.e., creates a new empty file if it
+   * does not already exist. */
   void touch(const std::string& uri) const {
     auto& ctx = ctx_.get();
     ctx.handle_error(

tiledb/sm/filesystem/azure.cc

@@ -730,16 +730,22 @@ void Azure::touch(const URI& uri) const {
     throw AzureException(
         "Cannot create file; URI is a directory: " + uri.to_string());
   }
-  if (is_blob(uri)) {
-    return;
-  }
 
   auto [container_name, blob_path] = parse_azure_uri(uri);
   try {
+    ::Azure::Core::IO::MemoryBodyStream stream(nullptr, 0);
+    ::Azure::Storage::Blobs::UploadBlockBlobOptions options;
+    // Set condition that the object does not exist.
+    options.AccessConditions.IfNoneMatch = ::Azure::ETag::Any();
+
     c.GetBlobContainerClient(container_name)
         .GetBlockBlobClient(blob_path)
-        .UploadFrom(nullptr, 0);
+        .Upload(stream, options);
   } catch (const ::Azure::Storage::StorageException& e) {
+    if (e.StatusCode == ::Azure::Core::Http::HttpStatusCode::Conflict) {
+      // Blob already exists.
+      return;
+    }
     throw AzureException(
         "Touch blob failed on: " + uri.to_string() + "; " + e.Message);
   }

tiledb/sm/filesystem/gcs.cc

@@ -741,10 +741,20 @@ Status GCS::touch(const URI& uri) const {
   RETURN_NOT_OK(parse_gcs_uri(uri, &bucket_name, &object_path));
 
   google::cloud::StatusOr<google::cloud::storage::ObjectMetadata>
-      object_metadata = client_->InsertObject(bucket_name, object_path, "");
+      object_metadata = client_->InsertObject(
+          bucket_name,
+          object_path,
+          "",
+          // An if-generation-match value of zero makes the request succeed only
+          // if the object does not exist.
+          // https://cloud.google.com/storage/docs/request-preconditions#special-case
+          google::cloud::storage::IfGenerationMatch(0));
 
   if (!object_metadata.ok()) {
-    const google::cloud::Status status = object_metadata.status();
+    const google::cloud::Status& status = object_metadata.status();
+    if (status.code() == google::cloud::StatusCode::kFailedPrecondition) {
+      return Status::Ok();
+    }
 
     throw GCSException(
         "Touch object failed on: " + uri.to_string() + " (" + status.message() +

tiledb/sm/filesystem/win.cc

@@ -161,10 +161,6 @@ Status Win::create_dir(const std::string& path) const {
 }
 
 Status Win::touch(const std::string& filename) const {
-  if (is_file(filename)) {
-    return Status::Ok();
-  }
-
   HANDLE file_h = CreateFile(
       filename.c_str(),
       GENERIC_WRITE,
@@ -180,6 +176,10 @@ Status Win::touch(const std::string& filename) const {
   tiledb::common::ScopedExecutor onexit1(closefileonexit);
   if (file_h == INVALID_HANDLE_VALUE) {
     auto gle = GetLastError();
+    if (gle == ERROR_FILE_EXISTS) {
+      // Do not fail if the file already exists.
+      return Status::Ok();
+    }
     return LOG_STATUS(Status_IOError(
         std::string("Failed to create file '") + filename + " (" +
         get_last_error_msg(gle, "CreateFile") + ")'"));