lxy-9602 commented on code in PR #222: URL: https://github.com/apache/paimon-cpp/pull/222#discussion_r3844137180
########## src/paimon/core/table/format/format_path_validation.h: ########## @@ -0,0 +1,89 @@ +/* + * 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 <map> +#include <memory> +#include <string> + +#include "paimon/result.h" +#include "paimon/status.h" +#include "paimon/table/format/format_table.h" + +namespace paimon { + +/// Checks that a path a caller handed over really names something of this format table. +/// +/// A split and a commit message both come back through an interface that takes the base type, so +/// what arrives may belong to another plan, another table or a plan made before the files moved, +/// and nothing further down re-checks the paths they name. +class FormatPathValidation { + public: + FormatPathValidation() = delete; + ~FormatPathValidation() = delete; + + /// Fails when `path` is not a file inside `location`. By path component, not by string + /// prefix: `<location>/../victim` starts with the location and still resolves outside it. + /// Only the path text is checked, so a symbolic link pointing out of the table is not caught. + static Status ValidatePathUnderLocation(const std::string& path, const std::string& location, + const std::string& what); Review Comment: Could we normalize the location and candidate path before performing the containment check? The current string comparison may reject equivalent local paths, for example a table location `/tmp/table` and a file path `file:///tmp/table/data.parquet`. `Path::ToString()` alone would not fully address this, since it normalizes `file:///...` to `file:/...` but does not make it equivalent to a scheme-less local path. Perhaps we could parse both paths with `PathUtil::ToPath()`, compare the scheme and authority separately—treating an empty scheme and `file` as equivalent for local paths—and perform the component check against the normalized `Path::path`. A regression test covering mixed local-path and `file:` URI forms would also be helpful. ########## src/paimon/core/table/format/format_table_commit.h: ########## @@ -0,0 +1,104 @@ +/* + * 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 <map> +#include <memory> +#include <string> +#include <vector> + +#include "paimon/core/table/format/format_commit_message.h" +#include "paimon/result.h" +#include "paimon/status.h" +#include "paimon/table/format/format_table.h" + +namespace paimon { + +/// Publishes the files a `FormatTableWrite` produced, by renaming each out of the `_temporary` +/// directory it was staged in. +/// +/// A directory has no metadata to switch, so a commit is not atomic across files: it renames them +/// one at a time, and a reader scanning midway sees the ones renamed so far. Each rename does +/// guarantee that a file becomes visible whole, never half-written. +/// +/// A commit that fails partway tries to remove the files it had already renamed, on a best-effort +/// basis: a file that cannot be removed is reported in the log and stays. An overwriting commit is +/// further limited - the data it replaces is deleted before the new files are published and cannot +/// be brought back, so a failure there leaves the table without the replaced data. +/// +/// Only the messages this job's own writers produced may be passed in. The checks here can tell +/// that a message's path belongs to this table, sits in the partition it declares, and is staged +/// rather than already published - not whose staged file it is. +/// +/// Not thread-safe. Separate commits may add to one table at once, each publishing only the files +/// its own messages name; two overwriting commits over the same directory race, since an overwrite +/// clears everything committed there before publishing anything. +class FormatTableCommit { + public: + /// @param table Table to commit to. + /// @param overwrite Whether the commit replaces the data already in the directories it writes + /// to, instead of adding to it. Without a static partition that means every partition + /// the commit touches; with one it means the partitions that spec covers, whether or + /// not this commit wrote to them. + /// @param static_partition Partition the commit writes to, keyed by partition field name. It + /// may name only the leading partition keys, in which case it stands for every + /// partition below that prefix. Empty means the partitions are whatever the written + /// files say they are. + static Result<std::unique_ptr<FormatTableCommit>> Create( + const std::shared_ptr<FormatTable>& table, bool overwrite, + const std::map<std::string, std::string>& static_partition); + + ~FormatTableCommit(); + + /// Renames every written file into place, first clearing what it replaces when the commit + /// overwrites. + Status Commit(const std::vector<FormatCommitMessage>& commit_messages); Review Comment: Could we either support or explicitly reject `dynamic-partition-overwrite=false` for format tables? Currently, overwrite derives the directories to clear only from commit messages, so it always behaves like dynamic partition overwrite and clears nothing when the message list is empty. Java replaces the whole table when this option is false, and always does so for unpartitioned tables. If this behavior is out of scope, failing fast may be safer than silently leaving stale data. ########## 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: Could we abort the writer or mark the write as terminal when `AddBatch` or `ReachTargetSize` fails? The writer may already be partially modified, so allowing subsequent `Write` or `PrepareCommit` calls could produce an incomplete file. This would also align the failure handling with the existing `RollingFileWriter` and Java implementation. ########## 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: ASSERT_NOK_WITH_MSG ########## src/paimon/core/table/format/format_table_file_store_commit.h: ########## @@ -0,0 +1,109 @@ +/* + * 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 <map> +#include <memory> +#include <optional> +#include <string> +#include <vector> + +#include "paimon/core/table/format/format_commit_message.h" +#include "paimon/defs.h" +#include "paimon/file_store_commit.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +class FormatTable; +class Metrics; + +/// Commits a format table through the `FileStoreCommit` interface, so that a caller holding a +/// table path commits it the way it commits any other table. Java Paimon does the same through +/// `FormatTable.newBatchWriteBuilder()`. +/// +/// Most of `FileStoreCommit` is about snapshots and manifests, which a format table has none of: +/// expiring them, rolling back to one, and filtering by a commit identifier recorded in one all +/// refer to state this table does not keep. Each is refused rather than quietly doing nothing, so +/// a caller moving between table types finds out at the call rather than from a table that did +/// not change. `RowIdCheckConflict()` is the one exception, since it returns a reference and has +/// no way to report a refusal. +/// +/// What is left is what a directory of files can do: `Commit()`, `Overwrite()` and `Abort()`. +class FormatTableFileStoreCommit : public FileStoreCommit { + public: + static Result<std::unique_ptr<FormatTableFileStoreCommit>> Create( + const std::shared_ptr<FormatTable>& table); + + ~FormatTableFileStoreCommit() override; + + Status Commit(const std::vector<std::shared_ptr<CommitMessage>>& commit_messages, + int64_t commit_identifier = BATCH_WRITE_COMMIT_IDENTIFIER, + std::optional<int64_t> watermark = std::nullopt) override; Review Comment: Please avoid using default parameters in production code. ########## src/paimon/core/table/format/format_table_read.cpp: ########## @@ -0,0 +1,413 @@ +/* + * 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_read.h" + +#include <algorithm> +#include <map> +#include <set> +#include <string> +#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/predicate/predicate_validator.h" +#include "paimon/common/reader/complete_row_kind_batch_reader.h" +#include "paimon/common/reader/concat_batch_reader.h" +#include "paimon/common/reader/data_file_reader_factory.h" +#include "paimon/common/reader/predicate_batch_reader.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/binary_row_partition_computer.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/core_options.h" +#include "paimon/core/io/field_mapping_reader.h" +#include "paimon/core/table/format/format_data_split.h" +#include "paimon/core/table/format/format_path_validation.h" +#include "paimon/core/table/format/lazy_concat_batch_reader.h" +#include "paimon/core/utils/field_mapping.h" +#include "paimon/format/file_format.h" +#include "paimon/format/file_format_factory.h" +#include "paimon/fs/file_system.h" +#include "paimon/predicate/predicate.h" + +namespace paimon { + +/// Readers, outermost first: CompleteRowKindBatchReader -> (PredicateBatchReader) +/// -> LazyConcatBatchReader across the split's files -> FieldMappingReader +/// -> (DelegatingPrefetchReader) -> (PrefetchFileBatchReader) -> FormatReader +/// +/// The same shape the managed table path builds, minus what a format table has none of: no +/// deletion vectors, no bitmap index, no row-tracking fields and no shredding. The last three +/// readers are built by `DataFileReaderFactory`, which is where the two paths meet. +class FormatTableRead::Impl { + public: + std::shared_ptr<FormatTable> table; + /// Columns the reader returns, in the order it returns them. + std::shared_ptr<arrow::Schema> read_schema; + /// The whole table schema: the mapping below splits the partition columns out of it. + std::shared_ptr<arrow::Schema> data_schema; + /// Splits the read schema into file columns and partition columns and rewrites the predicate + /// against the file's own fields. The same builder the managed table path uses. + std::shared_ptr<FieldMappingBuilder> field_mapping_builder; + /// Turns a split's partition values into the `BinaryRow` a `FieldMappingReader` fills its + /// partition columns from. Null when the table is not partitioned. + std::shared_ptr<BinaryRowPartitionComputer> partition_computer; + /// The predicate the returned reader applies exactly, or null when the caller filters itself. + std::shared_ptr<Predicate> filter_predicate; + std::shared_ptr<MemoryPool> pool; + /// Runs the reads a prefetching reader issues ahead of the batches being asked for. Null when + /// nothing asked for prefetch. + std::shared_ptr<Executor> executor; + std::string format_identifier; + /// What a file is opened with. The same struct the managed table path fills in, read by the + /// same component. + DataFileReadOptions read_options; +}; + +FormatTableRead::FormatTableRead(std::unique_ptr<Impl> impl, + const std::shared_ptr<MemoryPool>& pool) + : TableRead(pool), impl_(std::move(impl)) {} + +FormatTableRead::~FormatTableRead() = default; + +Result<std::unique_ptr<FormatTableRead>> FormatTableRead::Create( + const std::shared_ptr<FormatTable>& table, + const std::optional<std::vector<std::string>>& projection, + const std::shared_ptr<MemoryPool>& pool, const std::shared_ptr<Predicate>& predicate, + bool enable_predicate_filter) { + return CreateInternal(table, projection, pool, predicate, enable_predicate_filter, + /*read_context=*/nullptr); +} Review Comment: Is this function intended as a test-only interface, or is it also used in production code? If it is only for tests, could we move it to `private`, rename it to something like `TEST_Create`, or have the tests call `FormatTableRead::Create` directly from `read_context` instead? The main concern is that the parameters of `CreateInternal` look a bit unusual right now. For example, `enable_predicate_filter` and `predicate` already exist in `read_context`, but currently some of this information is passed separately while some is taken from `read_context`, which makes the interface feel inconsistent. ########## src/paimon/core/table/format/format_table_read.cpp: ########## @@ -0,0 +1,413 @@ +/* + * 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_read.h" + +#include <algorithm> +#include <map> +#include <set> +#include <string> +#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/predicate/predicate_validator.h" +#include "paimon/common/reader/complete_row_kind_batch_reader.h" +#include "paimon/common/reader/concat_batch_reader.h" +#include "paimon/common/reader/data_file_reader_factory.h" +#include "paimon/common/reader/predicate_batch_reader.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/binary_row_partition_computer.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/core_options.h" +#include "paimon/core/io/field_mapping_reader.h" +#include "paimon/core/table/format/format_data_split.h" +#include "paimon/core/table/format/format_path_validation.h" +#include "paimon/core/table/format/lazy_concat_batch_reader.h" +#include "paimon/core/utils/field_mapping.h" +#include "paimon/format/file_format.h" +#include "paimon/format/file_format_factory.h" +#include "paimon/fs/file_system.h" +#include "paimon/predicate/predicate.h" + +namespace paimon { + +/// Readers, outermost first: CompleteRowKindBatchReader -> (PredicateBatchReader) +/// -> LazyConcatBatchReader across the split's files -> FieldMappingReader +/// -> (DelegatingPrefetchReader) -> (PrefetchFileBatchReader) -> FormatReader +/// +/// The same shape the managed table path builds, minus what a format table has none of: no +/// deletion vectors, no bitmap index, no row-tracking fields and no shredding. The last three +/// readers are built by `DataFileReaderFactory`, which is where the two paths meet. +class FormatTableRead::Impl { + public: + std::shared_ptr<FormatTable> table; + /// Columns the reader returns, in the order it returns them. + std::shared_ptr<arrow::Schema> read_schema; + /// The whole table schema: the mapping below splits the partition columns out of it. + std::shared_ptr<arrow::Schema> data_schema; + /// Splits the read schema into file columns and partition columns and rewrites the predicate + /// against the file's own fields. The same builder the managed table path uses. + std::shared_ptr<FieldMappingBuilder> field_mapping_builder; + /// Turns a split's partition values into the `BinaryRow` a `FieldMappingReader` fills its + /// partition columns from. Null when the table is not partitioned. + std::shared_ptr<BinaryRowPartitionComputer> partition_computer; + /// The predicate the returned reader applies exactly, or null when the caller filters itself. + std::shared_ptr<Predicate> filter_predicate; + std::shared_ptr<MemoryPool> pool; + /// Runs the reads a prefetching reader issues ahead of the batches being asked for. Null when + /// nothing asked for prefetch. + std::shared_ptr<Executor> executor; + std::string format_identifier; + /// What a file is opened with. The same struct the managed table path fills in, read by the + /// same component. + DataFileReadOptions read_options; +}; + +FormatTableRead::FormatTableRead(std::unique_ptr<Impl> impl, + const std::shared_ptr<MemoryPool>& pool) + : TableRead(pool), impl_(std::move(impl)) {} + +FormatTableRead::~FormatTableRead() = default; + +Result<std::unique_ptr<FormatTableRead>> FormatTableRead::Create( + const std::shared_ptr<FormatTable>& table, + const std::optional<std::vector<std::string>>& projection, + const std::shared_ptr<MemoryPool>& pool, const std::shared_ptr<Predicate>& predicate, + bool enable_predicate_filter) { + return CreateInternal(table, projection, pool, predicate, enable_predicate_filter, + /*read_context=*/nullptr); +} + +Result<std::unique_ptr<FormatTableRead>> FormatTableRead::Create( + const std::shared_ptr<FormatTable>& table, const std::shared_ptr<ReadContext>& read_context) { + if (table == nullptr) { + return Status::Invalid("format table read requires a table"); + } + if (read_context == nullptr) { + return Status::Invalid("format table read requires a read context"); + } + if (read_context->GetRealtimeContext() != nullptr) { + return Status::NotImplemented( + "a format table has no real-time store to union with what is on disk"); + } + // A projected read schema can rename a column, prune a nested one and give it metadata of its + // own, while a format table's projection is a list of top-level names, so it is refused rather + // than read as if it had never been given. + if (read_context->GetReadSchema() != nullptr) { + return Status::NotImplemented( + "a format table read does not take a projected read schema; name the columns to read " + "instead"); + } + + std::optional<std::vector<std::string>> projection; + if (!read_context->GetReadFieldNames().empty()) { + projection = read_context->GetReadFieldNames(); + } else if (!read_context->GetReadFieldIds().empty()) { + // Resolved against the table's own schema, which is the only thing that knows the ids: a + // file another engine wrote carries none. + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<::ArrowSchema> c_schema, table->GetArrowSchema()); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> table_schema, + arrow::ImportSchema(c_schema.get())); + PAIMON_ASSIGN_OR_RAISE(std::vector<DataField> fields, + DataField::ConvertArrowSchemaToDataFields(table_schema)); + std::map<int32_t, std::string> name_by_id; + for (const DataField& field : fields) { + name_by_id.emplace(field.Id(), field.Name()); + } + std::vector<std::string> names; + names.reserve(read_context->GetReadFieldIds().size()); + for (int32_t field_id : read_context->GetReadFieldIds()) { + auto iter = name_by_id.find(field_id); + if (iter == name_by_id.end()) { + return Status::Invalid(fmt::format("field id {} is not a column of table {}", + field_id, table->FullName())); + } + names.push_back(iter->second); + } + projection = std::move(names); + } + + return CreateInternal(table, projection, read_context->GetMemoryPool(), + read_context->GetPredicate(), read_context->EnablePredicateFilter(), + read_context); +} + +Result<std::unique_ptr<FormatTableRead>> FormatTableRead::CreateInternal( + const std::shared_ptr<FormatTable>& table, + const std::optional<std::vector<std::string>>& projection, + const std::shared_ptr<MemoryPool>& pool, const std::shared_ptr<Predicate>& predicate, + bool enable_predicate_filter, const std::shared_ptr<ReadContext>& read_context) { + if (table == nullptr) { + return Status::Invalid("format table read requires a table"); + } + std::shared_ptr<MemoryPool> memory_pool = pool != nullptr ? pool : GetDefaultPool(); + + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<::ArrowSchema> c_schema, table->GetArrowSchema()); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> table_schema, + arrow::ImportSchema(c_schema.get())); + + const std::vector<std::string>& partition_keys = table->PartitionKeys(); + auto is_partition_key = [&partition_keys](const std::string& name) { + return std::find(partition_keys.begin(), partition_keys.end(), name) != + partition_keys.end(); + }; + + arrow::FieldVector read_fields; + if (projection) { + read_fields.reserve(projection->size()); + std::set<std::string> projected; + for (const std::string& name : *projection) { + // A read column is looked up by name, so twice has no meaning to act on. + if (!projected.insert(name).second) { + return Status::Invalid(fmt::format( + "column '{}' appears more than once in the projection, which paimon-cpp does " + "not allow", + name)); + } + std::shared_ptr<arrow::Field> field = table_schema->GetFieldByName(name); + if (field == nullptr) { + return Status::Invalid( + fmt::format("field '{}' is not a column of table {}", name, table->FullName())); + } + read_fields.push_back(std::move(field)); + } + } else { + read_fields = table_schema->fields(); + } + if (read_fields.empty()) { + return Status::Invalid("format table read requires at least one column to read"); + } + + auto impl = std::make_unique<Impl>(); + impl->table = table; + impl->read_schema = arrow::schema(read_fields); + impl->pool = memory_pool; + impl->format_identifier = FormatTable::FormatToString(table->GetFormat()); + // The whole schema, as the managed table path hands it over: the mapping splits the + // partition columns out itself and asks the file only for what is left. + impl->data_schema = table_schema; + + const bool has_non_partition_column = + std::any_of(table_schema->fields().begin(), table_schema->fields().end(), + [&is_partition_key](const std::shared_ptr<arrow::Field>& field) { + return !is_partition_key(field->name()); + }); + if (!has_non_partition_column) { + return Status::Invalid( + fmt::format("format table {} has no non-partition column, so its files hold nothing to " + "read", + table->FullName())); + } + + if (predicate != nullptr) { + // The same rules `InternalReadContext` applies to a managed table's predicate. The field + // index is not among them: everything downstream resolves a field by name. + PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithSchema( + *impl->read_schema, predicate, /*validate_field_idx=*/false)); + PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithLiterals(predicate)); + if (enable_predicate_filter) { + impl->filter_predicate = predicate; + } + } + + // The builder also hands the file reader only the conjuncts naming columns the file holds. + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<FieldMappingBuilder> field_mapping_builder, + FieldMappingBuilder::Create(impl->read_schema, partition_keys, predicate)); + impl->field_mapping_builder = std::move(field_mapping_builder); + + PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, + CoreOptions::FromMap(table->Options(), table->GetFileSystem())); + impl->read_options.read_batch_size = core_options.GetReadBatchSize(); + impl->read_options.adaptive_prefetch_strategy = core_options.EnableAdaptivePrefetchStrategy(); + if (read_context != nullptr) { + // Straight from the context, as the managed table path takes them. Without a context + // nobody asked for any of this, so a file is opened plainly. + impl->read_options.cache = read_context->GetCache(); + impl->read_options.prefetch_enabled = read_context->EnablePrefetch(); + impl->read_options.prefetch_max_parallel_num = read_context->GetPrefetchMaxParallelNum(); + impl->read_options.prefetch_batch_count = read_context->GetPrefetchBatchCount(); + impl->read_options.read_ahead_cache_enabled = read_context->ReadAheadCacheEnabled(); + impl->read_options.cache_config = read_context->GetCacheConfig(); + impl->executor = read_context->GetExecutor(); + } + if (!partition_keys.empty()) { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BinaryRowPartitionComputer> partition_computer, + BinaryRowPartitionComputer::Create( + partition_keys, table_schema, table->PartitionDefaultName(), + core_options.LegacyPartitionNameEnabled(), memory_pool)); + impl->partition_computer = std::move(partition_computer); + } + + return std::unique_ptr<FormatTableRead>(new FormatTableRead(std::move(impl), memory_pool)); +} + +Result<std::unique_ptr<BatchReader>> FormatTableRead::CreateSplitReader( + const std::shared_ptr<Split>& split) { + auto format_split = std::dynamic_pointer_cast<FormatDataSplit>(split); + if (format_split == nullptr) { + return Status::Invalid("format table read only accepts a FormatDataSplit"); + } + + // `CreateReader()` takes a `Split` the caller held on to, which may have been planned from + // another table or before these files moved, so whether a file belongs to this table is asked + // here rather than taken on trust. + PAIMON_RETURN_NOT_OK(FormatPathValidation::ValidatePartitionKeys( + impl_->table, format_split->partition, "split")); + for (const FormatDataSplit::FileMeta& file : format_split->files) { + PAIMON_RETURN_NOT_OK(FormatPathValidation::ValidatePathUnderLocation( + file.file_path, impl_->table->Location(), "split")); + // A split mixing partitions would read rows back under values they never had. + PAIMON_RETURN_NOT_OK(FormatPathValidation::ValidateFileInPartition( + impl_->table, file.file_path, format_split->partition, "split")); + PAIMON_RETURN_NOT_OK( + FormatPathValidation::ValidateFileIsVisible(impl_->table, file.file_path, "split")); + if (file.file_size < 0) { + return Status::Invalid(fmt::format("split gives {} a negative size", file.file_path)); + } + } + + // The partition values in the shape a `FieldMappingReader` reads them from; a directory named + // after the default partition name reads back as null. + BinaryRow partition = BinaryRow::EmptyRow(); + if (impl_->partition_computer != nullptr) { + PAIMON_ASSIGN_OR_RAISE(partition, + impl_->partition_computer->ToBinaryRow(format_split->partition)); + } + + // One reader builder serves the whole split, built by the same component the managed table + // path uses, so a format table's file is read with the cache and the read hints any other + // data file is. + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReaderBuilder> builder, + DataFileReaderFactory::CreateReaderBuilder( + impl_->format_identifier, impl_->table->Options(), + /*extra_format_options=*/{}, impl_->read_options, impl_->pool)); + std::shared_ptr<ReaderBuilder> reader_builder(std::move(builder)); + + // Captured by value, so a file's reader outlives this `FormatTableRead`. + std::shared_ptr<FormatTable> table = impl_->table; + std::shared_ptr<arrow::Schema> data_schema = impl_->data_schema; + std::shared_ptr<FieldMappingBuilder> field_mapping_builder = impl_->field_mapping_builder; + std::shared_ptr<MemoryPool> pool = impl_->pool; + std::shared_ptr<Executor> executor = impl_->executor; + std::string format_identifier = impl_->format_identifier; + DataFileReadOptions read_options = impl_->read_options; + + // Each file is named alongside its factory, so every failure says which file it was. + std::vector<LazyConcatBatchReader::Source> sources; + sources.reserve(format_split->files.size()); + for (const FormatDataSplit::FileMeta& file : format_split->files) { + LazyConcatBatchReader::Source source; + source.name = file.file_path; + source.open = [table, reader_builder, data_schema, field_mapping_builder, partition, pool, + executor, format_identifier, read_options, + file]() -> Result<std::unique_ptr<BatchReader>> { + // The same mapping for every file; built per file only because `FieldMappingReader` + // takes ownership of it. + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FieldMapping> field_mapping, + field_mapping_builder->CreateFieldMapping(data_schema)); + std::shared_ptr<arrow::Schema> file_read_schema = + DataField::ConvertDataFieldsToArrowSchema( + field_mapping->non_partition_info.non_partition_data_schema); + std::shared_ptr<Predicate> pushdown_predicate = + field_mapping->non_partition_info.non_partition_filter; + + // The split's size is whatever the caller gave it and `Open` trusts what it is handed: + // a stale length would truncate an object-store read or send it past the end, so the + // file system is asked for the real one. + PAIMON_ASSIGN_OR_RAISE(FileStatus status, + table->GetFileSystem()->GetFileStatus(file.file_path)); + if (status.IsDir()) { + return Status::Invalid("the split names a directory, not a data file"); + } + if (file.file_size != status.GetLen()) { + return Status::Invalid(fmt::format( + "the split says it is {} bytes but it is {}; the plan was made against a " + "different version of the file", + file.file_size, status.GetLen())); + } + // Opened through the same component the managed table path opens a data file with, Review Comment: There may be a potential performance hotspot here. In the read path, both StarRocks and DuckDB previously observed that, in small-file scenarios, calling OSS open after first fetching the file length could become a noticeable hotspot. They optimized this by opening the file directly using the file size from metadata, which helped reduce the open overhead (pr #189 ). It may be worth leaving a TODO here so we can optimize this later if it turns out to be a hotspot in practice. ########## src/paimon/core/schema/schema_validation.cpp: ########## @@ -134,27 +143,146 @@ bool SchemaValidation::IsComplexType(const std::shared_ptr<arrow::Field>& field) BlobUtils::IsBlobField(field)); } -Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) { - const auto& field_names = schema.FieldNames(); - PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.BucketKeys(), "bucket key")); - PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PrimaryKeys(), "primary key")); - PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PartitionKeys(), "partition key")); - PAIMON_RETURN_NOT_OK( - Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PartitionKeys()), - "Table column {} should include all partition fields {}", - field_names, schema.PartitionKeys())); - PAIMON_RETURN_NOT_OK( - Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PrimaryKeys()), - "Table column {} should include all primary key constraint {}", - field_names, schema.PrimaryKeys())); - - PAIMON_RETURN_NOT_OK( - ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PrimaryKeys(), "primary key")); - PAIMON_RETURN_NOT_OK( - ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PartitionKeys(), "partition")); +Status SchemaValidation::ValidateGenericSchema(const std::vector<DataField>& fields, + const std::vector<std::string>& bucket_keys, + const std::vector<std::string>& primary_keys, + const std::vector<std::string>& partition_keys) { + std::vector<std::string> field_names; + field_names.reserve(fields.size()); + for (const DataField& field : fields) { + field_names.push_back(field.Name()); + } + PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(bucket_keys, "bucket key")); + PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(primary_keys, "primary key")); + PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(partition_keys, "partition key")); + PAIMON_RETURN_NOT_OK(Preconditions::CheckState( + ObjectUtils::ContainsAll(field_names, partition_keys), + "Table column {} should include all partition fields {}", field_names, partition_keys)); + PAIMON_RETURN_NOT_OK(Preconditions::CheckState( + ObjectUtils::ContainsAll(field_names, primary_keys), + "Table column {} should include all primary key constraint {}", field_names, primary_keys)); + for (const auto& field_name : field_names) { + if (SpecialFields::IsSystemField(field_name)) { + return Status::Invalid( + fmt::format("field name '{}' in schema cannot be special field.", field_name)); + } + } + PAIMON_RETURN_NOT_OK(ValidateOnlyContainPrimitiveType(fields, primary_keys, "primary key")); + PAIMON_RETURN_NOT_OK(ValidateOnlyContainPrimitiveType(fields, partition_keys, "partition")); // TODO(lisizhuo.lsz): C++ Paimon do not support timestamp & decimal & float & double type in // partition keys for now. - PAIMON_RETURN_NOT_OK(ValidateNotContainSpecificType(schema.Fields(), schema.PartitionKeys())); + PAIMON_RETURN_NOT_OK(ValidateNotContainSpecificType(fields, partition_keys)); + return Status::OK(); +} + +Status SchemaValidation::ValidateNewTableSchema(const TableSchema& schema) { + const std::map<std::string, std::string>& options = schema.Options(); + PAIMON_ASSIGN_OR_RAISE(TableType table_type, TableTypeDefine::FromOptions(options)); + if (table_type != TableType::TABLE && table_type != TableType::MATERIALIZED_TABLE && + table_type != TableType::FORMAT_TABLE) { + // Quoted back rather than re-rendered from `table_type`, so the message says what was + // actually asked for. + auto type_iter = options.find(Options::TYPE); + return Status::NotImplemented(fmt::format( + "Cannot create a table whose '{}' is '{}': paimon-cpp does not implement this table " + "type.", + Options::TYPE, type_iter == options.end() ? std::string() : type_iter->second)); + } + if (table_type == TableType::FORMAT_TABLE) { + PAIMON_RETURN_NOT_OK(ValidateGenericTableSchema(schema)); + // At creation the schema's own options are the only ones there are, and `file-system` is + // resolved from them like every other option. + return ValidateFormatTableSchema(schema, schema.Options(), /*file_system=*/nullptr); Review Comment: Should `ValidateNewTableSchema` accept the caller’s already resolved `FileSystem`, so that we can pass it here instead of `nullptr`? `SchemaManager` already has `file_system_`. This is harmless today because validation performs no filesystem I/O, but passing the actual instance would better match the API contract and avoid resolving a different filesystem in the future. ########## 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, Review Comment: This check may not be necessary. The original write path in paimon-cpp does not perform this validation either, and it could introduce some performance overhead. Also, even if this value were written incorrectly, it should not have any practical impact, since the paritition field in the file is not read. ########## src/paimon/core/table/format/format_table_scan.cpp: ########## @@ -0,0 +1,236 @@ +/* + * 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_scan.h" + +#include <algorithm> +#include <string> +#include <utility> + +#include "fmt/format.h" +#include "fmt/ranges.h" +#include "paimon/common/utils/bin_packing.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/core/core_options.h" +#include "paimon/core/table/format/format_data_split.h" +#include "paimon/core/table/format/format_file_listing.h" +#include "paimon/core/table/format/format_path_validation.h" +#include "paimon/core/table/source/plan_impl.h" +#include "paimon/core/utils/partition_path_utils.h" +#include "paimon/fs/file_system.h" +#include "paimon/logging.h" + +namespace paimon { + +namespace { +Logger* ScanLogger() { + static std::unique_ptr<Logger> logger = Logger::GetLogger("FormatTableScan"); + return logger.get(); +} +} // namespace + +FormatTableScan::FormatTableScan(const std::shared_ptr<FormatTable>& table, + const std::map<std::string, std::string>& partition_filter, + const std::optional<int32_t>& limit, int64_t target_split_size, + int64_t open_file_cost) + : table_(table), + partition_filter_(partition_filter), + limit_(limit), + target_split_size_(target_split_size), + open_file_cost_(open_file_cost) {} + +FormatTableScan::~FormatTableScan() = default; + +Result<std::unique_ptr<FormatTableScan>> FormatTableScan::Create( + const std::shared_ptr<FormatTable>& table, + const std::map<std::string, std::string>& partition_filter, + const std::optional<int32_t>& limit) { + if (table == nullptr) { + return Status::Invalid("format table scan requires a table"); + } + const std::vector<std::string>& partition_keys = table->PartitionKeys(); + for (const auto& filter : partition_filter) { + const std::string& key = filter.first; + if (std::find(partition_keys.begin(), partition_keys.end(), key) == partition_keys.end()) { + return Status::Invalid( + fmt::format("partition filter field '{}' is not a partition key of table {}", key, + table->FullName())); + } + } + // Through `CoreOptions`, so `"128 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())); + return std::unique_ptr<FormatTableScan>( + new FormatTableScan(table, partition_filter, limit, core_options.GetSourceSplitTargetSize(), + core_options.GetSourceSplitOpenFileCost())); +} + +Result<std::vector<FormatTableScan::PartitionAndPath>> FormatTableScan::FindPartitions() const { + const std::vector<std::string>& partition_keys = table_->PartitionKeys(); + // Partitions are discovered by listing one directory level at a time, so the filter applied + // and the number of partitions that survived it are the only way to explain an empty plan. + PAIMON_LOG_DEBUG(ScanLogger(), "Finding partitions for format table %s, partition filter: %s", + table_->FullName().c_str(), fmt::format("{}", partition_filter_).c_str()); + std::shared_ptr<FileSystem> file_system = table_->GetFileSystem(); + + const bool only_value = table_->PartitionOnlyValueInPath(); + // The one hidden name that is table content: a null partition's directory in the value-only + // layout. + const std::string& default_partition_name = table_->PartitionDefaultName(); + + // One partition level at a time, keeping each partition paired with its directory. + std::vector<PartitionAndPath> level = { + {std::map<std::string, std::string>(), table_->Location()}}; + bool at_table_location = true; + for (const std::string& partition_key : partition_keys) { + std::vector<PartitionAndPath> next; + for (const auto& [partition, directory] : level) { + std::vector<BasicFileStatus> children; + Status status = file_system->ListDir(directory, &children); + if (status.IsNotExist()) { + // Gone since it was listed, or a table directory not created yet; either way + // the rest of the listing stands. + continue; + } + PAIMON_RETURN_NOT_OK(status); + for (const BasicFileStatus& child : children) { + if (!child.IsDir()) { + continue; + } + std::string name = PathUtil::GetName(child.GetPath()); + if (PartitionPathUtils::IsHiddenName(name) && + !(only_value && name == default_partition_name)) { + continue; + } + if (at_table_location && table_->LocationCarriesPaimonMetadata() && + FormatFileListing::IsReservedDirectory(name)) { + continue; + } + std::string value; + if (only_value) { + // Nothing but the level says which key a directory belongs to, so every + // directory here is one. + value = PartitionPathUtils::UnescapePathName(name); + } else { + std::optional<std::pair<std::string, std::string>> key_value = + PartitionPathUtils::ExtractPartitionKeyValue(name); + if (!key_value || key_value->first != partition_key) { + // Something else lives here: another layout, or a nested table. + continue; + } + value = std::move(key_value->second); + } + auto filter_iter = partition_filter_.find(partition_key); + if (filter_iter != partition_filter_.end() && filter_iter->second != value) { Review Comment: Could we consider evaluating partition filters on a typed `BinaryRow` instead of comparing raw path strings? ```cpp PAIMON_ASSIGN_OR_RAISE( BinaryRow partition, partition_computer_->ToBinaryRow(partition_spec)); PAIMON_ASSIGN_OR_RAISE( bool matched, partition_filter_->Test(partition_schema_, partition)); ``` This would align the behavior with Java and the regular table scan path, particularly for typed values and null partition semantics, while also making richer partition predicates easier to support later. -- 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]
