SteNicholas commented on code in PR #222: URL: https://github.com/apache/paimon-cpp/pull/222#discussion_r3954190934
########## 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: Test-only. It is private now, and the test entry point is `FormatTableRead::TEST_Create()`; production goes through `Create(table, read_context)`, which reads the projection, the predicate, whether to apply it exactly, the pool, the executor and the file-open options out of the context. `CreateInternal()` is only the shared body of the two, and takes every setting already resolved, so each has exactly one source and no `ReadContext` is read there as well — which is what made the old signature look 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: Good pointer — TODO added at the open site in `format_table_read.cpp`, naming #189 and what is missing here: a way for a `FormatDataSplit` to vouch that it reached the read unedited. `Open()` trusts the length it is handed, and a stale one would truncate an object-store read or send it past the end, which is why the stat is still there for now. ########## 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: Done. `FormatTableScan` builds a `BinaryRowPartitionComputer` at `Create()` and converts each filter entry once through `ToPartialBinaryRow()`; discovery converts a directory's value the same way and compares the two rows. So `1` and `01` name the same `INT` partition, a blank value stays apart from the null partition even though a directory name cannot tell the two apart, and a filter value that will not read into the column's type is refused up front rather than silently matching nothing. One key at a time, since discovery tests one directory level at a time and only a row of the same shape can be compared against it. ########## 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: Removed. Partition columns are not written to the file at all — the directory holds those values and the read rebuilds them from its name — so a row's own partition column is no longer compared against the batch's declaration, which is taken on trust as on the managed path. -- 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]
