SteNicholas commented on code in PR #222: URL: https://github.com/apache/paimon-cpp/pull/222#discussion_r3954192345
########## src/paimon/core/table/format/format_table_write.cpp: ########## @@ -0,0 +1,656 @@ +/* + * 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/core/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. A failure leaves no + /// open file behind and stops the write, since the file is then neither writable nor + /// publishable. + Status FinishFile(const std::string& directory); + + /// Closes `file` and appends the message that publishes it. Everything that can fail lives + /// here, so `FinishFile()` has one place to clean up after. + Status CloseFileAndStage(FormatTableWriteFile* file, + const std::map<std::string, std::string>& partition); + + /// Gives up on a file that was never staged: closes what is still open and removes the temp + /// file. Best effort, like `Abort()`, since the failure that got here is the one to report. + void DiscardOpenFile(FormatTableWriteFile* file); + + /// 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; + /// The failure that closing a file stopped at, or OK. It ends the write: the rows of that + /// file cannot be published, so publishing the others would quietly lose them. + Status finish_failure; + + /// 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)); + } + + Status status = CloseFileAndStage(&file, partition_iter->second); + if (!status.ok()) { + // Closing gets this far only once the writer is finished and gone, so there is nothing + // left to write into and nothing whole to publish. What is on disk is discarded and the + // write stops here: leaving the entry in `open_files` would have the next `Write()` or + // `PrepareCommit()` reach through a writer that is no longer there. + DiscardOpenFile(&file); + open_files.erase(file_iter); + open_partitions.erase(partition_iter); + finish_failure = status; + return status; + } + open_files.erase(file_iter); + open_partitions.erase(partition_iter); + return Status::OK(); +} + +Status FormatTableWrite::Impl::CloseFileAndStage( + FormatTableWriteFile* file, const std::map<std::string, std::string>& partition) { + PAIMON_RETURN_NOT_OK(file->writer->Flush()); + PAIMON_RETURN_NOT_OK(file->writer->Finish()); + file->writer.reset(); + // The size is read before closing, while the stream still knows how far it wrote. + PAIMON_ASSIGN_OR_RAISE(int64_t file_size, file->out->GetPos()); + PAIMON_RETURN_NOT_OK(file->out->Flush()); + PAIMON_RETURN_NOT_OK(file->out->Close()); + file->out.reset(); + + PAIMON_LOG_DEBUG(WriteLogger(), "Staged %s for %s, %ld rows, %ld bytes", + file->temp_file_path.c_str(), file->file_path.c_str(), file->record_count, + file_size); + staged_messages.emplace_back(file->temp_file_path, file->file_path, partition, + file->record_count, file_size); + return Status::OK(); +} + +void FormatTableWrite::Impl::DiscardOpenFile(FormatTableWriteFile* file) { + std::shared_ptr<FileSystem> file_system = table->GetFileSystem(); + if (file->writer != nullptr) { + Status status = file->writer->Finish(); + if (!status.ok()) { + PAIMON_LOG_WARN(WriteLogger(), "Failed to finish the writer of %s while discarding: %s", + file->temp_file_path.c_str(), status.ToString().c_str()); + } + file->writer.reset(); + } + // 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 status = file->out->Close(); + if (!status.ok()) { + PAIMON_LOG_WARN(WriteLogger(), "Failed to close %s while discarding: %s", + file->temp_file_path.c_str(), status.ToString().c_str()); + } + file->out.reset(); + } + Status status = file_system->Delete(file->temp_file_path, /*recursive=*/false); + if (!status.ok() && !status.IsNotExist()) { + PAIMON_LOG_WARN(WriteLogger(), "Failed to remove the temp file %s while discarding: %s", + file->temp_file_path.c_str(), status.ToString().c_str()); + } +} + +Status FormatTableWrite::Write(std::unique_ptr<RecordBatch>&& batch) { + if (const char* finished = impl_->FinishedReason(); finished != nullptr) { + return Status::Invalid(finished); + } + // A file that could not be closed ends the write: see `Impl::finish_failure`. + PAIMON_RETURN_NOT_OK(impl_->finish_failure); + if (batch == nullptr || batch->GetData() == nullptr) { + return Status::Invalid("format table write requires a batch"); + } + for (RecordBatch::RowKind row_kind : batch->GetRowKind()) { + if (row_kind != RecordBatch::RowKind::INSERT) { + return Status::Invalid( + "format table only supports INSERT rows: a directory of data files records no " + "row identity for an update or a delete to apply to"); + } + } + + // A partial partition would not name a single directory. + const std::map<std::string, std::string>& partition = batch->GetPartition(); + const std::vector<std::string>& partition_keys = impl_->table->PartitionKeys(); + if (partition.size() != partition_keys.size()) { + return Status::Invalid(fmt::format( + "batch carries {} partition values but table {} is partitioned by {} fields", + partition.size(), impl_->table->FullName(), partition_keys.size())); + } + for (const std::string& partition_key : partition_keys) { + if (partition.find(partition_key) == partition.end()) { + return Status::Invalid(fmt::format( + "batch does not carry a value for partition field '{}'", partition_key)); + } + } + + // `year=2025/month=01/`, or `2025/01/` under `format-table.partition-path-only-value`. The + // scan reads back whichever is written here, so both go through one place. + PAIMON_ASSIGN_OR_RAISE(FormatTablePartitionTarget target, impl_->GetPartitionTarget(partition)); + const std::string& directory = target.directory; + // In the partition the table rendered rather than the one the caller spelled, since a value + // can be written more than one way. + std::vector<std::pair<std::string, std::string>> ordered_partition; + ordered_partition.reserve(partition_keys.size()); + for (const std::string& partition_key : partition_keys) { + auto iter = target.partition.find(partition_key); + if (iter == target.partition.end()) { + return Status::Invalid( + fmt::format("partition field '{}' is missing from the partition the table rendered", + partition_key)); + } + ordered_partition.emplace_back(partition_key, iter->second); + } + + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::Array> array, + arrow::ImportArray(batch->GetData(), impl_->table_struct_type)); + PAIMON_RETURN_NOT_OK(ArrowUtils::CheckNullabilityMatch(impl_->table_schema, array)); + auto table_struct = checked_pointer_cast<arrow::StructArray>(array); + if (table_struct->null_count() != 0) { + return Status::Invalid( + "format table write does not support a null row: a row of the table must have a value " + "for every column, even if that value is null"); + } + + PAIMON_RETURN_NOT_OK(impl_->ValidatePartitionColumns(table_struct, ordered_partition)); + + // Partition columns are not written: the directory holds those values. + arrow::ArrayVector data_columns; + data_columns.reserve(impl_->data_column_indexes.size()); + for (int32_t index : impl_->data_column_indexes) { + data_columns.push_back(table_struct->field(index)); + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::StructArray> data_struct, + arrow::StructArray::Make(data_columns, impl_->data_schema->fields())); + + // A write that never receives a row leaves the directory as it found it. + if (data_struct->length() == 0) { + return Status::OK(); + } + + auto file_iter = impl_->open_files.find(directory); + if (file_iter == impl_->open_files.end()) { + PAIMON_ASSIGN_OR_RAISE(FormatTableWriteFile file, impl_->OpenFile(directory)); + file_iter = impl_->open_files.emplace(directory, std::move(file)).first; + impl_->open_partitions[directory] = target.partition; + } + + ArrowArray c_data_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*data_struct, &c_data_array)); + PAIMON_RETURN_NOT_OK(file_iter->second.writer->AddBatch(&c_data_array)); + file_iter->second.record_count += data_struct->length(); + Review Comment: Done. `FormatTableWrite::Impl` keeps a latched `write_failure`, set whenever a failure reaches the file's writer — `AddBatch()`, `ReachTargetSize()`, or the `FinishFile()` they trigger — and both `Write()` and `PrepareCommit()` refuse afterwards, as `RollingFileWriter` and the Java implementation do. `Abort()` still removes what is on disk. Tests cover a batch that cannot be added, a target-size check that fails, and a footer that cannot be written, each for both formats. ########## src/paimon/core/table/format/format_table_test.cpp: ########## @@ -0,0 +1,3176 @@ +/* + * 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.h" + +#include <algorithm> +#include <limits> +#include <map> +#include <memory> +#include <optional> +#include <string> +#include <tuple> +#include <vector> + +#include "arrow/api.h" +#include "arrow/c/bridge.h" +#include "gtest/gtest.h" +#include "paimon/cache/cache.h" +#include "paimon/commit_context.h" +#include "paimon/commit_message.h" +#include "paimon/common/table/special_fields.h" +#include "paimon/common/types/row_kind.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/schema/schema_manager.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/table/format/format_commit_message.h" +#include "paimon/core/table/format/format_data_split.h" +#include "paimon/core/table/format/format_table_commit.h" +#include "paimon/core/table/format/format_table_read.h" +#include "paimon/core/table/format/format_table_scan.h" +#include "paimon/core/table/format/format_table_write.h" +#include "paimon/defs.h" +#include "paimon/file_store_commit.h" +#include "paimon/file_store_write.h" +#include "paimon/fs/file_system.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/predicate/literal.h" +#include "paimon/predicate/predicate_builder.h" +#include "paimon/read_context.h" +#include "paimon/record_batch.h" +#include "paimon/scan_context.h" +#include "paimon/status.h" +#include "paimon/table/source/split.h" +#include "paimon/table/source/table_read.h" +#include "paimon/table/source/table_scan.h" +#include "paimon/testing/utils/testharness.h" +#include "paimon/write_context.h" + +namespace paimon::test { + +namespace { + +/// What `ListPartitions()` returns. Aliased because a macro argument cannot hold the comma in +/// `std::map<std::string, std::string>`: the preprocessor would read it as two arguments. +using PartitionList = std::vector<std::map<std::string, std::string>>; + +std::shared_ptr<arrow::Schema> MakeSchema() { + return arrow::schema({arrow::field("id", arrow::int32()), arrow::field("name", arrow::utf8()), + arrow::field("dt", arrow::utf8())}); +} + +/// Wraps a file system and writes down what a write did to each path, in order. Only the order +/// gives away a temp file deleted while its stream is still open, which on a store that flushes +/// from the destructor lands the write after the delete. +class CallOrderFileSystem : public FileSystem { + public: + explicit CallOrderFileSystem(const std::shared_ptr<FileSystem>& delegate) + : delegate_(delegate), calls_(std::make_shared<std::vector<std::string>>()) {} + + /// What happened, as "<verb> <path>" in the order it happened. + const std::vector<std::string>& Calls() const { + return *calls_; + } + + using FileSystem::Open; + + Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override { + return delegate_->Open(path); + } + Result<std::unique_ptr<OutputStream>> Create(const std::string& path, + bool overwrite) const override { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<OutputStream> out, + delegate_->Create(path, overwrite)); + calls_->push_back("create " + path); + return std::unique_ptr<OutputStream>( + new RecordingOutputStream(std::move(out), path, calls_)); + } + Status Mkdirs(const std::string& path) const override { + return delegate_->Mkdirs(path); + } + Status Rename(const std::string& src, const std::string& dst) const override { + return delegate_->Rename(src, dst); + } + Status Delete(const std::string& path, bool recursive = true) const override { + calls_->push_back("delete " + path); + return delegate_->Delete(path, recursive); + } + Result<FileStatus> GetFileStatus(const std::string& path) const override { + return delegate_->GetFileStatus(path); + } + Status ListDir(const std::string& directory, + std::vector<BasicFileStatus>* status_list) const override { + return delegate_->ListDir(directory, status_list); + } + Status ListFileStatus(const std::string& path, + std::vector<FileStatus>* status_list) const override { + return delegate_->ListFileStatus(path, status_list); + } + Result<bool> Exists(const std::string& path) const override { + return delegate_->Exists(path); + } + + private: + /// Records its own close, so a stream still open when its file was deleted can be told apart + /// from one that was closed first. + class RecordingOutputStream : public OutputStream { + public: + RecordingOutputStream(std::unique_ptr<OutputStream> delegate, const std::string& path, + const std::shared_ptr<std::vector<std::string>>& calls) + : delegate_(std::move(delegate)), path_(path), calls_(calls) {} + + Result<int64_t> Write(const char* buffer, int64_t size) override { + return delegate_->Write(buffer, size); + } + Status Flush() override { + return delegate_->Flush(); + } + Result<int64_t> GetPos() const override { + return delegate_->GetPos(); + } + Result<std::string> GetUri() const override { + return delegate_->GetUri(); + } + Status Close() override { + calls_->push_back("close " + path_); + return delegate_->Close(); + } + + private: + std::unique_ptr<OutputStream> delegate_; + std::string path_; + std::shared_ptr<std::vector<std::string>> calls_; + }; + + std::shared_ptr<FileSystem> delegate_; + /// Shared with every stream this hands out, so one list holds the whole sequence. + std::shared_ptr<std::vector<std::string>> calls_; +}; + +/// The one call a `FailingWriteFileSystem`'s streams refuse. +enum class FailingStreamCall { kGetPos, kFlush, kClose }; + +/// Wraps a file system and hands out streams that fail one call, so that a write can be stopped +/// where a real store would stop it: after the writer is finished and while the file it produced +/// is still hidden. +class FailingWriteFileSystem : public FileSystem { + public: + FailingWriteFileSystem(const std::shared_ptr<FileSystem>& delegate, FailingStreamCall failing) + : delegate_(delegate), failing_(failing) {} + + using FileSystem::Open; + + Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override { + return delegate_->Open(path); + } + Result<std::unique_ptr<OutputStream>> Create(const std::string& path, + bool overwrite) const override { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<OutputStream> out, + delegate_->Create(path, overwrite)); + return std::unique_ptr<OutputStream>(new FailingOutputStream(std::move(out), failing_)); + } + Status Mkdirs(const std::string& path) const override { + return delegate_->Mkdirs(path); + } + Status Rename(const std::string& src, const std::string& dst) const override { + return delegate_->Rename(src, dst); + } + Status Delete(const std::string& path, bool recursive = true) const override { + return delegate_->Delete(path, recursive); + } + Result<FileStatus> GetFileStatus(const std::string& path) const override { + return delegate_->GetFileStatus(path); + } + Status ListDir(const std::string& directory, + std::vector<BasicFileStatus>* status_list) const override { + return delegate_->ListDir(directory, status_list); + } + Status ListFileStatus(const std::string& path, + std::vector<FileStatus>* status_list) const override { + return delegate_->ListFileStatus(path, status_list); + } + Result<bool> Exists(const std::string& path) const override { + return delegate_->Exists(path); + } + + private: + class FailingOutputStream : public OutputStream { + public: + FailingOutputStream(std::unique_ptr<OutputStream> delegate, FailingStreamCall failing) + : delegate_(std::move(delegate)), failing_(failing) {} + + Result<int64_t> Write(const char* buffer, int64_t size) override { + return delegate_->Write(buffer, size); + } + Status Flush() override { + if (failing_ == FailingStreamCall::kFlush) { + return Status::IOError("injected flush failure"); + } + return delegate_->Flush(); + } + Result<int64_t> GetPos() const override { + if (failing_ == FailingStreamCall::kGetPos) { + return Status::IOError("injected get position failure"); + } + return delegate_->GetPos(); + } + Result<std::string> GetUri() const override { + return delegate_->GetUri(); + } + Status Close() override { + if (failing_ == FailingStreamCall::kClose) { + // Closed all the same, so the file it wrote can still be removed. + [[maybe_unused]] Status closed = delegate_->Close(); + return Status::IOError("injected close failure"); + } + return delegate_->Close(); + } + + private: + std::unique_ptr<OutputStream> delegate_; + FailingStreamCall failing_; + }; + + std::shared_ptr<FileSystem> delegate_; + FailingStreamCall failing_; +}; + +/// Creates a format table's schema on disk and loads the table. +Result<std::shared_ptr<FormatTable>> CreateTable( + const std::shared_ptr<FileSystem>& file_system, const std::string& path, + const std::vector<std::string>& partition_keys, + const std::map<std::string, std::string>& extra_options = {}) { + std::map<std::string, std::string> options = {{Options::TYPE, "format-table"}, + {Options::FILE_FORMAT, "parquet"}}; + for (const auto& [key, value] : extra_options) { + options[key] = value; + } + SchemaManager schema_manager(file_system, path); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<TableSchema> table_schema, + schema_manager.CreateTable(MakeSchema(), partition_keys, /*primary_keys=*/{}, options)); + return FormatTable::Create(file_system, path, Identifier("db", "tbl")); +} + +/// Builds one batch of the table's columns, inserts unless other row kinds are given. +Result<std::unique_ptr<RecordBatch>> MakeBatch( + const std::vector<int32_t>& ids, const std::vector<std::string>& names, const std::string& dt, + const std::map<std::string, std::string>& partition, + const std::vector<RecordBatch::RowKind>& row_kinds = {}) { + arrow::Int32Builder id_builder; + arrow::StringBuilder name_builder; + arrow::StringBuilder dt_builder; + for (size_t i = 0; i < ids.size(); i++) { + PAIMON_RETURN_NOT_OK_FROM_ARROW(id_builder.Append(ids[i])); + PAIMON_RETURN_NOT_OK_FROM_ARROW(name_builder.Append(names[i])); + PAIMON_RETURN_NOT_OK_FROM_ARROW(dt_builder.Append(dt)); + } + std::shared_ptr<arrow::Array> id_array; + std::shared_ptr<arrow::Array> name_array; + std::shared_ptr<arrow::Array> dt_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(id_builder.Finish(&id_array)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(name_builder.Finish(&name_array)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(dt_builder.Finish(&dt_array)); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::StructArray> struct_array, + arrow::StructArray::Make({id_array, name_array, dt_array}, MakeSchema()->fields())); + + auto c_array = std::make_unique<ArrowArray>(); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*struct_array, c_array.get())); + RecordBatchBuilder builder(c_array.get()); + builder.SetPartition(partition); + if (!row_kinds.empty()) { + builder.SetRowKinds(row_kinds); + } + return builder.Finish(); +} + +/// Builds a one-row batch whose partition column is null, declared as `default_partition_name`. +Result<std::unique_ptr<RecordBatch>> MakeBatchWithNullPartition( + const std::string& default_partition_name) { + arrow::Int32Builder id_builder; + arrow::StringBuilder name_builder; + arrow::StringBuilder dt_builder; + PAIMON_RETURN_NOT_OK_FROM_ARROW(id_builder.Append(1)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(name_builder.Append("alice")); + PAIMON_RETURN_NOT_OK_FROM_ARROW(dt_builder.AppendNull()); + std::shared_ptr<arrow::Array> id_array; + std::shared_ptr<arrow::Array> name_array; + std::shared_ptr<arrow::Array> dt_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(id_builder.Finish(&id_array)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(name_builder.Finish(&name_array)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(dt_builder.Finish(&dt_array)); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::StructArray> struct_array, + arrow::StructArray::Make({id_array, name_array, dt_array}, MakeSchema()->fields())); + + auto c_array = std::make_unique<ArrowArray>(); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*struct_array, c_array.get())); + RecordBatchBuilder builder(c_array.get()); + builder.SetPartition({{"dt", default_partition_name}}); + return builder.Finish(); +} + +/// Builds one batch of `count` rows, large enough that a small split target has to cut it up. +Result<std::unique_ptr<RecordBatch>> MakeManyRowBatch(int32_t count) { + std::vector<int32_t> ids; + std::vector<std::string> names; + ids.reserve(count); + names.reserve(count); + for (int32_t i = 0; i < count; i++) { + ids.push_back(i); + names.push_back("name-" + std::to_string(i)); + } + return MakeBatch(ids, names, "20240101", {}); +} + +/// Builds one batch of `count` wide, all-different rows starting at `start_id`. A file is +/// measured by the bytes its writer has finished with, and small repeated values sit in a +/// dictionary it has not written yet. +Result<std::unique_ptr<RecordBatch>> MakeWideRowBatch(int32_t count, int32_t start_id) { + constexpr size_t kNameLength = 1024; + std::vector<int32_t> ids; + std::vector<std::string> names; + ids.reserve(count); + names.reserve(count); + for (int32_t i = 0; i < count; i++) { + const int32_t id = start_id + i; + ids.push_back(id); + names.push_back(std::to_string(id) + + std::string(kNameLength, static_cast<char>('a' + (id % 26)))); + } + return MakeBatch(ids, names, "20240101", {}); +} + +/// The path a write stages `file_path` under: a `_temporary` directory beside where the file will +/// be published, holding a hidden name of its own. Java Paimon's `RenamingTwoPhaseOutputStream` +/// uses the same layout. +std::string StagedPath(const std::string& file_path) { + return PathUtil::JoinPath(PathUtil::GetParentDirPath(file_path), + "_temporary/.tmp.d9b7f0a2-0c11-4a35-9f6e-2f2f0f9e6c41"); +} + +/// Writes one batch and commits it, so the files become part of the table. +Status WriteAndCommit(const std::shared_ptr<FormatTable>& table, + std::unique_ptr<RecordBatch>&& batch, bool overwrite = false, + const std::map<std::string, std::string>& static_partition = {}) { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FormatTableWrite> write, + FormatTableWrite::Create(table, /*pool=*/nullptr)); + PAIMON_RETURN_NOT_OK(write->Write(std::move(batch))); + PAIMON_ASSIGN_OR_RAISE(std::vector<FormatCommitMessage> messages, write->PrepareCommit()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FormatTableCommit> commit, + FormatTableCommit::Create(table, overwrite, static_partition)); + return commit->Commit(messages); +} + +/// Imports a batch the reader handed out. `ASSERT_OK_AND_ASSIGN` only understands paimon's +/// `Result`, so arrow's has to be converted before a test body can use it. +Result<std::shared_ptr<arrow::RecordBatch>> ImportBatch(const BatchReader::ReadBatch& batch) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::RecordBatch> record_batch, + arrow::ImportRecordBatch(batch.first.get(), batch.second.get())); + return record_batch; +} + +/// Reads every row of a plan, returning the rows as `id|name|dt` strings. +Result<std::vector<std::string>> ReadAll(const std::shared_ptr<FormatTable>& table, + const std::vector<std::shared_ptr<Split>>& splits, + const std::shared_ptr<Predicate>& predicate = nullptr, + bool enable_predicate_filter = false) { + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<FormatTableRead> read, + FormatTableRead::Create(table, /*projection=*/std::nullopt, /*pool=*/nullptr, predicate, + enable_predicate_filter)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BatchReader> reader, read->CreateReader(splits)); + std::vector<std::string> rows; + while (true) { + PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch batch, reader->NextBatch()); + if (BatchReader::IsEofBatch(batch)) { + break; + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::RecordBatch> record_batch, + arrow::ImportRecordBatch(batch.first.get(), batch.second.get())); + // The leading field is `_VALUE_KIND`, which every `BatchReader` puts first; a format + // table has no row kinds of its own, so every row of it is an insert. + if (record_batch->schema()->field(0)->name() != SpecialFields::ValueKind().Name()) { + return Status::Invalid("a format table read must still carry the _VALUE_KIND field"); + } + auto row_kinds = checked_pointer_cast<arrow::Int8Array>(record_batch->column(0)); + for (int64_t i = 0; i < record_batch->num_rows(); i++) { + if (row_kinds->Value(i) != RowKind::Insert()->ToByteValue()) { + return Status::Invalid("a format table read must return inserts only"); + } + } + auto ids = checked_pointer_cast<arrow::Int32Array>(record_batch->column(1)); + auto names = checked_pointer_cast<arrow::StringArray>(record_batch->column(2)); + auto dts = checked_pointer_cast<arrow::StringArray>(record_batch->column(3)); + for (int64_t i = 0; i < record_batch->num_rows(); i++) { + rows.push_back(std::to_string(ids->Value(i)) + "|" + names->GetString(i) + "|" + + dts->GetString(i)); + } + } + reader->Close(); + return rows; +} + +} // namespace + +TEST(FormatTableTest, TestParseFormat) { + ASSERT_OK_AND_ASSIGN(FormatTable::Format parquet, FormatTable::ParseFormat("PARQUET")); + ASSERT_EQ(parquet, FormatTable::Format::PARQUET); + ASSERT_OK_AND_ASSIGN(FormatTable::Format orc, FormatTable::ParseFormat("orc")); + ASSERT_EQ(orc, FormatTable::Format::ORC); + ASSERT_EQ(FormatTable::FormatToString(FormatTable::Format::ORC), "orc"); + + // Format table formats with no reader here yet answer `NotImplemented`, which is a different + // answer from a name that is no format at all. + for (const char* format : {"csv", "text", "json", "mosaic"}) { + Result<FormatTable::Format> unimplemented = FormatTable::ParseFormat(format); + ASSERT_FALSE(unimplemented.ok()) << format; + ASSERT_TRUE(unimplemented.status().IsNotImplemented()) << format; + } + + Result<FormatTable::Format> unknown = FormatTable::ParseFormat("nonesuch"); + ASSERT_FALSE(unknown.ok()); + ASSERT_TRUE(unknown.status().IsInvalid()); +} + +TEST(FormatTableTest, TestCreateReadsOptions) { + std::unique_ptr<UniqueTestDirectory> dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + ASSERT_OK_AND_ASSIGN(std::shared_ptr<FormatTable> table, + CreateTable(dir->GetFileSystem(), dir->Str(), {"dt"})); + ASSERT_EQ(table->Location(), dir->Str()); + ASSERT_EQ(table->GetFormat(), FormatTable::Format::PARQUET); + ASSERT_EQ(table->PartitionKeys(), std::vector<std::string>({"dt"})); + ASSERT_EQ(table->FileCompression(), "snappy"); + ASSERT_EQ(table->PartitionDefaultName(), "__DEFAULT_PARTITION__"); + ASSERT_EQ(table->FullName(), "db.tbl"); +} + +TEST(FormatTableTest, TestFileCompressionComesFromCoreOptions) { + // The resolution order itself is `CoreOptionsTest.TestFormatTableFileCompression`'s business; + // what matters here is that the table asks for it rather than resolving compression again. + std::unique_ptr<UniqueTestDirectory> dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr<FormatTable> table, + CreateTable(dir->GetFileSystem(), dir->Str(), {}, + {{Options::FORMAT_TABLE_FILE_COMPRESSION, "lz4"}, {"compression", "zstd"}})); + ASSERT_EQ(table->FileCompression(), "lz4"); + + std::unique_ptr<UniqueTestDirectory> default_dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(default_dir); + ASSERT_OK_AND_ASSIGN(std::shared_ptr<FormatTable> default_table, + CreateTable(default_dir->GetFileSystem(), default_dir->Str(), {})); + ASSERT_EQ(default_table->FileCompression(), "snappy"); +} + +TEST(FormatTableTest, TestFileFormatDefaultsToParquet) { + std::unique_ptr<UniqueTestDirectory> dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + SchemaManager schema_manager(dir->GetFileSystem(), dir->Str()); + ASSERT_OK_AND_ASSIGN( + [[maybe_unused]] std::unique_ptr<TableSchema> table_schema, + schema_manager.CreateTable(MakeSchema(), /*partition_keys=*/{}, + /*primary_keys=*/{}, {{Options::TYPE, "format-table"}})); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr<FormatTable> table, + FormatTable::Create(dir->GetFileSystem(), dir->Str(), Identifier("db", "tbl"))); + ASSERT_EQ(table->GetFormat(), FormatTable::Format::PARQUET); + ASSERT_EQ(table->FileCompression(), "snappy"); +} + +TEST(FormatTableTest, TestUnknownTableTypeIsRejected) { + std::unique_ptr<UniqueTestDirectory> dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + SchemaManager schema_manager(dir->GetFileSystem(), dir->Str()); + // Refused at creation: read as a managed table it would look for snapshots it never had. + Result<std::unique_ptr<TableSchema>> unknown_type = schema_manager.CreateTable( + MakeSchema(), /*partition_keys=*/{}, /*primary_keys=*/{}, {{Options::TYPE, "nonesuch"}}); + ASSERT_FALSE(unknown_type.ok()); + ASSERT_TRUE(unknown_type.status().IsInvalid()); +} + +TEST(FormatTableTest, TestATableTypeThisLibraryCannotOpenIsRejectedAtCreation) { + std::unique_ptr<UniqueTestDirectory> dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + SchemaManager schema_manager(dir->GetFileSystem(), dir->Str()); + // A table type paimon names but this library cannot open: refused up front, with its own + // status code and the type quoted. + Result<std::unique_ptr<TableSchema>> object_table = + schema_manager.CreateTable(MakeSchema(), /*partition_keys=*/{}, /*primary_keys=*/{}, + {{Options::TYPE, "object-table"}}); + ASSERT_FALSE(object_table.ok()); + ASSERT_TRUE(object_table.status().IsNotImplemented()) << object_table.status().ToString(); + ASSERT_NE(std::string::npos, object_table.status().ToString().find("object-table")); +} + +TEST(FormatTableTest, TestAPartitionColumnOfAnUnsupportedTypeIsRefusedUpFront) { + // A partition value makes the round trip through its column type on the way to a directory + // name and back. `BINARY` cannot, so a table partitioned by one is refused where the table is + // decided rather than at the first read or write of a table that already looked created. + std::shared_ptr<arrow::Schema> binary_schema = + arrow::schema({arrow::field("id", arrow::int32()), arrow::field("name", arrow::utf8()), + arrow::field("bin", arrow::binary())}); + std::unique_ptr<UniqueTestDirectory> dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + SchemaManager schema_manager(dir->GetFileSystem(), dir->Str()); + Result<std::unique_ptr<TableSchema>> created = schema_manager.CreateTable( + binary_schema, /*partition_keys=*/{"bin"}, /*primary_keys=*/{}, + {{Options::TYPE, "format-table"}, {Options::FILE_FORMAT, "parquet"}}); + ASSERT_FALSE(created.ok()); + ASSERT_NE(std::string::npos, created.status().ToString().find("cannot be partitioned")) + << created.status().ToString(); + Review Comment: Done. ########## src/paimon/core/table/format/format_file_naming.h: ########## @@ -0,0 +1,73 @@ +/* + * 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 <cstdint> +#include <string> + +#include "paimon/result.h" + +namespace paimon { + +/// Names the data files one format table write produces. +/// +/// `{prefix}{uuid}-{n}.{extension}`, the convention every paimon writer follows: the uuid belongs +/// to this write and the counter to its files, so two concurrent writers cannot collide. +/// +/// A file is staged under `_temporary/.tmp.{uuid}` beside where it will end up and takes its real +/// name only on commit, as Java Paimon's `RenamingTwoPhaseOutputStream` does. Both the directory +/// and the name are hidden, which is the Hive-style convention for output that is not committed +/// table data and is what a scan of this table skips. +class FormatFileNaming { + public: + static constexpr char kDefaultDataFilePrefix[] = "data-"; + /// Directory a staged file waits in, shared with every other writer of the same table. + static constexpr char kTempDirName[] = "_temporary"; + static constexpr char kTempFilePrefix[] = ".tmp."; + + /// @param extension File extension without its dot, which is the format's identifier. + /// @param prefix File name prefix, from `data-file.prefix`. It may not be hidden by the + /// `_` / `.` convention, since a scan skips every such file. + static Result<FormatFileNaming> Create(const std::string& extension, const std::string& prefix); + + FormatFileNaming() = default; Review Comment: Done. `Create()` returns `Result<std::unique_ptr<FormatFileNaming>>`, the constructor is private and the public default constructor is gone, so an instance cannot exist without passing all four checks. `FormatTableWrite::Impl` holds it as a `std::unique_ptr` instead of a value member. ########## src/paimon/core/table/format/format_file_naming.h: ########## @@ -0,0 +1,73 @@ +/* + * 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 <cstdint> +#include <string> + +#include "paimon/result.h" + +namespace paimon { + +/// Names the data files one format table write produces. +/// +/// `{prefix}{uuid}-{n}.{extension}`, the convention every paimon writer follows: the uuid belongs +/// to this write and the counter to its files, so two concurrent writers cannot collide. +/// +/// A file is staged under `_temporary/.tmp.{uuid}` beside where it will end up and takes its real +/// name only on commit, as Java Paimon's `RenamingTwoPhaseOutputStream` does. Both the directory +/// and the name are hidden, which is the Hive-style convention for output that is not committed +/// table data and is what a scan of this table skips. Review Comment: Noted, and left as a TODO on `FormatFileNaming` naming `MultiPartUploadTwoPhaseOutputStream` and the reason: on an object store a rename is a server-side copy followed by a delete, so a commit pays for the data twice and a partly finished publish cannot be undone atomically — which is exactly why the rollback in `FormatTableCommit` is best-effort. Out of scope for this PR. -- 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]
