wgtmac commented on code in PR #516: URL: https://github.com/apache/iceberg-cpp/pull/516#discussion_r2703123877
########## src/iceberg/update/append_files.h: ########## @@ -0,0 +1,70 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include <memory> + +#include "iceberg/iceberg_export.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { + +/// \brief API for appending new files in a table. +/// +/// This API accumulates file additions, produces a new Snapshot of the table, and commits +/// that snapshot as the current. +/// +/// When committing, these changes will be applied to the latest table snapshot. Commit +/// conflicts will be resolved by applying the changes to the new latest snapshot and +/// reattempting the commit. +class ICEBERG_EXPORT AppendFiles { Review Comment: IMHO, we don't need this class. The Java impl's AppendFiles extends SnapshotUpdate so that one has more methods to call. We can just return `FastAppend` from `NewFastAppend()` and then return `AppendFiles|MergeAppend` from `NewAppend()`. ########## src/iceberg/update/fast_append.h: ########## @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +/// \file iceberg/update/fast_append.h + +#include <functional> +#include <memory> +#include <string> +#include <unordered_map> +#include <unordered_set> +#include <vector> + +#include "iceberg/iceberg_export.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_list.h" +#include "iceberg/result.h" +#include "iceberg/snapshot.h" +#include "iceberg/type_fwd.h" +#include "iceberg/update/append_files.h" +#include "iceberg/update/snapshot_update.h" + +namespace iceberg { + +/// \brief Append implementation that adds new manifest files for writes. +/// +/// FastAppend is optimized for appending new data files to a table, it creates new +/// manifest files for the added data without compacting or rewriting existing manifests, +/// making it faster for write-heavy workloads. +class ICEBERG_EXPORT FastAppend : public SnapshotUpdate, public AppendFiles { + public: + /// \brief Create a new FastAppend instance. + /// + /// \param table_name The name of the table + /// \param transaction The transaction to use for this update + /// \return A Result containing the FastAppend instance or an error + static Result<std::unique_ptr<FastAppend>> Make( + std::string table_name, std::shared_ptr<Transaction> transaction); + + /// \brief Append a data file to this update. + /// + /// \param file The data file to append + /// \return Reference to this for method chaining + FastAppend& AppendFile(std::shared_ptr<DataFile> file) override; + + /// \brief Append a manifest file to this update. + /// + /// The manifest must only contain added files (no existing or deleted files). + /// If the manifest doesn't have a snapshot ID assigned and snapshot ID inheritance + /// is enabled, it will be used directly. Otherwise, it will be copied with the + /// new snapshot ID. + /// + /// \param manifest The manifest file to append + /// \return Reference to this for method chaining + FastAppend& AppendManifest(const ManifestFile& manifest) override; + + /// \brief Set the target branch for this update. + /// + /// \param branch The branch name + /// \return Reference to this for method chaining + FastAppend& ToBranch(const std::string& branch); + + /// \brief Set a summary property. + /// + /// \param property The property name + /// \param value The property value + /// \return Reference to this for method chaining + FastAppend& Set(const std::string& property, const std::string& value); + + Kind kind() const override { return Kind::kUpdateSnapshot; } + + std::string operation() override; + + Result<std::vector<ManifestFile>> Apply( + const TableMetadata& metadata_to_update, + const std::shared_ptr<Snapshot>& snapshot) override; + std::unordered_map<std::string, std::string> Summary() override; + void CleanUncommitted(const std::unordered_set<std::string>& committed) override; + bool CleanupAfterCommit() const override; + + private: + explicit FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction); + + /// \brief Get the partition spec by spec ID. + Result<std::shared_ptr<PartitionSpec>> Spec(int32_t spec_id); + + /// \brief Copy a manifest file with a new snapshot ID. + /// + /// \param manifest The manifest to copy + /// \return The copied manifest file + Result<ManifestFile> CopyManifest(const ManifestFile& manifest); + + /// \brief Write new manifests for the accumulated data files. + /// + /// \return A vector of manifest files, or an error + Result<std::vector<ManifestFile>> WriteNewManifests(); + + private: + struct DataFilePtrHash { Review Comment: Should we move DataFilePtrHash and DataFilePtrEqual to manifest_entry.h (where DataFile is defined) or content_file_util.h (if DataFileSet is needed as below comment)? ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; Review Comment: Does it have side effect on directly setting snapshot_id on `append_manifests_` and `rewritten_append_manifests_`? If not, we can save allocation of temp `ManifestFile`s and batch insert `append_manifests_` and `rewritten_append_manifests_` to `manifests`. ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( Review Comment: We need to validate table_name to be non-empty and transaction is not null. ########## src/iceberg/update/fast_append.h: ########## @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +/// \file iceberg/update/fast_append.h + +#include <functional> +#include <memory> +#include <string> +#include <unordered_map> +#include <unordered_set> +#include <vector> + +#include "iceberg/iceberg_export.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_list.h" +#include "iceberg/result.h" +#include "iceberg/snapshot.h" +#include "iceberg/type_fwd.h" +#include "iceberg/update/append_files.h" +#include "iceberg/update/snapshot_update.h" + +namespace iceberg { + +/// \brief Append implementation that adds new manifest files for writes. +/// +/// FastAppend is optimized for appending new data files to a table, it creates new +/// manifest files for the added data without compacting or rewriting existing manifests, +/// making it faster for write-heavy workloads. +class ICEBERG_EXPORT FastAppend : public SnapshotUpdate, public AppendFiles { + public: + /// \brief Create a new FastAppend instance. + /// + /// \param table_name The name of the table + /// \param transaction The transaction to use for this update + /// \return A Result containing the FastAppend instance or an error + static Result<std::unique_ptr<FastAppend>> Make( + std::string table_name, std::shared_ptr<Transaction> transaction); + + /// \brief Append a data file to this update. + /// + /// \param file The data file to append + /// \return Reference to this for method chaining + FastAppend& AppendFile(std::shared_ptr<DataFile> file) override; + + /// \brief Append a manifest file to this update. + /// + /// The manifest must only contain added files (no existing or deleted files). + /// If the manifest doesn't have a snapshot ID assigned and snapshot ID inheritance + /// is enabled, it will be used directly. Otherwise, it will be copied with the + /// new snapshot ID. + /// + /// \param manifest The manifest file to append + /// \return Reference to this for method chaining + FastAppend& AppendManifest(const ManifestFile& manifest) override; + + /// \brief Set the target branch for this update. + /// + /// \param branch The branch name + /// \return Reference to this for method chaining + FastAppend& ToBranch(const std::string& branch); Review Comment: Can we simplify reuse `SetTargetBranch` or rename it to `ToBranch` if this is a better name? ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), + snapshot_manifests_span.end()); + manifests.insert(manifests.end(), snapshot_manifests.begin(), + snapshot_manifests.end()); + } + + return manifests; +} + +std::unordered_map<std::string, std::string> FastAppend::Summary() { + summary_.SetPartitionSummaryLimit( + base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); + return summary_.Build(); +} + +void FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) { + // Clean up new manifests that were written but not committed + if (!new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + new_manifests_.clear(); + } + + // Clean up only rewritten_append_manifests as they are always owned by the table + // Don't clean up append_manifests as they are added to the manifest list and are + // not compacted + if (!rewritten_append_manifests_.empty()) { + for (const auto& manifest : rewritten_append_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + } +} + +bool FastAppend::CleanupAfterCommit() const { + // Cleanup after committing is disabled for FastAppend unless there are + // rewritten_append_manifests because: + // 1.) Appended manifests are never rewritten + // 2.) Manifests which are written out as part of appendFile are already cleaned + // up between commit attempts in writeNewManifests Review Comment: ```suggestion // up between commit attempts in writeNewManifests ``` ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, Review Comment: nit: we can move `kInvalidSequenceNumber` and other similar constants to `constants.h` ########## src/iceberg/update/fast_append.h: ########## @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +/// \file iceberg/update/fast_append.h + +#include <functional> +#include <memory> +#include <string> +#include <unordered_map> +#include <unordered_set> +#include <vector> + +#include "iceberg/iceberg_export.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_list.h" +#include "iceberg/result.h" +#include "iceberg/snapshot.h" +#include "iceberg/type_fwd.h" +#include "iceberg/update/append_files.h" +#include "iceberg/update/snapshot_update.h" + +namespace iceberg { + +/// \brief Append implementation that adds new manifest files for writes. +/// +/// FastAppend is optimized for appending new data files to a table, it creates new +/// manifest files for the added data without compacting or rewriting existing manifests, +/// making it faster for write-heavy workloads. +class ICEBERG_EXPORT FastAppend : public SnapshotUpdate, public AppendFiles { + public: + /// \brief Create a new FastAppend instance. + /// + /// \param table_name The name of the table + /// \param transaction The transaction to use for this update + /// \return A Result containing the FastAppend instance or an error + static Result<std::unique_ptr<FastAppend>> Make( + std::string table_name, std::shared_ptr<Transaction> transaction); + + /// \brief Append a data file to this update. + /// + /// \param file The data file to append + /// \return Reference to this for method chaining + FastAppend& AppendFile(std::shared_ptr<DataFile> file) override; + + /// \brief Append a manifest file to this update. + /// + /// The manifest must only contain added files (no existing or deleted files). + /// If the manifest doesn't have a snapshot ID assigned and snapshot ID inheritance + /// is enabled, it will be used directly. Otherwise, it will be copied with the + /// new snapshot ID. + /// + /// \param manifest The manifest file to append + /// \return Reference to this for method chaining + FastAppend& AppendManifest(const ManifestFile& manifest) override; + + /// \brief Set the target branch for this update. + /// + /// \param branch The branch name + /// \return Reference to this for method chaining + FastAppend& ToBranch(const std::string& branch); + + /// \brief Set a summary property. + /// + /// \param property The property name + /// \param value The property value + /// \return Reference to this for method chaining + FastAppend& Set(const std::string& property, const std::string& value); + + Kind kind() const override { return Kind::kUpdateSnapshot; } + + std::string operation() override; + + Result<std::vector<ManifestFile>> Apply( + const TableMetadata& metadata_to_update, + const std::shared_ptr<Snapshot>& snapshot) override; + std::unordered_map<std::string, std::string> Summary() override; + void CleanUncommitted(const std::unordered_set<std::string>& committed) override; + bool CleanupAfterCommit() const override; + + private: + explicit FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction); + + /// \brief Get the partition spec by spec ID. + Result<std::shared_ptr<PartitionSpec>> Spec(int32_t spec_id); + + /// \brief Copy a manifest file with a new snapshot ID. + /// + /// \param manifest The manifest to copy + /// \return The copied manifest file + Result<ManifestFile> CopyManifest(const ManifestFile& manifest); + + /// \brief Write new manifests for the accumulated data files. + /// + /// \return A vector of manifest files, or an error + Result<std::vector<ManifestFile>> WriteNewManifests(); + + private: + struct DataFilePtrHash { + size_t operator()(const std::shared_ptr<DataFile>& file) const { + if (!file) { + return 0; + } + return std::hash<std::string>{}(file->file_path); + } + }; + + struct DataFilePtrEqual { + bool operator()(const std::shared_ptr<DataFile>& left, + const std::shared_ptr<DataFile>& right) const { + if (left == right) { + return true; + } + if (!left || !right) { + return false; + } + return left->file_path == right->file_path; + } + }; + + std::string table_name_; + SnapshotSummaryBuilder summary_; + std::unordered_map<int32_t, std::unordered_set<std::shared_ptr<DataFile>, + DataFilePtrHash, DataFilePtrEqual>> + new_data_files_by_spec_; + std::vector<ManifestFile> append_manifests_; + std::vector<ManifestFile> rewritten_append_manifests_; + std::vector<ManifestFile> new_manifests_; + bool has_new_files_{false}; + int32_t copy_manifest_count_{0}; Review Comment: We shouldn't add this. ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), + snapshot_manifests_span.end()); + manifests.insert(manifests.end(), snapshot_manifests.begin(), + snapshot_manifests.end()); + } + + return manifests; +} + +std::unordered_map<std::string, std::string> FastAppend::Summary() { + summary_.SetPartitionSummaryLimit( + base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); + return summary_.Build(); +} + +void FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) { + // Clean up new manifests that were written but not committed + if (!new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + new_manifests_.clear(); + } + + // Clean up only rewritten_append_manifests as they are always owned by the table + // Don't clean up append_manifests as they are added to the manifest list and are + // not compacted + if (!rewritten_append_manifests_.empty()) { + for (const auto& manifest : rewritten_append_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + } +} + +bool FastAppend::CleanupAfterCommit() const { + // Cleanup after committing is disabled for FastAppend unless there are + // rewritten_append_manifests because: + // 1.) Appended manifests are never rewritten + // 2.) Manifests which are written out as part of appendFile are already cleaned + // up between commit attempts in writeNewManifests + return !rewritten_append_manifests_.empty(); +} + +Result<std::shared_ptr<PartitionSpec>> FastAppend::Spec(int32_t spec_id) { + return base().PartitionSpecById(spec_id); +} + +Result<ManifestFile> FastAppend::CopyManifest(const ManifestFile& manifest) { + const TableMetadata& current = base(); + ICEBERG_ASSIGN_OR_RAISE(auto schema, current.Schema()); + ICEBERG_ASSIGN_OR_RAISE(auto spec, + current.PartitionSpecById(manifest.partition_spec_id)); + + // Read the manifest entries + ICEBERG_ASSIGN_OR_RAISE( + auto reader, + ManifestReader::Make(manifest, transaction_->table()->io(), schema, spec)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries()); + + // Create a new manifest writer + // Generate a unique manifest path using the transaction's metadata location + std::string filename = std::format("copy-m{}.avro", copy_manifest_count_++); + std::string new_manifest_path = transaction_->MetadataFileLocation(filename); + int64_t snapshot_id = SnapshotId(); + ICEBERG_ASSIGN_OR_RAISE( + auto writer, ManifestWriter::MakeWriter( + current.format_version, snapshot_id, new_manifest_path, + transaction_->table()->io(), spec, schema, ManifestContent::kData, + /*first_row_id=*/current.next_row_id)); + + // Write all entries as added entries with the new snapshot ID + for (auto& entry : entries) { Review Comment: We need something like the Java `ManifestFiles.copyAppendManifest` function. Perhaps we need to add a manifest_util_internal.h for reusing these functions. ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), + snapshot_manifests_span.end()); + manifests.insert(manifests.end(), snapshot_manifests.begin(), + snapshot_manifests.end()); + } + + return manifests; +} + +std::unordered_map<std::string, std::string> FastAppend::Summary() { + summary_.SetPartitionSummaryLimit( + base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); + return summary_.Build(); +} + +void FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) { + // Clean up new manifests that were written but not committed + if (!new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + new_manifests_.clear(); + } + + // Clean up only rewritten_append_manifests as they are always owned by the table Review Comment: ```suggestion // Clean up only rewritten append manifests as they are always owned by the table ``` ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), + snapshot_manifests_span.end()); + manifests.insert(manifests.end(), snapshot_manifests.begin(), + snapshot_manifests.end()); + } + + return manifests; +} + +std::unordered_map<std::string, std::string> FastAppend::Summary() { + summary_.SetPartitionSummaryLimit( + base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); + return summary_.Build(); +} + +void FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) { + // Clean up new manifests that were written but not committed + if (!new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + new_manifests_.clear(); + } + + // Clean up only rewritten_append_manifests as they are always owned by the table + // Don't clean up append_manifests as they are added to the manifest list and are + // not compacted + if (!rewritten_append_manifests_.empty()) { + for (const auto& manifest : rewritten_append_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + } +} + +bool FastAppend::CleanupAfterCommit() const { + // Cleanup after committing is disabled for FastAppend unless there are + // rewritten_append_manifests because: + // 1.) Appended manifests are never rewritten + // 2.) Manifests which are written out as part of appendFile are already cleaned + // up between commit attempts in writeNewManifests + return !rewritten_append_manifests_.empty(); +} + +Result<std::shared_ptr<PartitionSpec>> FastAppend::Spec(int32_t spec_id) { + return base().PartitionSpecById(spec_id); +} + +Result<ManifestFile> FastAppend::CopyManifest(const ManifestFile& manifest) { + const TableMetadata& current = base(); + ICEBERG_ASSIGN_OR_RAISE(auto schema, current.Schema()); + ICEBERG_ASSIGN_OR_RAISE(auto spec, + current.PartitionSpecById(manifest.partition_spec_id)); + + // Read the manifest entries + ICEBERG_ASSIGN_OR_RAISE( + auto reader, + ManifestReader::Make(manifest, transaction_->table()->io(), schema, spec)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries()); + + // Create a new manifest writer + // Generate a unique manifest path using the transaction's metadata location + std::string filename = std::format("copy-m{}.avro", copy_manifest_count_++); + std::string new_manifest_path = transaction_->MetadataFileLocation(filename); + int64_t snapshot_id = SnapshotId(); + ICEBERG_ASSIGN_OR_RAISE( + auto writer, ManifestWriter::MakeWriter( + current.format_version, snapshot_id, new_manifest_path, + transaction_->table()->io(), spec, schema, ManifestContent::kData, + /*first_row_id=*/current.next_row_id)); + + // Write all entries as added entries with the new snapshot ID + for (auto& entry : entries) { + ICEBERG_PRECHECK(entry.status == ManifestStatus::kAdded, + "Manifest to copy must only contain added entries"); + entry.snapshot_id = snapshot_id; + ICEBERG_RETURN_UNEXPECTED(writer->WriteAddedEntry(entry)); + } + + ICEBERG_RETURN_UNEXPECTED(writer->Close()); + ICEBERG_ASSIGN_OR_RAISE(auto new_manifest, writer->ToManifestFile()); + + summary_.AddedManifest(new_manifest); + + return new_manifest; +} + +Result<std::vector<ManifestFile>> FastAppend::WriteNewManifests() { + // If there are new files and manifests were already written, clean them up + if (has_new_files_ && !new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + ICEBERG_RETURN_UNEXPECTED(DeleteFile(manifest.manifest_path)); + } + new_manifests_.clear(); + } + + // Write new manifests if there are new data files + if (new_manifests_.empty() && !new_data_files_by_spec_.empty()) { + for (const auto& [spec_id, data_files] : new_data_files_by_spec_) { + ICEBERG_ASSIGN_OR_RAISE(auto spec, Spec(spec_id)); + std::vector<std::shared_ptr<DataFile>> files; + files.reserve(data_files.size()); + std::ranges::copy(data_files, std::back_inserter(files)); + ICEBERG_ASSIGN_OR_RAISE(auto written_manifests, WriteDataManifests(files, spec)); Review Comment: We need to revisit the signature of `WriteDataManifests` to use iterator begin and end of DataFile. We can add a TODO comment to WriteDataManifests for now. ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), + snapshot_manifests_span.end()); + manifests.insert(manifests.end(), snapshot_manifests.begin(), + snapshot_manifests.end()); + } + + return manifests; +} + +std::unordered_map<std::string, std::string> FastAppend::Summary() { + summary_.SetPartitionSummaryLimit( + base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); + return summary_.Build(); +} + +void FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) { + // Clean up new manifests that were written but not committed + if (!new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + new_manifests_.clear(); + } + + // Clean up only rewritten_append_manifests as they are always owned by the table + // Don't clean up append_manifests as they are added to the manifest list and are + // not compacted + if (!rewritten_append_manifests_.empty()) { + for (const auto& manifest : rewritten_append_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + } +} + +bool FastAppend::CleanupAfterCommit() const { + // Cleanup after committing is disabled for FastAppend unless there are + // rewritten_append_manifests because: Review Comment: ```suggestion // rewritten_append_manifests_ because: ``` ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), + snapshot_manifests_span.end()); + manifests.insert(manifests.end(), snapshot_manifests.begin(), + snapshot_manifests.end()); + } + + return manifests; +} + +std::unordered_map<std::string, std::string> FastAppend::Summary() { + summary_.SetPartitionSummaryLimit( + base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); + return summary_.Build(); +} + +void FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) { + // Clean up new manifests that were written but not committed + if (!new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + new_manifests_.clear(); + } + + // Clean up only rewritten_append_manifests as they are always owned by the table + // Don't clean up append_manifests as they are added to the manifest list and are + // not compacted + if (!rewritten_append_manifests_.empty()) { + for (const auto& manifest : rewritten_append_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + } +} + +bool FastAppend::CleanupAfterCommit() const { + // Cleanup after committing is disabled for FastAppend unless there are + // rewritten_append_manifests because: + // 1.) Appended manifests are never rewritten + // 2.) Manifests which are written out as part of appendFile are already cleaned + // up between commit attempts in writeNewManifests + return !rewritten_append_manifests_.empty(); +} + +Result<std::shared_ptr<PartitionSpec>> FastAppend::Spec(int32_t spec_id) { + return base().PartitionSpecById(spec_id); +} + +Result<ManifestFile> FastAppend::CopyManifest(const ManifestFile& manifest) { + const TableMetadata& current = base(); + ICEBERG_ASSIGN_OR_RAISE(auto schema, current.Schema()); + ICEBERG_ASSIGN_OR_RAISE(auto spec, + current.PartitionSpecById(manifest.partition_spec_id)); + + // Read the manifest entries + ICEBERG_ASSIGN_OR_RAISE( + auto reader, + ManifestReader::Make(manifest, transaction_->table()->io(), schema, spec)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries()); + + // Create a new manifest writer + // Generate a unique manifest path using the transaction's metadata location + std::string filename = std::format("copy-m{}.avro", copy_manifest_count_++); + std::string new_manifest_path = transaction_->MetadataFileLocation(filename); + int64_t snapshot_id = SnapshotId(); + ICEBERG_ASSIGN_OR_RAISE( + auto writer, ManifestWriter::MakeWriter( + current.format_version, snapshot_id, new_manifest_path, + transaction_->table()->io(), spec, schema, ManifestContent::kData, + /*first_row_id=*/current.next_row_id)); + + // Write all entries as added entries with the new snapshot ID + for (auto& entry : entries) { + ICEBERG_PRECHECK(entry.status == ManifestStatus::kAdded, + "Manifest to copy must only contain added entries"); + entry.snapshot_id = snapshot_id; + ICEBERG_RETURN_UNEXPECTED(writer->WriteAddedEntry(entry)); + } + + ICEBERG_RETURN_UNEXPECTED(writer->Close()); + ICEBERG_ASSIGN_OR_RAISE(auto new_manifest, writer->ToManifestFile()); + + summary_.AddedManifest(new_manifest); Review Comment: This is just to copy a manifest file metadata and we shouldn't regard it as an added manifest. ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), + snapshot_manifests_span.end()); + manifests.insert(manifests.end(), snapshot_manifests.begin(), + snapshot_manifests.end()); + } + + return manifests; +} + +std::unordered_map<std::string, std::string> FastAppend::Summary() { + summary_.SetPartitionSummaryLimit( + base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); + return summary_.Build(); +} + +void FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) { + // Clean up new manifests that were written but not committed + if (!new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + new_manifests_.clear(); + } + + // Clean up only rewritten_append_manifests as they are always owned by the table + // Don't clean up append_manifests as they are added to the manifest list and are + // not compacted + if (!rewritten_append_manifests_.empty()) { + for (const auto& manifest : rewritten_append_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + } +} + +bool FastAppend::CleanupAfterCommit() const { + // Cleanup after committing is disabled for FastAppend unless there are + // rewritten_append_manifests because: + // 1.) Appended manifests are never rewritten + // 2.) Manifests which are written out as part of appendFile are already cleaned Review Comment: ```suggestion // 2.) Manifests which are written out as part of AppendFile are already cleaned ``` ########## src/iceberg/update/fast_append.h: ########## @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +/// \file iceberg/update/fast_append.h + +#include <functional> +#include <memory> +#include <string> +#include <unordered_map> +#include <unordered_set> +#include <vector> + +#include "iceberg/iceberg_export.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_list.h" +#include "iceberg/result.h" +#include "iceberg/snapshot.h" +#include "iceberg/type_fwd.h" +#include "iceberg/update/append_files.h" +#include "iceberg/update/snapshot_update.h" + +namespace iceberg { + +/// \brief Append implementation that adds new manifest files for writes. +/// +/// FastAppend is optimized for appending new data files to a table, it creates new +/// manifest files for the added data without compacting or rewriting existing manifests, +/// making it faster for write-heavy workloads. +class ICEBERG_EXPORT FastAppend : public SnapshotUpdate, public AppendFiles { + public: + /// \brief Create a new FastAppend instance. + /// + /// \param table_name The name of the table + /// \param transaction The transaction to use for this update + /// \return A Result containing the FastAppend instance or an error + static Result<std::unique_ptr<FastAppend>> Make( + std::string table_name, std::shared_ptr<Transaction> transaction); + + /// \brief Append a data file to this update. + /// + /// \param file The data file to append + /// \return Reference to this for method chaining + FastAppend& AppendFile(std::shared_ptr<DataFile> file) override; + + /// \brief Append a manifest file to this update. + /// + /// The manifest must only contain added files (no existing or deleted files). + /// If the manifest doesn't have a snapshot ID assigned and snapshot ID inheritance + /// is enabled, it will be used directly. Otherwise, it will be copied with the + /// new snapshot ID. + /// + /// \param manifest The manifest file to append + /// \return Reference to this for method chaining + FastAppend& AppendManifest(const ManifestFile& manifest) override; + + /// \brief Set the target branch for this update. + /// + /// \param branch The branch name + /// \return Reference to this for method chaining + FastAppend& ToBranch(const std::string& branch); + + /// \brief Set a summary property. + /// + /// \param property The property name + /// \param value The property value + /// \return Reference to this for method chaining + FastAppend& Set(const std::string& property, const std::string& value); + + Kind kind() const override { return Kind::kUpdateSnapshot; } Review Comment: Should this be moved to the base SnapshotUpdate class? ########## src/iceberg/update/fast_append.h: ########## @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +/// \file iceberg/update/fast_append.h + +#include <functional> +#include <memory> +#include <string> +#include <unordered_map> +#include <unordered_set> +#include <vector> + +#include "iceberg/iceberg_export.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_list.h" +#include "iceberg/result.h" +#include "iceberg/snapshot.h" +#include "iceberg/type_fwd.h" +#include "iceberg/update/append_files.h" +#include "iceberg/update/snapshot_update.h" + +namespace iceberg { + +/// \brief Append implementation that adds new manifest files for writes. +/// +/// FastAppend is optimized for appending new data files to a table, it creates new +/// manifest files for the added data without compacting or rewriting existing manifests, +/// making it faster for write-heavy workloads. +class ICEBERG_EXPORT FastAppend : public SnapshotUpdate, public AppendFiles { + public: + /// \brief Create a new FastAppend instance. + /// + /// \param table_name The name of the table + /// \param transaction The transaction to use for this update + /// \return A Result containing the FastAppend instance or an error + static Result<std::unique_ptr<FastAppend>> Make( + std::string table_name, std::shared_ptr<Transaction> transaction); + + /// \brief Append a data file to this update. + /// + /// \param file The data file to append + /// \return Reference to this for method chaining + FastAppend& AppendFile(std::shared_ptr<DataFile> file) override; + + /// \brief Append a manifest file to this update. + /// + /// The manifest must only contain added files (no existing or deleted files). + /// If the manifest doesn't have a snapshot ID assigned and snapshot ID inheritance + /// is enabled, it will be used directly. Otherwise, it will be copied with the + /// new snapshot ID. + /// + /// \param manifest The manifest file to append + /// \return Reference to this for method chaining + FastAppend& AppendManifest(const ManifestFile& manifest) override; + + /// \brief Set the target branch for this update. + /// + /// \param branch The branch name + /// \return Reference to this for method chaining + FastAppend& ToBranch(const std::string& branch); + + /// \brief Set a summary property. + /// + /// \param property The property name + /// \param value The property value + /// \return Reference to this for method chaining + FastAppend& Set(const std::string& property, const std::string& value); Review Comment: Should we directly define this as a virtual function in the `SnapshotUpdate`? ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); Review Comment: ```suggestion rewritten_append_manifests_.push_back(std::move(copied_manifest)); ``` ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); Review Comment: Reserve the capacity of it? ########## src/iceberg/update/fast_append.h: ########## @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +/// \file iceberg/update/fast_append.h + +#include <functional> +#include <memory> +#include <string> +#include <unordered_map> +#include <unordered_set> +#include <vector> + +#include "iceberg/iceberg_export.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_list.h" +#include "iceberg/result.h" +#include "iceberg/snapshot.h" +#include "iceberg/type_fwd.h" +#include "iceberg/update/append_files.h" +#include "iceberg/update/snapshot_update.h" + +namespace iceberg { + +/// \brief Append implementation that adds new manifest files for writes. +/// +/// FastAppend is optimized for appending new data files to a table, it creates new +/// manifest files for the added data without compacting or rewriting existing manifests, +/// making it faster for write-heavy workloads. +class ICEBERG_EXPORT FastAppend : public SnapshotUpdate, public AppendFiles { + public: + /// \brief Create a new FastAppend instance. + /// + /// \param table_name The name of the table + /// \param transaction The transaction to use for this update + /// \return A Result containing the FastAppend instance or an error + static Result<std::unique_ptr<FastAppend>> Make( + std::string table_name, std::shared_ptr<Transaction> transaction); + + /// \brief Append a data file to this update. + /// + /// \param file The data file to append + /// \return Reference to this for method chaining + FastAppend& AppendFile(std::shared_ptr<DataFile> file) override; + + /// \brief Append a manifest file to this update. + /// + /// The manifest must only contain added files (no existing or deleted files). + /// If the manifest doesn't have a snapshot ID assigned and snapshot ID inheritance + /// is enabled, it will be used directly. Otherwise, it will be copied with the + /// new snapshot ID. + /// + /// \param manifest The manifest file to append + /// \return Reference to this for method chaining + FastAppend& AppendManifest(const ManifestFile& manifest) override; + + /// \brief Set the target branch for this update. + /// + /// \param branch The branch name + /// \return Reference to this for method chaining + FastAppend& ToBranch(const std::string& branch); + + /// \brief Set a summary property. + /// + /// \param property The property name + /// \param value The property value + /// \return Reference to this for method chaining + FastAppend& Set(const std::string& property, const std::string& value); + + Kind kind() const override { return Kind::kUpdateSnapshot; } + + std::string operation() override; + + Result<std::vector<ManifestFile>> Apply( + const TableMetadata& metadata_to_update, + const std::shared_ptr<Snapshot>& snapshot) override; + std::unordered_map<std::string, std::string> Summary() override; + void CleanUncommitted(const std::unordered_set<std::string>& committed) override; + bool CleanupAfterCommit() const override; + + private: + explicit FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction); + + /// \brief Get the partition spec by spec ID. + Result<std::shared_ptr<PartitionSpec>> Spec(int32_t spec_id); + + /// \brief Copy a manifest file with a new snapshot ID. + /// + /// \param manifest The manifest to copy + /// \return The copied manifest file + Result<ManifestFile> CopyManifest(const ManifestFile& manifest); + + /// \brief Write new manifests for the accumulated data files. + /// + /// \return A vector of manifest files, or an error + Result<std::vector<ManifestFile>> WriteNewManifests(); + + private: + struct DataFilePtrHash { + size_t operator()(const std::shared_ptr<DataFile>& file) const { + if (!file) { + return 0; + } + return std::hash<std::string>{}(file->file_path); + } + }; + + struct DataFilePtrEqual { + bool operator()(const std::shared_ptr<DataFile>& left, + const std::shared_ptr<DataFile>& right) const { + if (left == right) { + return true; + } + if (!left || !right) { + return false; + } + return left->file_path == right->file_path; + } + }; + + std::string table_name_; + SnapshotSummaryBuilder summary_; + std::unordered_map<int32_t, std::unordered_set<std::shared_ptr<DataFile>, Review Comment: Do you think it is worth adding a `DataFileSet` to `content_file_util.h` because it will be used by multiple places? The Java DataFileSet preserves the insertion order of DataFiles which might be useful for row id in v3. ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { Review Comment: ```suggestion if (can_inherit_snapshot_id() && manifest.added_snapshot_id == kInvalidSnapshotId) { ``` ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), + snapshot_manifests_span.end()); + manifests.insert(manifests.end(), snapshot_manifests.begin(), + snapshot_manifests.end()); + } + + return manifests; +} + +std::unordered_map<std::string, std::string> FastAppend::Summary() { + summary_.SetPartitionSummaryLimit( + base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); + return summary_.Build(); +} + +void FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) { + // Clean up new manifests that were written but not committed + if (!new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + new_manifests_.clear(); + } + + // Clean up only rewritten_append_manifests as they are always owned by the table + // Don't clean up append_manifests as they are added to the manifest list and are Review Comment: ```suggestion // Don't clean up append manifests as they are added to the manifest list and are ``` ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), Review Comment: Use `std::make_move_iterator` on new_written_manifests ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), + snapshot_manifests_span.end()); + manifests.insert(manifests.end(), snapshot_manifests.begin(), + snapshot_manifests.end()); + } + + return manifests; +} + +std::unordered_map<std::string, std::string> FastAppend::Summary() { + summary_.SetPartitionSummaryLimit( + base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); + return summary_.Build(); +} + +void FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) { + // Clean up new manifests that were written but not committed + if (!new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + new_manifests_.clear(); + } + + // Clean up only rewritten_append_manifests as they are always owned by the table + // Don't clean up append_manifests as they are added to the manifest list and are + // not compacted + if (!rewritten_append_manifests_.empty()) { + for (const auto& manifest : rewritten_append_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + } +} + +bool FastAppend::CleanupAfterCommit() const { + // Cleanup after committing is disabled for FastAppend unless there are + // rewritten_append_manifests because: + // 1.) Appended manifests are never rewritten + // 2.) Manifests which are written out as part of appendFile are already cleaned + // up between commit attempts in writeNewManifests + return !rewritten_append_manifests_.empty(); +} + +Result<std::shared_ptr<PartitionSpec>> FastAppend::Spec(int32_t spec_id) { + return base().PartitionSpecById(spec_id); +} + +Result<ManifestFile> FastAppend::CopyManifest(const ManifestFile& manifest) { + const TableMetadata& current = base(); + ICEBERG_ASSIGN_OR_RAISE(auto schema, current.Schema()); + ICEBERG_ASSIGN_OR_RAISE(auto spec, + current.PartitionSpecById(manifest.partition_spec_id)); + + // Read the manifest entries + ICEBERG_ASSIGN_OR_RAISE( + auto reader, + ManifestReader::Make(manifest, transaction_->table()->io(), schema, spec)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries()); + + // Create a new manifest writer + // Generate a unique manifest path using the transaction's metadata location + std::string filename = std::format("copy-m{}.avro", copy_manifest_count_++); Review Comment: Shouldn't we call `SnapshotUpdate::ManifestPath()` ? ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), Review Comment: Can we avoid the creation of `snapshot_manifests`? ########## src/iceberg/update/fast_append.cc: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/fast_append.h" + +#include <format> +#include <iterator> +#include <ranges> +#include <vector> + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/manifest/manifest_writer.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<FastAppend>> FastAppend::Make( + std::string table_name, std::shared_ptr<Transaction> transaction) { + return std::unique_ptr<FastAppend>( + new FastAppend(std::move(table_name), std::move(transaction))); +} + +FastAppend::FastAppend(std::string table_name, std::shared_ptr<Transaction> transaction) + : SnapshotUpdate(std::move(transaction)), table_name_(std::move(table_name)) {} + +FastAppend& FastAppend::AppendFile(std::shared_ptr<DataFile> file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, Spec(spec_id)); + + auto& data_files = new_data_files_by_spec_[spec_id]; + auto [iter, inserted] = data_files.insert(file); + if (inserted) { + has_new_files_ = true; + ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file)); + } + + return *this; +} + +FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) { + ICEBERG_BUILDER_CHECK(!manifest.has_existing_files(), + "Cannot append manifest with existing files"); + ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(), + "Cannot append manifest with deleted files"); + ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId, + "Snapshot id must be assigned during commit"); + ICEBERG_BUILDER_CHECK(manifest.sequence_number == TableMetadata::kInvalidSequenceNumber, + "Sequence number must be assigned during commit"); + + if (can_inherit_snapshot_id() && (manifest.added_snapshot_id == kInvalidSnapshotId)) { + summary_.AddedManifest(manifest); + append_manifests_.push_back(manifest); + } else { + // The manifest must be rewritten with this update's snapshot ID + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest)); + rewritten_append_manifests_.push_back(copied_manifest); + } + + return *this; +} + +FastAppend& FastAppend::ToBranch(const std::string& branch) { + ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch)); + return *this; +} + +FastAppend& FastAppend::Set(const std::string& property, const std::string& value) { + summary_.Set(property, value); + return *this; +} + +std::string FastAppend::operation() { return DataOperation::kAppend; } + +Result<std::vector<ManifestFile>> FastAppend::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& snapshot) { + std::vector<ManifestFile> manifests; + + ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); + if (!new_written_manifests.empty()) { + manifests.insert(manifests.end(), new_written_manifests.begin(), + new_written_manifests.end()); + } + + // Transform append manifests and rewritten append manifests with snapshot ID + int64_t snapshot_id = SnapshotId(); + for (const auto& manifest : append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + for (const auto& manifest : rewritten_append_manifests_) { + ManifestFile updated = manifest; + updated.added_snapshot_id = snapshot_id; + manifests.push_back(updated); + } + + // Add all manifests from the snapshot + if (snapshot != nullptr) { + // Use SnapshotCache to get manifests, similar to snapshot_update.cc + auto cached_snapshot = SnapshotCache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests_span, + cached_snapshot.Manifests(transaction_->table()->io())); + std::vector<ManifestFile> snapshot_manifests(snapshot_manifests_span.begin(), + snapshot_manifests_span.end()); + manifests.insert(manifests.end(), snapshot_manifests.begin(), + snapshot_manifests.end()); + } + + return manifests; +} + +std::unordered_map<std::string, std::string> FastAppend::Summary() { + summary_.SetPartitionSummaryLimit( + base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); + return summary_.Build(); +} + +void FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) { + // Clean up new manifests that were written but not committed + if (!new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + new_manifests_.clear(); + } + + // Clean up only rewritten_append_manifests as they are always owned by the table + // Don't clean up append_manifests as they are added to the manifest list and are + // not compacted + if (!rewritten_append_manifests_.empty()) { + for (const auto& manifest : rewritten_append_manifests_) { + if (committed.find(manifest.manifest_path) == committed.end()) { + std::ignore = DeleteFile(manifest.manifest_path); + } + } + } +} + +bool FastAppend::CleanupAfterCommit() const { + // Cleanup after committing is disabled for FastAppend unless there are + // rewritten_append_manifests because: + // 1.) Appended manifests are never rewritten + // 2.) Manifests which are written out as part of appendFile are already cleaned + // up between commit attempts in writeNewManifests + return !rewritten_append_manifests_.empty(); +} + +Result<std::shared_ptr<PartitionSpec>> FastAppend::Spec(int32_t spec_id) { + return base().PartitionSpecById(spec_id); +} + +Result<ManifestFile> FastAppend::CopyManifest(const ManifestFile& manifest) { + const TableMetadata& current = base(); + ICEBERG_ASSIGN_OR_RAISE(auto schema, current.Schema()); + ICEBERG_ASSIGN_OR_RAISE(auto spec, + current.PartitionSpecById(manifest.partition_spec_id)); + + // Read the manifest entries + ICEBERG_ASSIGN_OR_RAISE( + auto reader, + ManifestReader::Make(manifest, transaction_->table()->io(), schema, spec)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries()); + + // Create a new manifest writer + // Generate a unique manifest path using the transaction's metadata location + std::string filename = std::format("copy-m{}.avro", copy_manifest_count_++); + std::string new_manifest_path = transaction_->MetadataFileLocation(filename); + int64_t snapshot_id = SnapshotId(); + ICEBERG_ASSIGN_OR_RAISE( + auto writer, ManifestWriter::MakeWriter( + current.format_version, snapshot_id, new_manifest_path, + transaction_->table()->io(), spec, schema, ManifestContent::kData, + /*first_row_id=*/current.next_row_id)); + + // Write all entries as added entries with the new snapshot ID + for (auto& entry : entries) { + ICEBERG_PRECHECK(entry.status == ManifestStatus::kAdded, + "Manifest to copy must only contain added entries"); + entry.snapshot_id = snapshot_id; + ICEBERG_RETURN_UNEXPECTED(writer->WriteAddedEntry(entry)); + } + + ICEBERG_RETURN_UNEXPECTED(writer->Close()); + ICEBERG_ASSIGN_OR_RAISE(auto new_manifest, writer->ToManifestFile()); + + summary_.AddedManifest(new_manifest); + + return new_manifest; +} + +Result<std::vector<ManifestFile>> FastAppend::WriteNewManifests() { + // If there are new files and manifests were already written, clean them up + if (has_new_files_ && !new_manifests_.empty()) { + for (const auto& manifest : new_manifests_) { + ICEBERG_RETURN_UNEXPECTED(DeleteFile(manifest.manifest_path)); Review Comment: Ignore the error? -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
