shangxinli commented on code in PR #776: URL: https://github.com/apache/iceberg-cpp/pull/776#discussion_r3641298872
########## src/iceberg/update/replace_partitions.h: ########## @@ -0,0 +1,134 @@ +/* + * 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/replace_partitions.h + +#include <cstdint> +#include <memory> +#include <optional> +#include <string> + +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" +#include "iceberg/update/merging_snapshot_update.h" +#include "iceberg/util/partition_value_util.h" + +namespace iceberg { + +/// \brief Replaces partitions in a table with new data files. +/// +/// ReplacePartitions dynamically identifies which partitions to overwrite based +/// on the data files added via AddFile(). All existing data files in each +/// touched partition are marked DELETED, and the new files are written as the +/// sole data in those partitions. Partitions not referenced by any added file +/// are left unchanged. +/// +/// This operation produces a snapshot with operation="overwrite" and +/// "replace-partitions"="true" in the summary. For unpartitioned tables, all +/// existing files are replaced. +/// +/// When committing, these changes are applied to the latest table snapshot. +/// Commit conflicts are resolved by re-applying to the new latest snapshot +/// and reattempting the commit. +/// +/// \note This is provided to implement SQL compatible with Hive table +/// operations but is not recommended. Instead, use OverwriteFiles to +/// explicitly overwrite data. +class ICEBERG_EXPORT ReplacePartitions : public MergingSnapshotUpdate { + public: + /// \brief Create a new ReplacePartitions instance. + /// + /// \param table_name The name of the table + /// \param ctx The transaction context + /// \return A Result containing the ReplacePartitions instance or an error + static Result<std::unique_ptr<ReplacePartitions>> Make( + std::string table_name, std::shared_ptr<TransactionContext> ctx); + + /// \brief Add a data file and mark its partition for replacement. + /// + /// Each call registers the file's partition so all existing data files in + /// that partition are replaced. Duplicate files (same path) are ignored. + /// + /// \param file The data file to add (must have partition_spec_id set) + /// \return Reference to this for method chaining + ReplacePartitions& AddFile(const std::shared_ptr<DataFile>& file); + + /// \brief Fail the commit if any existing data file would be deleted. + /// + /// This validation is useful to ensure the operation is only applied to + /// tables where no data currently exists in the affected partitions. + /// + /// \return Reference to this for method chaining + ReplacePartitions& ValidateAppendOnly(); + + /// \brief Set the snapshot ID used as the baseline for conflict validation. + /// + /// Validations check changes that occurred after this snapshot ID. If not + /// set, all ancestor snapshots through the initial snapshot are validated. + /// + /// \param snapshot_id A snapshot ID + /// \return Reference to this for method chaining + ReplacePartitions& ValidateFromSnapshot(int64_t snapshot_id); + + /// \brief Enable validation that no conflicting data files were added concurrently. + /// + /// Fails the commit if a concurrent operation added a data file in any of + /// the partitions being replaced after the snapshot set by + /// ValidateFromSnapshot(). + /// + /// \return Reference to this for method chaining + ReplacePartitions& ValidateNoConflictingData(); + + /// \brief Enable validation that no conflicting delete files were added concurrently. + /// + /// Fails the commit if a concurrent operation added a delete file covering + /// any of the partitions being replaced after the snapshot set by + /// ValidateFromSnapshot(). + /// + /// \return Reference to this for method chaining + ReplacePartitions& ValidateNoConflictingDeletes(); + + std::string operation() override; + + protected: + Status Validate(const TableMetadata& current_metadata, + const std::shared_ptr<Snapshot>& snapshot) override; + + private: + explicit ReplacePartitions(std::string table_name, + std::shared_ptr<TransactionContext> ctx); + + std::optional<int64_t> starting_snapshot_id_; + bool validate_conflicting_data_{false}; + bool validate_conflicting_deletes_{false}; + // True once an AddFile() call has staged a file whose partition spec is Review Comment: Reworded to "True when replacement uses an AlwaysTrue row filter." ########## src/iceberg/update/replace_partitions.h: ########## @@ -0,0 +1,134 @@ +/* + * 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/replace_partitions.h + +#include <cstdint> +#include <memory> +#include <optional> +#include <string> + +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" +#include "iceberg/update/merging_snapshot_update.h" +#include "iceberg/util/partition_value_util.h" + +namespace iceberg { + +/// \brief Replaces partitions in a table with new data files. +/// +/// ReplacePartitions dynamically identifies which partitions to overwrite based +/// on the data files added via AddFile(). All existing data files in each +/// touched partition are marked DELETED, and the new files are written as the +/// sole data in those partitions. Partitions not referenced by any added file +/// are left unchanged. +/// +/// This operation produces a snapshot with operation="overwrite" and +/// "replace-partitions"="true" in the summary. For unpartitioned tables, all +/// existing files are replaced. +/// +/// When committing, these changes are applied to the latest table snapshot. +/// Commit conflicts are resolved by re-applying to the new latest snapshot +/// and reattempting the commit. +/// +/// \note This is provided to implement SQL compatible with Hive table +/// operations but is not recommended. Instead, use OverwriteFiles to +/// explicitly overwrite data. +class ICEBERG_EXPORT ReplacePartitions : public MergingSnapshotUpdate { + public: + /// \brief Create a new ReplacePartitions instance. + /// + /// \param table_name The name of the table + /// \param ctx The transaction context + /// \return A Result containing the ReplacePartitions instance or an error + static Result<std::unique_ptr<ReplacePartitions>> Make( + std::string table_name, std::shared_ptr<TransactionContext> ctx); + + /// \brief Add a data file and mark its partition for replacement. + /// + /// Each call registers the file's partition so all existing data files in + /// that partition are replaced. Duplicate files (same path) are ignored. + /// + /// \param file The data file to add (must have partition_spec_id set) + /// \return Reference to this for method chaining + ReplacePartitions& AddFile(const std::shared_ptr<DataFile>& file); + + /// \brief Fail the commit if any existing data file would be deleted. + /// + /// This validation is useful to ensure the operation is only applied to + /// tables where no data currently exists in the affected partitions. + /// + /// \return Reference to this for method chaining + ReplacePartitions& ValidateAppendOnly(); + + /// \brief Set the snapshot ID used as the baseline for conflict validation. + /// + /// Validations check changes that occurred after this snapshot ID. If not + /// set, all ancestor snapshots through the initial snapshot are validated. + /// + /// \param snapshot_id A snapshot ID + /// \return Reference to this for method chaining + ReplacePartitions& ValidateFromSnapshot(int64_t snapshot_id); + + /// \brief Enable validation that no conflicting data files were added concurrently. + /// + /// Fails the commit if a concurrent operation added a data file in any of + /// the partitions being replaced after the snapshot set by + /// ValidateFromSnapshot(). + /// + /// \return Reference to this for method chaining + ReplacePartitions& ValidateNoConflictingData(); + + /// \brief Enable validation that no conflicting delete files were added concurrently. Review Comment: Updated the doc to note this also rejects concurrent data-file removals, not just newly added delete files, so the replace can't undelete rows a concurrent commit removed. ########## src/iceberg/update/replace_partitions.cc: ########## @@ -0,0 +1,153 @@ +/* + * 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/replace_partitions.h" + +#include <algorithm> + +#include "iceberg/expression/expressions.h" +#include "iceberg/partition_spec.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" // IWYU pragma: keep +#include "iceberg/table_metadata.h" +#include "iceberg/transaction.h" +#include "iceberg/transform.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result<std::unique_ptr<ReplacePartitions>> ReplacePartitions::Make( + std::string table_name, std::shared_ptr<TransactionContext> ctx) { + ICEBERG_PRECHECK(!table_name.empty(), "Table name cannot be empty"); + ICEBERG_PRECHECK(ctx != nullptr, "Cannot create ReplacePartitions without a context"); + return std::unique_ptr<ReplacePartitions>( + new ReplacePartitions(std::move(table_name), std::move(ctx))); +} + +ReplacePartitions::ReplacePartitions(std::string table_name, + std::shared_ptr<TransactionContext> ctx) + : MergingSnapshotUpdate(std::move(table_name), std::move(ctx)) { + SetSummaryProperty(SnapshotSummaryFields::kReplacePartitions, "true"); +} + +ReplacePartitions& ReplacePartitions::AddFile(const 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, base().PartitionSpecById(spec_id)); + + ICEBERG_BUILDER_RETURN_IF_ERROR(AddDataFile(file)); + // A spec is effectively unpartitioned if it has no fields, or every field + // uses the void transform. Java's PartitionSpec.isUnpartitioned() covers + // both cases; mirror that so an all-void spec triggers the table-wide + // replace path instead of a DropPartition with void-valued partition keys. + auto is_unpartitioned = [](const PartitionSpec& s) { + return s.fields().empty() || + std::all_of(s.fields().begin(), s.fields().end(), [](const PartitionField& f) { + return f.transform()->transform_type() == TransformType::kVoid; + }); + }; + if (is_unpartitioned(*spec)) { + // Unpartitioned spec: Java's BaseReplacePartitions treats this as a + // table-wide replace rather than a spec-scoped DropPartition with empty + // partition values. Mirror that so every existing data file is dropped + // and conflict validation runs against AlwaysTrue. + ICEBERG_BUILDER_RETURN_IF_ERROR(DeleteByRowFilter(Expressions::AlwaysTrue())); + replace_by_row_filter_ = true; + } else { + ICEBERG_BUILDER_RETURN_IF_ERROR(DropPartition(spec_id, file->partition)); + replaced_partitions_.add(spec_id, file->partition); + } + return *this; +} + +ReplacePartitions& ReplacePartitions::ValidateAppendOnly() { + FailAnyDelete(); + return *this; +} + +ReplacePartitions& ReplacePartitions::ValidateFromSnapshot(int64_t snapshot_id) { + starting_snapshot_id_ = snapshot_id; + return *this; +} + +ReplacePartitions& ReplacePartitions::ValidateNoConflictingData() { + validate_conflicting_data_ = true; + return *this; +} + +ReplacePartitions& ReplacePartitions::ValidateNoConflictingDeletes() { + validate_conflicting_deletes_ = true; + return *this; +} + +std::string ReplacePartitions::operation() { return DataOperation::kOverwrite; } + +Status ReplacePartitions::Validate(const TableMetadata& current_metadata, Review Comment: Added src/iceberg/test/replace_partitions_test.cc with direct tests: partitioned and unpartitioned/all-void replacement, empty and mixed specs, append-only validation (pass + translated conflict), and same/different-partition data and delete concurrency. Retry-summary retention and DV cleanup run through the base MergingSnapshotUpdate paths, which have their own coverage; happy to add ReplacePartitions-specific cases for those too if you'd prefer. -- 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]
