SteNicholas commented on code in PR #222: URL: https://github.com/apache/paimon-cpp/pull/222#discussion_r3841132119
########## src/paimon/core/table/format/format_table_write.cpp: ########## @@ -0,0 +1,616 @@ +/* + * 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 "paimon/table/format/format_table_write.h" + +#include <algorithm> +#include <map> +#include <optional> +#include <string> +#include <string_view> +#include <utility> +#include <vector> + +#include "arrow/api.h" +#include "arrow/c/bridge.h" +#include "arrow/c/helpers.h" +#include "fmt/format.h" +#include "paimon/common/data/binary_row.h" +#include "paimon/common/utils/arrow/arrow_utils.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/binary_row_partition_computer.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/field_type_utils.h" +#include "paimon/common/utils/hadoop_compression.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/casting/cast_executor.h" +#include "paimon/core/casting/cast_executor_factory.h" +#include "paimon/core/casting/casting_utils.h" +#include "paimon/core/core_options.h" +#include "paimon/core/table/format/format_file_naming.h" +#include "paimon/core/table/format/format_path_validation.h" +#include "paimon/defs.h" +#include "paimon/format/file_format.h" +#include "paimon/format/file_format_factory.h" +#include "paimon/format/format_writer.h" +#include "paimon/format/writer_builder.h" +#include "paimon/fs/file_system.h" +#include "paimon/logging.h" + +namespace paimon { + +namespace { + +Logger* WriteLogger() { + static std::unique_ptr<Logger> logger = Logger::GetLogger("FormatTableWrite"); + return logger.get(); +} + +/// The extension a compression adds to a data file's name: a hadoop compression by its own +/// extension, anything else by the option's text. +std::string CompressionFileExtension(const std::string& compression) { + if (compression.empty()) { + return std::string(); + } + std::optional<HadoopCompression::Kind> kind = HadoopCompression::FromName(compression); + if (kind) { + return HadoopCompression::ToFileExtension(*kind); + } + return compression; +} + +/// Renders a partition column as the text a partition directory is named with. +Result<std::shared_ptr<arrow::StringArray>> RenderPartitionColumnAsText( + const std::shared_ptr<arrow::Array>& column, const std::string& field_name, + bool legacy_partition_name, arrow::MemoryPool* pool) { + if (column->type_id() == arrow::Type::STRING) { + return checked_pointer_cast<arrow::StringArray>(column); + } + std::shared_ptr<arrow::Array> source = column; + // `partition.legacy-name` renders with the type's own `toString`, which for a DATE is the + // day count rather than `YYYY-MM-DD`. DATE is the only partition type the two disagree on. + // `DataConverterUtils` answers it the same way for the managed table path. + if (legacy_partition_name && column->type_id() == arrow::Type::DATE32) { + PAIMON_ASSIGN_OR_RAISE( + source, + CastingUtils::Cast(column, arrow::int32(), arrow::compute::CastOptions::Safe(), pool)); + } + PAIMON_ASSIGN_OR_RAISE(FieldType source_type, + FieldTypeUtils::ConvertToFieldType(source->type()->id())); + std::shared_ptr<CastExecutor> cast_executor = + CastExecutorFactory::GetCastExecutorFactory()->GetCastExecutor(source_type, + FieldType::STRING); + if (cast_executor == nullptr) { + return Status::NotImplemented( + fmt::format("cannot name a partition directory after field '{}' of type {}", field_name, + column->type()->ToString())); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> casted, + cast_executor->Cast(source, arrow::utf8(), pool)); + return checked_pointer_cast<arrow::StringArray>(casted); +} + +} // namespace + +/// Where one partition's files go, and the partition values that directory spells out. +struct FormatTablePartitionTarget { + std::string directory; + /// The partition as the table renders it, which need not be how the caller spelled it. The + /// commit message carries these, so it cannot disagree with the directory its file sits in. + std::map<std::string, std::string> partition; +}; + +/// The file currently being written for one partition. +struct FormatTableWriteFile { + std::shared_ptr<OutputStream> out; + std::unique_ptr<FormatWriter> writer; + std::string temp_file_path; + std::string file_path; + int64_t record_count = 0; +}; + +class FormatTableWrite::Impl { + public: + /// Where a partition's files belong. Cached, since deriving it reads the values into their + /// column types, renders them back out, escapes them and re-walks a whole path. + Result<FormatTablePartitionTarget> GetPartitionTarget( + const std::map<std::string, std::string>& partition); + + /// Opens a new hidden file in `directory` for the partition that directory stands for. + Result<FormatTableWriteFile> OpenFile(const std::string& directory); + + /// Closes the file open in `directory` and records it for committing. + Status FinishFile(const std::string& directory); + + /// Checks that every row belongs to the partition the batch declares. Partition columns are + /// not written, so a disagreeing row would read back with the declared value and lose its + /// own. + Status ValidatePartitionColumns( + const std::shared_ptr<arrow::StructArray>& batch, + const std::vector<std::pair<std::string, std::string>>& ordered_partition); + + std::shared_ptr<FormatTable> table; + std::shared_ptr<MemoryPool> pool; + std::unique_ptr<arrow::MemoryPool> arrow_pool; + /// Full table schema, used to check the incoming batch. + std::shared_ptr<arrow::Schema> table_schema; + std::shared_ptr<arrow::DataType> table_struct_type; + /// Columns actually stored in the files: the table's, minus the partition ones. + std::shared_ptr<arrow::Schema> data_schema; + /// Index in the table schema of each data column. + std::vector<int32_t> data_column_indexes; + /// Index in the table schema of each partition column, in partition key order. + std::vector<int32_t> partition_column_indexes; + std::string format_identifier; + std::string file_compression; + /// From `partition.legacy-name`. It decides how a row's partition column is rendered, so the + /// directory name and the row check both go by it. + bool legacy_partition_name = true; + /// Reads a partition into its column types and renders it back out, the way Java Paimon's + /// writer does. Null when the table is not partitioned. + std::unique_ptr<BinaryRowPartitionComputer> partition_computer; + int64_t target_file_size = 0; + int64_t target_file_row_num = 0; + int32_t write_batch_size = 0; + FormatFileNaming naming; + + /// Keyed by the partition the caller declared. See `GetPartitionTarget()`. + std::map<std::map<std::string, std::string>, FormatTablePartitionTarget> partition_targets; + /// Open file per partition, keyed by the partition's directory. + std::map<std::string, FormatTableWriteFile> open_files; + /// Partition values of each open file, by the same key. + std::map<std::string, std::map<std::string, std::string>> open_partitions; + /// Written and closed but not yet published. Kept after `PrepareCommit()` hands out a copy, + /// so that an `Abort()` still knows what to remove. + std::vector<FormatCommitMessage> staged_messages; + bool prepared = false; + bool aborted = false; + + /// Why this write will take no more rows, or null while it still will. Prepared and aborted + /// call for different work from the caller, so the refusal names which one it is. + const char* FinishedReason() const { + if (aborted) { + return "format table write has been aborted"; + } + if (prepared) { + return "format table write has already prepared its commit"; + } + return nullptr; + } +}; + +FormatTableWrite::FormatTableWrite(std::unique_ptr<Impl> impl) : impl_(std::move(impl)) {} + +FormatTableWrite::~FormatTableWrite() { + if (impl_ != nullptr && impl_->FinishedReason() == nullptr) { + // `Abort()` logs its cleanup failures and returns OK today; the status is still checked. + Status status = Abort(); + if (!status.ok()) { + PAIMON_LOG_WARN(WriteLogger(), "Failed to abort an abandoned write of table %s: %s", + impl_->table->FullName().c_str(), status.ToString().c_str()); + } + } +} + +Result<std::unique_ptr<FormatTableWrite>> FormatTableWrite::Create( + const std::shared_ptr<FormatTable>& table, const std::shared_ptr<MemoryPool>& pool) { + if (table == nullptr) { + return Status::Invalid("format table write requires a table"); + } + auto impl = std::make_unique<Impl>(); + impl->table = table; + impl->pool = pool != nullptr ? pool : GetDefaultPool(); + + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<::ArrowSchema> c_schema, table->GetArrowSchema()); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(impl->table_schema, arrow::ImportSchema(c_schema.get())); + impl->table_struct_type = arrow::struct_(impl->table_schema->fields()); + + const std::vector<std::string>& partition_keys = table->PartitionKeys(); + arrow::FieldVector data_fields; + for (int32_t i = 0; i < impl->table_schema->num_fields(); i++) { + const std::shared_ptr<arrow::Field>& field = impl->table_schema->field(i); + if (std::find(partition_keys.begin(), partition_keys.end(), field->name()) == + partition_keys.end()) { + data_fields.push_back(field); + impl->data_column_indexes.push_back(i); + } + } + if (data_fields.empty()) { + return Status::Invalid(fmt::format( + "format table {} has no non-partition column, so its files would hold nothing", + table->FullName())); + } + impl->data_schema = arrow::schema(data_fields); + + // In partition key order, which is the order the directories nest in. + for (const std::string& partition_key : partition_keys) { + int32_t index = impl->table_schema->GetFieldIndex(partition_key); + if (index < 0) { + return Status::Invalid(fmt::format("partition field '{}' is not a column of table {}", + partition_key, table->FullName())); + } + impl->partition_column_indexes.push_back(index); + } + + impl->format_identifier = FormatTable::FormatToString(table->GetFormat()); + impl->file_compression = table->FileCompression(); + + // Through `CoreOptions`, so `"256 mb"` means what it does elsewhere and a default lives in + // one place. + PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, + CoreOptions::FromMap(table->Options(), table->GetFileSystem())); + + // parquet and orc record their compression inside the file, so the name keeps the plain + // `.parquet` unless `file.suffix.include.compression` asks for it, and then the compression + // goes in front: `data-<uuid>-0.snappy.parquet`. + const std::string compression_extension = CompressionFileExtension(impl->file_compression); + std::string extension = impl->format_identifier; + if (!compression_extension.empty() && core_options.FileSuffixIncludeCompression()) { + extension = compression_extension + "." + extension; + } + + impl->legacy_partition_name = core_options.LegacyPartitionNameEnabled(); + if (!partition_keys.empty()) { + PAIMON_ASSIGN_OR_RAISE( + impl->partition_computer, + BinaryRowPartitionComputer::Create(partition_keys, impl->table_schema, + table->PartitionDefaultName(), + impl->legacy_partition_name, impl->pool)); + } + // A format table has no primary keys, so its target file size is the append-table default. + impl->target_file_size = core_options.GetTargetFileSize(/*has_primary_key=*/false); + impl->target_file_row_num = core_options.GetTargetFileRowNum(); + impl->write_batch_size = core_options.GetWriteBatchSize(); + impl->arrow_pool = GetArrowPool(impl->pool); + PAIMON_ASSIGN_OR_RAISE(impl->naming, + FormatFileNaming::Create(extension, core_options.DataFilePrefix())); + + return std::unique_ptr<FormatTableWrite>(new FormatTableWrite(std::move(impl))); +} + +Result<FormatTablePartitionTarget> FormatTableWrite::Impl::GetPartitionTarget( + const std::map<std::string, std::string>& partition) { + auto iter = partition_targets.find(partition); + if (iter != partition_targets.end()) { + return iter->second; + } + FormatTablePartitionTarget target; + target.partition = partition; + if (partition_computer != nullptr) { + // The round trip Java Paimon's writer makes when it renders a partition out of a row: the + // values are read into their column types and rendered back, so the directory is named the + // way the table's options say rather than the way the caller spelled the value. + PAIMON_ASSIGN_OR_RAISE(BinaryRow row, partition_computer->ToBinaryRow(partition)); + // Aliased, or the comma inside the type would read as a second macro argument. + using RenderedPartition = std::vector<std::pair<std::string, std::string>>; + PAIMON_ASSIGN_OR_RAISE(RenderedPartition rendered, + partition_computer->GeneratePartitionVector(row)); + target.partition.clear(); + for (auto& [key, value] : rendered) { + target.partition.emplace(std::move(key), std::move(value)); + } + } + PAIMON_ASSIGN_OR_RAISE(target.directory, + FormatPathValidation::BuildPartitionDirectory(table, target.partition)); + partition_targets.emplace(partition, target); + return target; +} + +Status FormatTableWrite::Impl::ValidatePartitionColumns( + const std::shared_ptr<arrow::StructArray>& batch, + const std::vector<std::pair<std::string, std::string>>& ordered_partition) { + // A null, an empty string and a whitespace-only string alike stand for the default partition + // name, so all three land in the same directory. + const std::string& default_partition_name = table->PartitionDefaultName(); + for (size_t i = 0; i < ordered_partition.size(); i++) { + const std::string& partition_key = ordered_partition[i].first; + const std::string& declared_value = ordered_partition[i].second; + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr<arrow::StringArray> text_column, + RenderPartitionColumnAsText(batch->field(partition_column_indexes[i]), partition_key, + legacy_partition_name, arrow_pool.get())); + // False for every partition the table rendered, since `GeneratePartitionVector()` has + // already replaced a null or blank value with the default partition name. Kept so that a + // value reaching here without that round trip cannot take the fast path below, where a + // blank has to compare equal to the default partition name rather than to itself. + const bool declared_is_blank = StringUtils::IsNullOrWhitespaceOnly(declared_value); + for (int64_t row = 0; row < text_column->length(); row++) { + const bool is_null = text_column->IsNull(row); + const std::string_view rendered = + is_null ? std::string_view() : text_column->GetView(row); + // The ordinary case: a row that renders exactly as the batch declared. + if (!is_null && !declared_is_blank && rendered == declared_value) { + continue; + } + const std::string_view row_value = StringUtils::IsNullOrWhitespaceOnly(rendered) + ? std::string_view(default_partition_name) + : rendered; + if (row_value == declared_value) { + continue; + } + return Status::Invalid(fmt::format( + "row {} of the batch has '{}' in partition column '{}', but the batch declares " + "partition '{}={}'. The partition columns are not written to the file, so this " + "row would be stored under a partition it does not belong to and read back " + "with the declared value.", + row, row_value, partition_key, partition_key, declared_value)); + } + } + return Status::OK(); +} + +Result<FormatTableWriteFile> FormatTableWrite::Impl::OpenFile(const std::string& directory) { + FormatTableWriteFile file; + file.file_path = PathUtil::JoinPath(directory, naming.NextFileName()); + PAIMON_ASSIGN_OR_RAISE(std::string temp_relative_path, naming.NextTempFilePath()); + file.temp_file_path = PathUtil::JoinPath(directory, temp_relative_path); + + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileFormat> file_format, + FileFormatFactory::Get(format_identifier, table->Options())); + ::ArrowSchema c_schema; + ArrowSchemaMarkReleased(&c_schema); + ScopeGuard schema_guard([&c_schema]() { ArrowSchemaRelease(&c_schema); }); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*data_schema, &c_schema)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriterBuilder> writer_builder, + file_format->CreateWriterBuilder(&c_schema, write_batch_size)); + writer_builder->WithMemoryPool(pool); + + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<OutputStream> out, + table->GetFileSystem()->Create(file.temp_file_path, /*overwrite=*/false)); + file.out = std::move(out); + // From here the temp file exists on disk and only this guard knows about it, so a failure + // below has to remove it or it is left behind for good. + std::shared_ptr<FileSystem> file_system = table->GetFileSystem(); + ScopeGuard temp_file_guard([&file, &file_system]() { + // The stream is closed before the file goes: a stream dropped without `Close()` may still + // flush, and on an object store that write would land after the delete. + if (file.out != nullptr) { + Status closed = file.out->Close(); + if (!closed.ok()) { + PAIMON_LOG_WARN(WriteLogger(), + "Failed to close %s after its writer could not be opened: %s", + file.temp_file_path.c_str(), closed.ToString().c_str()); + } + file.out.reset(); + } + Status status = file_system->Delete(file.temp_file_path, /*recursive=*/false); + if (!status.ok()) { + PAIMON_LOG_WARN(WriteLogger(), + "Failed to remove the temp file %s after its writer " + "could not be opened: %s", + file.temp_file_path.c_str(), status.ToString().c_str()); + } + }); + PAIMON_ASSIGN_OR_RAISE(file.writer, writer_builder->Build(file.out, file_compression)); + temp_file_guard.Release(); + return file; +} + +Status FormatTableWrite::Impl::FinishFile(const std::string& directory) { + auto file_iter = open_files.find(directory); + if (file_iter == open_files.end()) { + return Status::OK(); + } + // The file stays in `open_files` until it is recorded for committing: dropped earlier, a + // failure below would leave a temp path no `Abort()` knows about. + FormatTableWriteFile& file = file_iter->second; + // The partition is written and erased together with the file, so this cannot miss; should + // that ever break, it must not quietly commit a file under no partition. + auto partition_iter = open_partitions.find(directory); + if (partition_iter == open_partitions.end()) { + return Status::Invalid( + fmt::format("no partition was recorded for the file open in {}", directory)); + } + + PAIMON_RETURN_NOT_OK(file.writer->Flush()); + PAIMON_RETURN_NOT_OK(file.writer->Finish()); + file.writer.reset(); Review Comment: Good catch, and you are right about the consequence: `GetPos()`, the stream's `Flush()` and `Close()` all run after the writer has been finished and reset, so a failure there left the entry in `open_files` with a null writer, and the next `Write()` or `PrepareCommit()` went straight through it. Fixed by making the whole close either finish the file or leave nothing behind: - Everything that can fail moved into `Impl::CloseFileAndStage()`, so `FinishFile()` has one place to clean up after. On failure it now calls `DiscardOpenFile()` (finish the writer if it is still there, close the stream before deleting so a late flush cannot resurrect the file, remove the temp file), erases the entry from `open_files` and `open_partitions`, and records the failure in `Impl::finish_failure`. - `Write()` and `PrepareCommit()` return that failure before touching anything, so the write is terminal: the rows of the file that could not be closed cannot be published, and publishing the other files would quietly lose them. - `Abort()` now reuses `DiscardOpenFile()` instead of repeating the same cleanup. Note the fix covers the writer's own `Flush()`/`Finish()` too, not just the three stream calls, since they are handled uniformly now. Test: `TestAFileThatCannotBeClosedEndsTheWrite` adds a `FailingWriteFileSystem` whose streams refuse one chosen call, and runs the three injections (`GetPos`, `Flush`, `Close`) twice over — once with `target-file-row-num=1` so the failure lands inside `Write()`, and once so it lands inside `PrepareCommit()`. It asserts that the failing call reports the error, that a later `Write()` and a later `PrepareCommit()` both answer with a failure rather than dereferencing the writer, that `Abort()` still succeeds, and that a scan afterwards sees nothing left behind. ########## src/paimon/core/table/format/format_table_commit.cpp: ########## @@ -0,0 +1,327 @@ +/* + * 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 "paimon/table/format/format_table_commit.h" + +#include <algorithm> +#include <map> +#include <set> +#include <string> +#include <utility> +#include <vector> + +#include "fmt/format.h" +#include "fmt/ranges.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/table/format/format_file_listing.h" +#include "paimon/core/table/format/format_file_naming.h" +#include "paimon/core/table/format/format_path_validation.h" +#include "paimon/core/utils/partition_path_utils.h" +#include "paimon/fs/file_system.h" +#include "paimon/logging.h" + +namespace paimon { + +namespace { + +Logger* CommitLogger() { + static std::unique_ptr<Logger> logger = Logger::GetLogger("FormatTableCommit"); + return logger.get(); +} + +/// Fails when a commit message does not describe a file of this table. +/// +/// A message is a public struct the caller may have built itself, while committing it renames one +/// path and an overwrite clears the directory around it. Only the shape of a message is checked, +/// not who produced it. +Status ValidateCommitMessage(const FormatCommitMessage& message, + const std::shared_ptr<FormatTable>& table, + const std::map<std::string, std::string>& static_partition) { + const std::string what = fmt::format("commit message {}", message.ToString()); + PAIMON_RETURN_NOT_OK(FormatPathValidation::ValidatePathUnderLocation(message.file_path, + table->Location(), what)); + PAIMON_RETURN_NOT_OK(FormatPathValidation::ValidatePathUnderLocation(message.temp_file_path, + table->Location(), what)); + + // `<publish directory>/_temporary/.tmp.<uuid>`, which a scan skips and which is what Java + // Paimon's `RenamingTwoPhaseOutputStream` stages under. + const std::string publish_directory = PathUtil::GetParentDirPath(message.file_path); + const std::string directory_prefix = publish_directory + "/"; + if (!StringUtils::StartsWith(message.temp_file_path, directory_prefix) || + !FormatFileNaming::IsTempFilePath(message.temp_file_path.substr(directory_prefix.size()))) { + return Status::Invalid(fmt::format( + "{} does not stage its file under '{}/{}...' beside where it will be published, so it " + "may already be visible or belong to another directory", + what, FormatFileNaming::kTempDirName, FormatFileNaming::kTempFilePrefix)); + } + // Otherwise an overwrite could clear the old data and publish a file nothing can ever read. + PAIMON_RETURN_NOT_OK( + FormatPathValidation::ValidateFileIsVisible(table, message.file_path, what)); + if (message.record_count < 0 || message.file_size < 0) { + return Status::Invalid(fmt::format("{} reports a negative row count or file size", what)); + } + + // Otherwise an overwrite would clear the wrong partition. + PAIMON_RETURN_NOT_OK( + FormatPathValidation::ValidatePartitionKeys(table, message.partition, what)); + PAIMON_RETURN_NOT_OK(FormatPathValidation::ValidateFileInPartition(table, message.file_path, + message.partition, what)); + + // Otherwise the file would be published into a partition this commit never cleared. + for (const auto& [key, value] : static_partition) { + auto iter = message.partition.find(key); + if (iter == message.partition.end() || iter->second != value) { + return Status::Invalid(fmt::format( + "{} is not in the static partition '{}={}' this commit writes", what, key, value)); + } + } + return Status::OK(); +} + +/// Fails when `static_partition` cannot name a directory of this table. The keys must be a prefix +/// of the partition keys, since a partition directory nests below the one before it. +Status ValidateStaticPartition(const std::map<std::string, std::string>& static_partition, + const std::vector<std::string>& partition_keys, + const std::string& table_name) { + if (static_partition.empty()) { + return Status::OK(); + } + if (partition_keys.empty()) { + return Status::Invalid(fmt::format( + "format table {} is not partitioned, so a static partition names nothing", table_name)); + } + for (const auto& entry : static_partition) { + const std::string& key = entry.first; + if (std::find(partition_keys.begin(), partition_keys.end(), key) == partition_keys.end()) { + return Status::Invalid( + fmt::format("'{}' is not a partition key of format table {}", key, table_name)); + } + } + bool missing_leading_key = false; + for (const std::string& partition_key : partition_keys) { + const bool named = static_partition.find(partition_key) != static_partition.end(); + if (named && missing_leading_key) { + return Status::Invalid( + fmt::format("static partition column '{}' of format table {} cannot be given " + "without the partition columns it nests under", + partition_key, table_name)); + } + if (!named) { + missing_leading_key = true; + } + } + return Status::OK(); +} + +} // namespace + +std::string FormatCommitMessage::ToString() const { + return fmt::format( + "FormatCommitMessage{{file_path={}, temp_file_path={}, partition={}, record_count={}, " + "file_size={}}}", + file_path, temp_file_path, partition, record_count, file_size); +} + +FormatTableCommit::FormatTableCommit(const std::shared_ptr<FormatTable>& table, bool overwrite, + const std::map<std::string, std::string>& static_partition) + : table_(table), overwrite_(overwrite), static_partition_(static_partition) {} + +FormatTableCommit::~FormatTableCommit() = default; + +Result<std::unique_ptr<FormatTableCommit>> FormatTableCommit::Create( + const std::shared_ptr<FormatTable>& table, bool overwrite, + const std::map<std::string, std::string>& static_partition) { + if (table == nullptr) { + return Status::Invalid("format table commit requires a table"); + } + PAIMON_RETURN_NOT_OK( + ValidateStaticPartition(static_partition, table->PartitionKeys(), table->FullName())); + return std::unique_ptr<FormatTableCommit>( + new FormatTableCommit(table, overwrite, static_partition)); +} + +Status FormatTableCommit::DeletePreviousDataFiles(const std::string& directory, + int32_t partition_levels) const { + std::shared_ptr<FileSystem> file_system = table_->GetFileSystem(); + FormatDataFileListingOptions listing; + listing.partition_levels = partition_levels; + listing.only_value_in_path = table_->PartitionOnlyValueInPath(); + listing.default_part_name = table_->PartitionDefaultName(); + // Only right at the location are `schema` and `branch` metadata; below it they are data. + PAIMON_ASSIGN_OR_RAISE(bool at_location, + FormatPathValidation::IsTableLocation(table_, directory)); + listing.skip_reserved_directories = at_location && table_->LocationCarriesPaimonMetadata(); + std::vector<FormatDataSplit::FileMeta> files; + // Committed data files only: a staging directory holds another writer's uncommitted output. + PAIMON_RETURN_NOT_OK(FormatFileListing::ListDataFiles(file_system, directory, listing, &files)); + for (const FormatDataSplit::FileMeta& file : files) { + Status status = file_system->Delete(file.file_path, /*recursive=*/false); + if (!status.ok() && !status.IsNotExist()) { + return status; + } + } + return Status::OK(); +} + +Status FormatTableCommit::Commit(const std::vector<FormatCommitMessage>& commit_messages) { + Status status = CommitImpl(commit_messages); + if (!status.ok()) { + // The write has already prepared its commit, so nothing else will clean up what is still + // staged. `Abort()` logs its own failures and returns OK today; the status is still read. + Status abort_status = Abort(commit_messages); + if (!abort_status.ok()) { + PAIMON_LOG_WARN(CommitLogger(), "Failed to clean up table %s after a failed commit: %s", + table_->FullName().c_str(), abort_status.ToString().c_str()); + } + } + return status; +} + +Status FormatTableCommit::CommitImpl(const std::vector<FormatCommitMessage>& commit_messages) { + std::shared_ptr<FileSystem> file_system = table_->GetFileSystem(); + const std::vector<std::string>& partition_keys = table_->PartitionKeys(); + // An overwrite deletes committed data, so what it was asked to replace is worth logging even + // when it succeeds, as the managed table commit does at the same level. + PAIMON_LOG_INFO(CommitLogger(), "Ready to %s %zu messages to format table %s", + overwrite_ ? "overwrite with" : "commit", commit_messages.size(), + table_->FullName().c_str()); + + // Before anything is renamed or deleted: an overwrite clears the directory a message names. + for (const FormatCommitMessage& message : commit_messages) { + PAIMON_RETURN_NOT_OK(ValidateCommitMessage(message, table_, static_partition_)); + } + + // Every message is checked against what is on disk before anything moves. An overwrite makes + // this critical: it clears the old data first, so a staged file that turns out to be missing + // would leave the table with the old rows gone and the new ones never arriving. + std::set<std::string> targets; + for (const FormatCommitMessage& message : commit_messages) { + if (!targets.insert(message.file_path).second) { + return Status::Invalid(fmt::format( + "two commit messages would publish {}, so one would overwrite the other", + message.file_path)); + } + Result<FileStatus> staged = file_system->GetFileStatus(message.temp_file_path); + if (!staged.ok()) { + // `Invalid` whatever the file system said, since the fault is the message; its text + // is kept all the same. + return Status::Invalid(fmt::format("the staged file {} cannot be read: {}", + message.temp_file_path, staged.status().ToString())); + } + // `rename` moves a directory as readily as a file. + if (staged.value().IsDir()) { + return Status::Invalid(fmt::format("the staged path {} is a directory, not a file", + message.temp_file_path)); + } + if (message.file_size != staged.value().GetLen()) { + return Status::Invalid( + fmt::format("the staged file {} is {} bytes but the commit message says {}", + message.temp_file_path, staged.value().GetLen(), message.file_size)); + } + } + + // What an overwrite replaces is cleared first: it removes committed files only, and this + // commit's own are still hidden, so neither can take the other out. + if (!static_partition_.empty()) { + // The spec names the leading keys in order, so the path may be a prefix with the + // partitions of the unnamed keys below it. + std::vector<std::pair<std::string, std::string>> ordered_partition; + ordered_partition.reserve(static_partition_.size()); + for (const std::string& partition_key : partition_keys) { + auto iter = static_partition_.find(partition_key); + if (iter == static_partition_.end()) { + break; + } + ordered_partition.emplace_back(partition_key, iter->second); + } + PAIMON_ASSIGN_OR_RAISE(std::string partition_path, + PartitionPathUtils::GeneratePartitionPath( + ordered_partition, table_->PartitionOnlyValueInPath())); + std::string directory = PathUtil::JoinPath(table_->Location(), partition_path); + // The spec may name a partition no message covers: an overwrite of a directory a scan + // skips would clear files that are not this table's data. + PAIMON_RETURN_NOT_OK(FormatPathValidation::ValidateDirectoryIsVisible(table_, directory, + "static partition")); + PAIMON_ASSIGN_OR_RAISE(bool exists, file_system->Exists(directory)); + if (!exists) { + // Nothing to clear, but created regardless: an overwrite leaves an empty partition + // behind rather than removing it from the table. + PAIMON_RETURN_NOT_OK(file_system->Mkdirs(directory)); + } else if (overwrite_) { + PAIMON_RETURN_NOT_OK(DeletePreviousDataFiles( + directory, static_cast<int32_t>(partition_keys.size() - ordered_partition.size()))); + } + } else if (overwrite_) { + std::set<std::string> directories; + for (const FormatCommitMessage& message : commit_messages) { + directories.insert(PathUtil::GetParentDirPath(message.file_path)); Review Comment: You are right, and this is a real hole: `ValidateFileInPartition()` accepts a file below the partition directory, so the overwrite was clearing `dt=2024/part-0` while old files at `dt=2024` and in `dt=2024/part-1` survived, leaving old and new rows together. Fixed by deriving the directory to clear from the partition the message declares rather than from the file's parent: ```cpp PAIMON_ASSIGN_OR_RAISE( std::string directory, FormatPathValidation::BuildPartitionDirectory(table_, message.partition)); ``` `DeletePreviousDataFiles()` then clears the whole partition, subdirectories included, since the listing descends. The partition and the path were already checked against each other earlier in the same commit, so either would name the same partition; only this one names all of it. It also matches the static-partition branch just above, which builds its directory from the spec the same way. Java derives the partition from the committer's target path instead, which is equivalent there because its writer never nests a file below the partition root. Test: `TestOverwriteReplacesTheWholePartitionOfANestedFile` commits an overwrite whose message names `dt=20240101/part-0/data-nested-0.parquet`, with old data both at the partition root and in a sibling `dt=20240101/part-1/`, and asserts the scan afterwards returns only the new row. -- 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]
