SteNicholas commented on code in PR #222: URL: https://github.com/apache/paimon-cpp/pull/222#discussion_r3840609888
########## src/paimon/core/table/format/format_file_naming_test.cpp: ########## @@ -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. + */ + +#include "paimon/core/table/format/format_file_naming.h" + +#include <string> + +#include "gtest/gtest.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/status.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +TEST(FormatFileNamingTest, TestNamesAreDataPrefixedAndNumbered) { + ASSERT_OK_AND_ASSIGN(FormatFileNaming naming, FormatFileNaming::Create("parquet")); + std::string first = naming.NextFileName(); + std::string second = naming.NextFileName(); + + ASSERT_TRUE(StringUtils::StartsWith(first, "data-")); + ASSERT_TRUE(StringUtils::EndsWith(first, "-0.parquet")); + ASSERT_TRUE(StringUtils::EndsWith(second, "-1.parquet")); + // Both files of one write share its uuid. + ASSERT_EQ(first.substr(0, first.size() - std::string("-0.parquet").size()), + second.substr(0, second.size() - std::string("-1.parquet").size())); +} + +TEST(FormatFileNamingTest, TestTwoWritesDoNotCollide) { + ASSERT_OK_AND_ASSIGN(FormatFileNaming first_write, FormatFileNaming::Create("parquet")); + ASSERT_OK_AND_ASSIGN(FormatFileNaming second_write, FormatFileNaming::Create("parquet")); + ASSERT_NE(first_write.NextFileName(), second_write.NextFileName()); +} + +TEST(FormatFileNamingTest, TestTempNameIsHidden) { + std::string temp_name = FormatFileNaming::ToTempFileName("data-abc-0.parquet"); + ASSERT_EQ(temp_name, ".data-abc-0.parquet.tmp"); + // Both marks are what a scan and an orphan-file cleaner already skip. + ASSERT_EQ(temp_name[0], '.'); + ASSERT_TRUE(StringUtils::EndsWith(temp_name, ".tmp")); +} + +TEST(FormatFileNamingTest, TestPrefixComesFromDataFilePrefix) { + ASSERT_OK_AND_ASSIGN(FormatFileNaming naming, FormatFileNaming::Create("parquet", "part-")); + std::string name = naming.NextFileName(); + ASSERT_TRUE(StringUtils::StartsWith(name, "part-")); + ASSERT_TRUE(StringUtils::EndsWith(name, "-0.parquet")); +} + +TEST(FormatFileNamingTest, TestRejectsAPrefixThatIsNotOneFileNameComponent) { + // The prefix goes straight into a file name that is joined onto a directory, and the file is + // created before any commit sees it - so a separator or a `..` here would put data outside the + // table with nothing left to stop it. + for (const char* prefix : + {"nested/", "../outside-", "nested/../../outside-", "a\\b", "..", "."}) { + Result<FormatFileNaming> naming = FormatFileNaming::Create("parquet", prefix); + ASSERT_FALSE(naming.ok()) << prefix; Review Comment: Done. ########## src/paimon/core/table/format/format_path_validation.h: ########## @@ -0,0 +1,94 @@ +/* + * 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 { + Review Comment: Done. `FormatPathValidation` is a utility class holding the seven checks. ########## src/paimon/core/table/format/format_path_validation.cpp: ########## @@ -0,0 +1,288 @@ +/* + * 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_path_validation.h" + +#include <cstddef> +#include <optional> +#include <string> +#include <utility> +#include <vector> + +#include "fmt/format.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/core/table/format/format_file_listing.h" +#include "paimon/core/utils/partition_path_utils.h" + +namespace paimon { + +namespace { + +/// A table location as a prefix of the paths below it. +struct LocationPrefix { + /// The location without its trailing separator, so that a location written either way + /// compares the same. + std::string root; + /// Index in a path under the location where the first component below it starts. + size_t components_at = 0; +}; + +/// Resolves `directory` into the prefix the paths below it share. One place decides what "under +/// the table location" means, so that no two checks here can disagree about it. +/// +/// A location that is nothing but separators is the file system root, which is its own separator: +/// the component after it starts one character in, not two. An empty location names no directory, +/// and treating it as a prefix would make every absolute path pass. +Result<LocationPrefix> ResolveLocationPrefix(const std::string& directory, const char* subject, + const std::string& what) { + size_t end = directory.size(); + while (end > 0 && directory[end - 1] == '/') { + end--; + } + if (end == 0) { + if (directory.empty()) { + return Status::Invalid(fmt::format( + "{} cannot be checked: its {} is empty, and an empty path is a prefix of nothing", + what, subject)); + } + return LocationPrefix{"/", 1}; + } + return LocationPrefix{directory.substr(0, end), end + 1}; +} + +bool IsUnderPrefix(const std::string& path, const LocationPrefix& prefix) { + if (path.size() <= prefix.components_at || + path.compare(0, prefix.root.size(), prefix.root) != 0) { + return false; + } + // The root is its own separator when the location is the file system root; every other + // location is followed by one. + return prefix.components_at == prefix.root.size() || path[prefix.root.size()] == '/'; +} + +} // namespace + +Status ValidatePathUnderLocation(const std::string& path, const std::string& location, + const std::string& what) { + PAIMON_ASSIGN_OR_RAISE(LocationPrefix prefix, + ResolveLocationPrefix(location, "table location", what)); + if (!IsUnderPrefix(path, prefix)) { + return Status::Invalid(fmt::format( + "{} names '{}', which is not under the table location '{}'", what, path, location)); + } + + // Every component below the root has to be a name: `<table>/../victim` passes any prefix test + // and still resolves outside the table. + size_t begin = prefix.components_at; + while (begin <= path.size()) { + size_t end = path.find('/', begin); + if (end == std::string::npos) { + end = path.size(); + } + const std::string component = path.substr(begin, end - begin); + if (component.empty() || component == "." || component == "..") { + return Status::Invalid(fmt::format( + "{} names '{}', whose path does not stay inside the table location", what, path)); + } + begin = end + 1; + } + return Status::OK(); +} + +namespace { + +/// Fails when a scan would not reach `path`, whose last component names a file when `ends_in_file` +/// and a directory otherwise. One walk serves both: the distinction matters only for the last +/// component, which as a directory may be reserved or stand for a null partition. +Status ValidateComponentsAreVisible(const std::shared_ptr<FormatTable>& table, + const std::string& path, bool ends_in_file, + const std::string& what) { + const std::vector<std::string>& partition_keys = table->PartitionKeys(); + PAIMON_ASSIGN_OR_RAISE(LocationPrefix prefix, + ResolveLocationPrefix(table->Location(), "table location", what)); + const bool only_value = table->PartitionOnlyValueInPath(); + + size_t begin = prefix.components_at; + size_t level = 0; + while (begin <= path.size()) { + size_t end = path.find('/', begin); + const bool is_last = end == std::string::npos; + if (is_last) { + end = path.size(); + } + const std::string component = path.substr(begin, end - begin); + const bool is_directory = !is_last || !ends_in_file; + + // The one hidden name a scan reads is the directory standing for a null partition value in + // the value-only layout, and only where a partition directory belongs. + const bool is_default_partition_dir = is_directory && only_value && + level < partition_keys.size() && + component == table->PartitionDefaultName(); + if (PartitionPathUtils::IsHiddenName(component) && !is_default_partition_dir) { + return Status::Invalid(fmt::format( + "{} names '{}', which a scan of this table would skip: '{}' is hidden, and that is " + "how an uncommitted job marks its output", + what, path, component)); + } + // Only right below the location, and only for a table whose schema lives there. In the + // value-only layout a partition value lands here unescaped, so a partition could + // otherwise be named `schema` and be written over the table's own metadata. + if (level == 0 && is_directory && table->LocationCarriesPaimonMetadata() && + IsReservedFormatTableDirectory(component)) { + return Status::Invalid(fmt::format( + "{} names '{}', where '{}' is this table's own metadata rather than data", what, + path, component)); + } + begin = end + 1; + level++; + } + return Status::OK(); +} + +} // namespace + +Status ValidateFileIsVisible(const std::shared_ptr<FormatTable>& table, + const std::string& file_path, const std::string& what) { + return ValidateComponentsAreVisible(table, file_path, /*ends_in_file=*/true, what); +} + +Result<bool> IsTableLocation(const std::shared_ptr<FormatTable>& table, + const std::string& directory) { + const std::string what = fmt::format("table {}", table->FullName()); + PAIMON_ASSIGN_OR_RAISE(LocationPrefix location, + ResolveLocationPrefix(table->Location(), "table location", what)); + PAIMON_ASSIGN_OR_RAISE(LocationPrefix candidate, + ResolveLocationPrefix(directory, "directory", what)); + return location.root == candidate.root; +} + +Status ValidateDirectoryIsVisible(const std::shared_ptr<FormatTable>& table, + const std::string& directory, const std::string& what) { + // A directory written with a trailing separator names the same one without it, and the walk + // below would otherwise see an empty last component. + PAIMON_ASSIGN_OR_RAISE(LocationPrefix directory_prefix, + ResolveLocationPrefix(directory, "directory", what)); + return ValidateComponentsAreVisible(table, directory_prefix.root, /*ends_in_file=*/false, what); +} + +Status ValidatePartitionKeys(const std::shared_ptr<FormatTable>& table, + const std::map<std::string, std::string>& partition, + const std::string& what) { + const std::vector<std::string>& partition_keys = table->PartitionKeys(); + if (partition.size() != partition_keys.size()) { + return Status::Invalid( + fmt::format("{} carries {} partition values but table {} is partitioned by {} fields", + what, partition.size(), 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("{} does not carry a value for partition field '{}'", + what, partition_key)); + } + } + return Status::OK(); +} + +Status ValidateFileInPartition(const std::shared_ptr<FormatTable>& table, + const std::string& file_path, + const std::map<std::string, std::string>& partition, + const std::string& what) { + const std::vector<std::string>& partition_keys = table->PartitionKeys(); + PAIMON_ASSIGN_OR_RAISE(LocationPrefix prefix, + ResolveLocationPrefix(table->Location(), "table location", what)); + // The components between the location and the file name; the leading + // `partition_keys.size()` of them are the partition directories. + std::vector<std::string> components; + size_t begin = prefix.components_at; + while (begin < file_path.size()) { + size_t end = file_path.find('/', begin); + if (end == std::string::npos) { + break; + } + components.push_back(file_path.substr(begin, end - begin)); + begin = end + 1; + } + if (components.size() < partition_keys.size()) { + return Status::Invalid( + fmt::format("{} names '{}', which sits above the {} partition directories of table {}", + what, file_path, partition_keys.size(), table->FullName())); + } + + const bool only_value = table->PartitionOnlyValueInPath(); + for (size_t i = 0; i < partition_keys.size(); i++) { + const std::string& partition_key = partition_keys[i]; + std::string value; + if (only_value) { + value = PartitionPathUtils::UnescapePathName(components[i]); + } else { + std::optional<std::pair<std::string, std::string>> key_value = + PartitionPathUtils::ExtractPartitionKeyValue(components[i]); + if (!key_value || key_value->first != partition_key) { + return Status::Invalid( + fmt::format("{} names '{}', whose directory '{}' is not a partition of '{}'", + what, file_path, components[i], partition_key)); + } + value = key_value->second; + } + auto iter = partition.find(partition_key); + if (iter == partition.end() || iter->second != value) { + return Status::Invalid(fmt::format( + "{} sits in the '{}' partition of '{}' but claims '{}'", what, value, partition_key, + iter == partition.end() ? std::string("nothing") : iter->second)); + } + } + return Status::OK(); +} + +Result<std::string> PartitionDirectory(const std::shared_ptr<FormatTable>& table, + const std::map<std::string, std::string>& partition) { + const std::vector<std::string>& partition_keys = table->PartitionKeys(); Review Comment: `partition.legacy-name` is honoured now. The write reads the declared partition into its column types and renders it back through `BinaryRowPartitionComputer`, the same component the managed table path uses, so a `DATE` partition is written as its day count with the option on and as `YYYY-MM-DD` with it off, whatever the caller spelled. `TestLegacyPartitionNameDecidesHowADateIsWritten` covers both. The directory itself is still built by `FormatPathValidation::BuildPartitionDirectory()` rather than by `FileStorePathFactory`. A format table has no bucket directory, and `FileStorePathFactory`'s `data-file.path-directory` does not apply to it, which is also why Java's `FormatTableFileWriter` passes `pathFactory.root()` instead of `dataFilePath()`. Everything the factory would decide about a file name is shared, though: the prefix, the format extension and the `file.suffix.include.compression` suffix all follow the same rules. Happy to fold the directory building into the factory as well if you would rather have one code path. ########## src/paimon/core/table/format/format_table_read.cpp: ########## @@ -0,0 +1,438 @@ +/* + * 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_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/predicate/predicate_validator.h" +#include "paimon/common/reader/concat_batch_reader.h" +#include "paimon/common/reader/predicate_batch_reader.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/options_utils.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/table/format/format_path_validation.h" +#include "paimon/core/table/format/lazy_concat_batch_reader.h" +#include "paimon/core/table/format/limit_batch_reader.h" +#include "paimon/core/table/format/partition_completing_batch_reader.h" +#include "paimon/defs.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" +#include "paimon/predicate/predicate_builder.h" +#include "paimon/predicate/predicate_utils.h" +#include "paimon/table/format/format_data_split.h" + +namespace paimon { + +namespace { + +/// Whether every field the predicate names is one the reader will produce. +Result<bool> PredicateFitsSchema(const std::shared_ptr<Predicate>& predicate, + const std::set<std::string>& available_fields) { + std::set<std::string> named_fields; + PAIMON_RETURN_NOT_OK(PredicateUtils::GetAllNames(predicate, &named_fields)); + for (const std::string& field : named_fields) { + if (available_fields.find(field) == available_fields.end()) { + return false; + } + } + return true; +} + +std::set<std::string> FieldNamesOf(const arrow::FieldVector& fields) { + std::set<std::string> names; + for (const std::shared_ptr<arrow::Field>& field : fields) { + names.insert(field->name()); + } + return names; +} + +} // namespace + +/// Everything the read needs that the public header should not have to name. +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 read columns that live in the data files, in table order. + std::shared_ptr<arrow::Schema> data_read_schema; + /// For each read column, its index in `data_read_schema`, or -1 for a partition column. + std::vector<int32_t> read_to_data_index; + std::optional<int32_t> limit; + /// The conjuncts the file readers can evaluate, i.e. those naming only columns the file is + /// asked for. Null when there is none. + std::shared_ptr<Predicate> pushdown_predicate; + /// 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; + std::string format_identifier; + int32_t batch_size = 1024; +}; + +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::optional<int32_t>& limit, + const std::shared_ptr<MemoryPool>& pool, const std::shared_ptr<Predicate>& predicate, + bool enable_predicate_filter) { + 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 the same column 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->limit = limit; + impl->pool = memory_pool; + impl->format_identifier = FormatTable::FormatToString(table->GetFormat()); + + // Every non-partition column of the table, in table order: what a data file holds. + arrow::FieldVector file_fields; + for (const std::shared_ptr<arrow::Field>& field : table_schema->fields()) { + if (!is_partition_key(field->name())) { + file_fields.push_back(field); + } + } + if (file_fields.empty()) { + return Status::Invalid( + fmt::format("format table {} has no non-partition column, so its files hold nothing to " + "read", + table->FullName())); + } + + // Fields the predicate names, checked against the table up front: nothing downstream could + // evaluate a field the table does not have. + std::set<std::string> predicate_fields; + if (predicate != nullptr) { + PAIMON_RETURN_NOT_OK(PredicateUtils::GetAllNames(predicate, &predicate_fields)); + for (const std::string& field : predicate_fields) { + if (table_schema->GetFieldIndex(field) < 0) { + return Status::Invalid(fmt::format( + "predicate field '{}' is not a column of table {}", field, table->FullName())); + } + } + // The rest is what every read path in this library checks: that each conjunct's declared + // type is the column's, and that its literals are neither null nor of another type. The + // field index is not checked, since everything downstream resolves a field by name. + PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithSchema( + *table_schema, predicate, /*validate_field_idx=*/false)); + PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithLiterals(predicate)); + } + + // The columns read out of a file, in the order the caller asked for them. + arrow::FieldVector data_read_fields; + for (const std::shared_ptr<arrow::Field>& read_field : read_fields) { + if (!is_partition_key(read_field->name())) { + data_read_fields.push_back(read_field); + } + } + // A column only the predicate names is read as well, so the format can skip what its own + // statistics let it skip; the batch drops it again. + for (const std::shared_ptr<arrow::Field>& field : file_fields) { + if (predicate_fields.find(field->name()) == predicate_fields.end()) { + continue; + } + bool already_read = std::any_of(data_read_fields.begin(), data_read_fields.end(), + [&field](const std::shared_ptr<arrow::Field>& read_field) { + return read_field->name() == field->name(); + }); + if (!already_read) { + data_read_fields.push_back(field); + } + } + if (data_read_fields.empty()) { + // A partition-only projection still has to read something: only the file says how many + // rows to repeat the constant values for. + data_read_fields.push_back(file_fields.front()); Review Comment: Done. `PartitionCompletingBatchReader` is gone. The read builds a `FieldMappingBuilder` from the read schema and the partition keys and wraps each file in a `FieldMappingReader`, so partition completion, the read-schema order and the predicate split all come from the same component the managed table path uses; only the conjuncts naming columns the file holds are pushed into the format reader. ########## src/paimon/core/table/format/format_table_read.cpp: ########## @@ -0,0 +1,438 @@ +/* + * 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_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/predicate/predicate_validator.h" +#include "paimon/common/reader/concat_batch_reader.h" +#include "paimon/common/reader/predicate_batch_reader.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/options_utils.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/table/format/format_path_validation.h" +#include "paimon/core/table/format/lazy_concat_batch_reader.h" +#include "paimon/core/table/format/limit_batch_reader.h" +#include "paimon/core/table/format/partition_completing_batch_reader.h" +#include "paimon/defs.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" +#include "paimon/predicate/predicate_builder.h" +#include "paimon/predicate/predicate_utils.h" +#include "paimon/table/format/format_data_split.h" + +namespace paimon { + +namespace { + +/// Whether every field the predicate names is one the reader will produce. +Result<bool> PredicateFitsSchema(const std::shared_ptr<Predicate>& predicate, + const std::set<std::string>& available_fields) { + std::set<std::string> named_fields; + PAIMON_RETURN_NOT_OK(PredicateUtils::GetAllNames(predicate, &named_fields)); + for (const std::string& field : named_fields) { + if (available_fields.find(field) == available_fields.end()) { + return false; + } + } + return true; +} + +std::set<std::string> FieldNamesOf(const arrow::FieldVector& fields) { + std::set<std::string> names; + for (const std::shared_ptr<arrow::Field>& field : fields) { + names.insert(field->name()); + } + return names; +} + +} // namespace + +/// Everything the read needs that the public header should not have to name. +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 read columns that live in the data files, in table order. + std::shared_ptr<arrow::Schema> data_read_schema; + /// For each read column, its index in `data_read_schema`, or -1 for a partition column. + std::vector<int32_t> read_to_data_index; + std::optional<int32_t> limit; + /// The conjuncts the file readers can evaluate, i.e. those naming only columns the file is + /// asked for. Null when there is none. + std::shared_ptr<Predicate> pushdown_predicate; + /// 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; + std::string format_identifier; + int32_t batch_size = 1024; +}; + +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::optional<int32_t>& limit, + const std::shared_ptr<MemoryPool>& pool, const std::shared_ptr<Predicate>& predicate, + bool enable_predicate_filter) { + 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 the same column 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->limit = limit; + impl->pool = memory_pool; + impl->format_identifier = FormatTable::FormatToString(table->GetFormat()); + + // Every non-partition column of the table, in table order: what a data file holds. + arrow::FieldVector file_fields; + for (const std::shared_ptr<arrow::Field>& field : table_schema->fields()) { + if (!is_partition_key(field->name())) { + file_fields.push_back(field); + } + } + if (file_fields.empty()) { + return Status::Invalid( + fmt::format("format table {} has no non-partition column, so its files hold nothing to " + "read", + table->FullName())); + } + + // Fields the predicate names, checked against the table up front: nothing downstream could + // evaluate a field the table does not have. + std::set<std::string> predicate_fields; + if (predicate != nullptr) { + PAIMON_RETURN_NOT_OK(PredicateUtils::GetAllNames(predicate, &predicate_fields)); + for (const std::string& field : predicate_fields) { + if (table_schema->GetFieldIndex(field) < 0) { + return Status::Invalid(fmt::format( + "predicate field '{}' is not a column of table {}", field, table->FullName())); + } + } + // The rest is what every read path in this library checks: that each conjunct's declared + // type is the column's, and that its literals are neither null nor of another type. The + // field index is not checked, since everything downstream resolves a field by name. + PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithSchema( + *table_schema, predicate, /*validate_field_idx=*/false)); + PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithLiterals(predicate)); + } + + // The columns read out of a file, in the order the caller asked for them. + arrow::FieldVector data_read_fields; + for (const std::shared_ptr<arrow::Field>& read_field : read_fields) { + if (!is_partition_key(read_field->name())) { + data_read_fields.push_back(read_field); + } + } + // A column only the predicate names is read as well, so the format can skip what its own + // statistics let it skip; the batch drops it again. + for (const std::shared_ptr<arrow::Field>& field : file_fields) { + if (predicate_fields.find(field->name()) == predicate_fields.end()) { + continue; + } + bool already_read = std::any_of(data_read_fields.begin(), data_read_fields.end(), + [&field](const std::shared_ptr<arrow::Field>& read_field) { + return read_field->name() == field->name(); + }); + if (!already_read) { + data_read_fields.push_back(field); + } + } + if (data_read_fields.empty()) { + // A partition-only projection still has to read something: only the file says how many + // rows to repeat the constant values for. + data_read_fields.push_back(file_fields.front()); + } + impl->data_read_schema = arrow::schema(data_read_fields); + + impl->read_to_data_index.reserve(read_fields.size()); + for (const std::shared_ptr<arrow::Field>& read_field : read_fields) { + if (is_partition_key(read_field->name())) { + impl->read_to_data_index.push_back(-1); + } else { + impl->read_to_data_index.push_back( + static_cast<int32_t>(impl->data_read_schema->GetFieldIndex(read_field->name()))); + } + } + + if (predicate != nullptr) { + // A conjunct goes to whichever can evaluate every field it names: the file reader, which + // sees the non-partition columns, or the batch, which carries the projected ones plus the + // partitions. That leaves exactly the conjuncts naming a partition key out of the + // pushdown. + const std::set<std::string> data_read_field_names = + FieldNamesOf(impl->data_read_schema->fields()); + const std::set<std::string> read_field_names = FieldNamesOf(read_fields); + + std::vector<std::shared_ptr<Predicate>> pushdown_conjuncts; + std::vector<std::shared_ptr<Predicate>> filter_conjuncts; + for (const std::shared_ptr<Predicate>& conjunct : PredicateUtils::SplitAnd(predicate)) { + PAIMON_ASSIGN_OR_RAISE(bool fits_file, + PredicateFitsSchema(conjunct, data_read_field_names)); + if (fits_file) { + pushdown_conjuncts.push_back(conjunct); + } + PAIMON_ASSIGN_OR_RAISE(bool fits_batch, + PredicateFitsSchema(conjunct, read_field_names)); + if (fits_batch) { + filter_conjuncts.push_back(conjunct); + } else if (enable_predicate_filter) { + // Dropping it would return rows the caller asked to have filtered out. + return Status::Invalid(fmt::format( + "predicate '{}' of table {} names a column the projection drops, so the " + "reader cannot apply it; project that column or leave " + "enable_predicate_filter off", + conjunct->ToString(), table->FullName())); + } + } + if (!pushdown_conjuncts.empty()) { + PAIMON_ASSIGN_OR_RAISE(impl->pushdown_predicate, + PredicateBuilder::And(pushdown_conjuncts)); + } + if (enable_predicate_filter && !filter_conjuncts.empty()) { + PAIMON_ASSIGN_OR_RAISE(impl->filter_predicate, PredicateBuilder::And(filter_conjuncts)); + } + } + + PAIMON_ASSIGN_OR_RAISE(impl->batch_size, OptionsUtils::GetValueFromMap<int32_t>( + table->Options(), Options::READ_BATCH_SIZE, 1024)); + if (impl->batch_size <= 0) { + return Status::Invalid(fmt::format("{} must be larger than 0, but is {}", + Options::READ_BATCH_SIZE, impl->batch_size)); + } + + 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"); + } + + // A split is public and can be decoded from bytes this process did not write. Deserialization + // can only check the numbers; whether a file belongs to this table is asked here. + PAIMON_RETURN_NOT_OK(ValidatePartitionKeys(impl_->table, format_split->Partition(), "split")); + for (const FormatDataSplit::FileMeta& file : format_split->Files()) { + PAIMON_RETURN_NOT_OK( + 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(ValidateFileInPartition(impl_->table, file.file_path, + format_split->Partition(), "split")); + PAIMON_RETURN_NOT_OK(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)); + } + } + + std::vector<PartitionCompletingBatchReader::ColumnSource> column_sources; + column_sources.reserve(impl_->read_to_data_index.size()); + const std::map<std::string, std::string>& partition = format_split->Partition(); + for (int32_t i = 0; i < impl_->read_schema->num_fields(); i++) { + const std::shared_ptr<arrow::Field>& read_field = impl_->read_schema->field(i); + int32_t data_index = impl_->read_to_data_index[i]; + if (data_index >= 0) { + column_sources.push_back({data_index, std::string(), false}); + continue; + } + auto partition_iter = partition.find(read_field->name()); + if (partition_iter == partition.end()) { + return Status::Invalid(fmt::format( + "split does not carry a value for partition field '{}'", read_field->name())); + } + // A directory named after the default partition name stands for a null partition value. + bool is_null = partition_iter->second == impl_->table->PartitionDefaultName(); + column_sources.push_back({-1, partition_iter->second, is_null}); + } + + // The format and builder depend only on the table, so one serves the whole split. Shared, + // because each file's reader is built later, when that file is reached. + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<FileFormat> owned_format, + FileFormatFactory::Get(impl_->format_identifier, impl_->table->Options())); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReaderBuilder> owned_builder, + owned_format->CreateReaderBuilder(impl_->batch_size)); + owned_builder->WithMemoryPool(impl_->pool); + std::shared_ptr<FileFormat> file_format(std::move(owned_format)); + std::shared_ptr<ReaderBuilder> reader_builder(std::move(owned_builder)); + + // Captured by value, so a file's reader stays valid however long this `FormatTableRead` + // lives. + std::shared_ptr<FormatTable> table = impl_->table; + std::shared_ptr<arrow::Schema> data_read_schema = impl_->data_read_schema; + std::shared_ptr<arrow::Schema> read_schema = impl_->read_schema; + std::shared_ptr<Predicate> pushdown_predicate = impl_->pushdown_predicate; + std::shared_ptr<MemoryPool> pool = impl_->pool; + auto shared_column_sources = + std::make_shared<std::vector<PartitionCompletingBatchReader::ColumnSource>>( + std::move(column_sources)); + + // Each file is named alongside the factory that opens it, 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, file_format, reader_builder, data_read_schema, read_schema, + shared_column_sources, pool, pushdown_predicate, + file]() -> Result<std::unique_ptr<BatchReader>> { + // Captured only to outlive the builder it created, which some formats need. + (void)file_format; + // `Open` takes a status the caller vouches for, and the split's size is whatever it + // was given: a stale or forged 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())); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream, + table->GetFileSystem()->Open(status)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileBatchReader> file_reader, + reader_builder->Build(input_stream)); + + ::ArrowSchema c_read_schema; + ArrowSchemaMarkReleased(&c_read_schema); + ScopeGuard read_schema_guard( + [&c_read_schema]() { ArrowSchemaRelease(&c_read_schema); }); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*data_read_schema, &c_read_schema)); + PAIMON_RETURN_NOT_OK(file_reader->SetReadSchema(&c_read_schema, pushdown_predicate, + /*selection_bitmap=*/std::nullopt)); + + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<PartitionCompletingBatchReader> reader, + PartitionCompletingBatchReader::Create(std::move(file_reader), data_read_schema, + read_schema, *shared_column_sources, pool)); + return std::unique_ptr<BatchReader>(std::move(reader)); + }; + sources.push_back(std::move(source)); + } + + return std::make_unique<LazyConcatBatchReader>(std::move(sources), impl_->pool); +} + +Result<std::unique_ptr<BatchReader>> FormatTableRead::ApplyPredicateFilter( + std::unique_ptr<BatchReader>&& reader) { + if (impl_->filter_predicate == nullptr) { + return std::move(reader); + } + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<PredicateBatchReader> filtered, + PredicateBatchReader::Create(std::move(reader), impl_->filter_predicate, impl_->pool)); + return filtered; +} + +Result<std::unique_ptr<BatchReader>> FormatTableRead::ApplyLimit( + std::unique_ptr<BatchReader>&& reader) { + if (!impl_->limit) { + return std::move(reader); + } + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<LimitBatchReader> limited, + LimitBatchReader::Create(std::move(reader), std::max(0, *impl_->limit), impl_->pool)); + return limited; Review Comment: Agreed, and removed. `LimitBatchReader` is gone and `FormatTableRead` does not bound the reader it hands out. `FormatTableScan` keeps the limit only to return an empty plan for a non-positive one, which is what Java's `FormatTableScanPlan.splits()` does, since a `FormatDataSplit` has no row count to drop splits by. -- 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]
