feat: implement ComposeMany (#3239)

This fixes #3016.

Author: dopiera

google/cloud/storage/client.cc

@@ -336,6 +336,36 @@ StatusOr<PolicyDocumentResult> Client::SignPolicyDocument(
       internal::Base64Encode(signed_blob->signed_blob)};
 }
 
+namespace internal {
+
+ScopedDeleter::ScopedDeleter(std::function<Status(ObjectMetadata)> delete_fun)
+    : delete_fun_(std::move(delete_fun)) {}
+
+ScopedDeleter::~ScopedDeleter() { ExecuteDelete(); }
+
+void ScopedDeleter::Add(ObjectMetadata object) {
+  object_list_.emplace_back(std::move(object));
+}
+
+Status ScopedDeleter::ExecuteDelete() {
+  std::vector<ObjectMetadata> object_list;
+  // make sure the dtor will not do this again
+  object_list.swap(object_list_);
+
+  for (auto& object : object_list) {
+    Status status = delete_fun_(std::move(object));
+    // Fail on first error. If the service is unavailable, every deletion
+    // would potentially keep retrying until the timeout passes - this would
+    // take way too much time and would be pointless.
+    if (!status.ok()) {
+      return status;
+    }
+  }
+  return Status();
+}
+
+}  // namespace internal
+
 }  // namespace STORAGE_CLIENT_NS
 }  // namespace storage
 }  // namespace cloud

google/cloud/storage/client.h

@@ -3171,6 +3171,213 @@ Status DeleteByPrefix(Client& client, std::string const& bucket_name,
   return Status();
 }
 
+namespace internal {
+
+// Just a wrapper to allow for use in `google::cloud::internal::apply`.
+struct ComposeApplyHelper {
+  template <typename... Options>
+  StatusOr<ObjectMetadata> operator()(Options... options) const {
+    return client.ComposeObject(
+        std::move(bucket_name), std::move(source_objects),
+        std::move(destination_object_name), std::move(options)...);
+  }
+
+  Client& client;
+  std::string bucket_name;
+  std::vector<ComposeSourceObject> source_objects;
+  std::string destination_object_name;
+};
+
+// A helper to defer deletion of temporary GCS objects.
+class ScopedDeleter {
+ public:
+  // The actual deletion depends on local's types in a very non-trivial way,
+  // so we abstract this away by providing the function to delete one object.
+  ScopedDeleter(std::function<Status(ObjectMetadata)> delete_fun);
+  ~ScopedDeleter();
+
+  /// Defer object's deletion to this objects destruction (or ExecuteDelete())
+  void Add(ObjectMetadata object);
+
+  /// Execute all the deferred deletions now.
+  Status ExecuteDelete();
+
+ private:
+  std::function<Status(ObjectMetadata)> delete_fun_;
+  std::vector<ObjectMetadata> object_list_;
+};
+
+}  // namespace internal
+
+/**
+ * Compose existing objects into a new object in the same bucket.
+ *
+ * Contrary to `Client::ComposeObject`, this function doesn't have a limit on
+ * the number of source objects.
+ *
+ * The implementation may need to perform multiple Client::ComposeObject calls
+ * to create intermediate, temporary objects which are then further composed.
+ * Due to the lack of atomicity of this series of operations, stray temporary
+ * objects might be left over if there are transient failuers. In order to allow
+ * the user to easily control for such situations, the user is expected to
+ * provide a unique @p prefix parameter, which will become the prefix of all the
+ * temporary objects created by this function. Once this function finishes, the
+ * user may safely remove all objects with the provided prefix (e.g. via
+ * DeleteByPrefix()). We recommend using CreateRandomPrefix() for selecting a
+ * random prefix within a bucket.
+ *
+ * @param client the client on which to perform the operations needed by this
+ *     function
+ * @param bucket_name the name of the bucket used for source object and
+ *     destination object.
+ * @param source_objects objects used to compose `destination_object_name`.
+ * @param destination_object_name the composed object name.
+ * @param prefix prefix for temporary objects created by this function; the user
+ *     should guarantee that there are no objects with this prefix (except for a
+ *     marker, as created by `CreateRandomPrefix`)
+ * @param ignore_cleanup_failures if the composition succeeds but cleanup of
+ *     temporary objects fails, depending on this parameter either a success
+ *     will be returned (`true`) or the relevant cleanup error (`false`)
+ * @param options a list of optional query parameters and/or request headers.
+ *     Valid types for this operation include `DestinationPredefinedAcl`,
+ *     `EncryptionKey`, `IfGenerationMatch`, `IfMetagenerationMatch`
+ *     `KmsKeyName`, `QuotaUser`, `UserIp`, `UserProject` and
+ *     `WithObjectMetadata`.
+ *
+ * @par Idempotency
+ * This operation is not idempotent. While each request performed by this
+ * function is retried based on the client policies, the operation itself stops
+ * on the first request that fails.
+ *
+ * @par Example
+ * @snippet storage_object_samples.cc compose object from many
+ */
+template <typename... Options>
+StatusOr<ObjectMetadata> ComposeMany(
+    Client& client, std::string const& bucket_name,
+    std::vector<ComposeSourceObject> source_objects, std::string const& prefix,
+    std::string destination_object_name, bool ignore_cleanup_failures,
+    Options&&... options) {
+  using internal::Among;
+  using internal::NotAmong;
+  using internal::StaticTupleFilter;
+  std::size_t const max_num_objects = 32;
+
+  if (prefix.empty()) {
+    // In theory, nothing prevents us from using an empty prefix, but this is
+    // most likely a misuse. We recommend running `DeleteByPrefix` on this
+    // prefix, so this might not be the best idea.
+    return Status(StatusCode::kInvalidArgument,
+                  "ComposeMany requires a non-empty prefix to create temporary "
+                  "objects. Please check the documentation.");
+  }
+
+  if (source_objects.empty()) {
+    return Status(StatusCode::kInvalidArgument,
+                  "ComposeMany requires at least one source object.");
+  }
+
+  auto all_options = std::tie(options...);
+
+  // TODO(#3247): this list of type should somehow be generated
+  static_assert(
+      std::tuple_size<decltype(
+              StaticTupleFilter<NotAmong<
+                  DestinationPredefinedAcl, EncryptionKey, IfGenerationMatch,
+                  IfMetagenerationMatch, KmsKeyName, QuotaUser, UserIp,
+                  UserProject, WithObjectMetadata>::TPred>(
+                  all_options))>::value == 0,
+      "This functions accepts only options of type DestinationPredefinedAcl, "
+      "EncryptionKey, IfGenerationMatch, IfMetagenerationMatch, KmsKeyName, "
+      "QuotaUser, UserIp, UserProject or WithObjectMetadata.");
+
+  internal::ScopedDeleter deleter([&](ObjectMetadata const& object) {
+    return google::cloud::internal::apply(
+        internal::DeleteApplyHelper{client, bucket_name, object.name()},
+        std::tuple_cat(
+            std::make_tuple(IfGenerationMatch(object.generation())),
+            StaticTupleFilter<NotAmong<Versions>::TPred>(all_options)));
+  });
+
+  std::size_t num_tmp_objects = 0;
+  auto tmpobject_name_gen = [&num_tmp_objects, &prefix] {
+    return prefix + ".compose-tmp-" + std::to_string(num_tmp_objects++);
+  };
+
+  auto to_source_objects = [](std::vector<ObjectMetadata> objects) {
+    std::vector<ComposeSourceObject> sources(objects.size());
+    std::transform(objects.begin(), objects.end(), sources.begin(),
+                   [](ObjectMetadata m) {
+                     return ComposeSourceObject{m.name(), m.generation(), {}};
+                   });
+    return sources;
+  };
+
+  auto composer = [&](std::vector<ComposeSourceObject> compose_range,
+                      bool is_final) -> StatusOr<ObjectMetadata> {
+    if (is_final) {
+      return google::cloud::internal::apply(
+          internal::ComposeApplyHelper{client, bucket_name,
+                                       std::move(compose_range),
+                                       std::move(destination_object_name)},
+          std::tuple_cat(std::make_tuple(IfGenerationMatch(0)), all_options));
+    }
+    return google::cloud::internal::apply(
+        internal::ComposeApplyHelper{client, bucket_name,
+                                     std::move(compose_range),
+                                     tmpobject_name_gen()},
+        StaticTupleFilter<
+            NotAmong<IfGenerationMatch, IfMetagenerationMatch>::TPred>(
+            all_options));
+  };
+
+  auto reduce = [&](std::vector<ComposeSourceObject> source_objects)
+      -> StatusOr<std::vector<ObjectMetadata>> {
+    std::vector<ObjectMetadata> objects;
+    for (auto range_begin = source_objects.begin();
+         range_begin != source_objects.end();) {
+      std::size_t range_size = std::min<std::size_t>(
+          std::distance(range_begin, source_objects.end()), max_num_objects);
+      auto range_end = std::next(range_begin, range_size);
+      std::vector<ComposeSourceObject> compose_range(range_size);
+      std::move(range_begin, range_end, compose_range.begin());
+
+      bool const is_final_composition =
+          source_objects.size() <= max_num_objects;
+      auto object = composer(std::move(compose_range), is_final_composition);
+      if (!object) {
+        return std::move(object).status();
+      }
+      objects.push_back(*std::move(object));
+      if (!is_final_composition) {
+        deleter.Add(objects.back());
+      }
+      range_begin = range_end;
+    }
+    return objects;
+  };
+
+  StatusOr<ObjectMetadata> result;
+  do {
+    StatusOr<std::vector<ObjectMetadata>> objects = reduce(source_objects);
+    if (!objects) {
+      return objects.status();
+    }
+    if (objects->size() == 1) {
+      if (!ignore_cleanup_failures) {
+        auto delete_status = deleter.ExecuteDelete();
+        if (!delete_status.ok()) {
+          return delete_status;
+        }
+      }
+      result = std::move((*objects)[0]);
+      break;
+    }
+    source_objects = to_source_objects(*std::move(objects));
+  } while (source_objects.size() > 1);
+  return result;
+}
+
 }  // namespace STORAGE_CLIENT_NS
 }  // namespace storage
 }  // namespace cloud

google/cloud/storage/examples/run_examples_utils.sh

@@ -398,6 +398,11 @@ run_all_object_examples() {
       "${object_name}"
   run_example ./storage_object_samples delete-object \
       "${bucket_name}" "${composed_object_name}"
+  run_example ./storage_object_samples compose-object-from-many \
+      "${bucket_name}" "${composed_object_name}" "${object_name}" \
+      "${object_name}"
+  run_example ./storage_object_samples delete-object \
+      "${bucket_name}" "${composed_object_name}"
   run_example ./storage_object_samples copy-object \
       "${bucket_name}" "${object_name}" \
       "${bucket_name}" "${copied_object_name}"

google/cloud/storage/examples/storage_object_samples.cc

@@ -1097,6 +1097,48 @@ void ComposeObjectFromEncryptedObjects(google::cloud::storage::Client client,
    std::move(compose_objects));
 }
 
+void ComposeObjectFromMany(google::cloud::storage::Client client, int& argc,
+                           char* argv[]) {
+  if (argc < 4) {
+    throw Usage{
+        "compose-object-from-many <bucket-name> <destination-object-name>"
+        " <object_1> ..."};
+  }
+  auto bucket_name = ConsumeArg(argc, argv);
+  auto destination_object_name = ConsumeArg(argc, argv);
+  std::vector<google::cloud::storage::ComposeSourceObject> compose_objects;
+  while (argc > 1) {
+    compose_objects.push_back({ConsumeArg(argc, argv), {}, {}});
+  }
+  //! [compose object from many] [START storage_compose_file_from_many]
+  namespace gcs = google::cloud::storage;
+  using ::google::cloud::StatusOr;
+  [](gcs::Client client, std::string bucket_name,
+     std::string destination_object_name,
+     std::vector<gcs::ComposeSourceObject> compose_objects) {
+    StatusOr<gcs::ObjectMetadata> prefix_md =
+        gcs::CreateRandomPrefix(client, bucket_name, ".tmpfiles");
+    if (!prefix_md) {
+      throw std::runtime_error(prefix_md.status().message());
+    }
+    std::string const prefix = prefix_md->name();
+    StatusOr<gcs::ObjectMetadata> composed_object =
+        ComposeMany(client, bucket_name, compose_objects, prefix,
+                    destination_object_name, false);
+
+    if (!composed_object) {
+      throw std::runtime_error(composed_object.status().message());
+    }
+
+    std::cout << "Composed new object " << composed_object->name()
+              << " in bucket " << composed_object->bucket()
+              << "\nFull metadata: " << *composed_object << "\n";
+  }
+  //! [compose object from many] [END storage_compose_file_from_many]
+  (std::move(client), bucket_name, destination_object_name,
+   std::move(compose_objects));
+}
+
 void WriteObjectWithKmsKey(google::cloud::storage::Client client, int& argc,
                            char* argv[]) {
   if (argc != 4) {
@@ -1777,6 +1819,7 @@ int main(int argc, char* argv[]) try {
       {"compose-object", &ComposeObject},
       {"compose-object-from-encrypted-objects",
        &ComposeObjectFromEncryptedObjects},
+      {"compose-object-from-many", &ComposeObjectFromMany},
       {"write-object-with-kms-key", &WriteObjectWithKmsKey},
       {"rewrite-object", &RewriteObject},
       {"rewrite-object-non-blocking", &RewriteObjectNonBlocking},

google/cloud/storage/internal/tuple_filter.h

@@ -142,6 +142,18 @@ StaticTupleFilter(Tuple&& t) {
       std::forward<Tuple>(t)));
 }
 
+/**
+ * A factory of template predicates checking for presence on a type list
+ *
+ * @tparam Types the list of types which for which the predicate returns true.
+ */
+template <typename... Types>
+struct Among {
+  template <typename T>
+  using TPred = google::cloud::internal::disjunction<
+      std::is_same<typename std::decay<T>::type, Types>...>;
+};
+
 /**
  * A factory of template predicates checking for lack of presence on a type list
  *