SteNicholas commented on code in PR #447: URL: https://github.com/apache/paimon-cpp/pull/447#discussion_r4361383248
########## src/paimon/core/table/file_store_table.cpp: ########## @@ -0,0 +1,1218 @@ +/* + * 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/file_store_table.h" + +#include <algorithm> +#include <map> +#include <memory> +#include <optional> +#include <string> +#include <utility> +#include <vector> + +#include "fmt/format.h" +#include "paimon/catalog/catalog.h" +#include "paimon/catalog/identifier.h" +#include "paimon/commit_context.h" +#include "paimon/common/predicate/predicate_validator.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/binary_row_partition_computer.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/options_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/sensitive_config_utils.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/catalog/catalog_snapshot_commit.h" +#include "paimon/core/catalog/catalog_utils.h" +#include "paimon/core/catalog/renaming_snapshot_commit.h" +#include "paimon/core/catalog/snapshot_commit.h" +#include "paimon/core/catalog/version_managed_catalog.h" +#include "paimon/core/core_options.h" +#include "paimon/core/disk/io_manager.h" +#include "paimon/core/index/index_file_handler.h" +#include "paimon/core/index/pk/primary_key_index_definitions.h" +#include "paimon/core/manifest/index_manifest_file.h" +#include "paimon/core/operation/commit/realtime_commit_properties.h" +#include "paimon/core/operation/expire_snapshots.h" +#include "paimon/core/operation/file_store_commit_impl.h" +#include "paimon/core/operation/file_store_scan.h" +#include "paimon/core/operation/internal_read_context.h" +#include "paimon/core/options/table_type.h" +#include "paimon/core/realtime/realtime_context_impl.h" +#include "paimon/core/schema/schema_manager.h" +#include "paimon/core/schema/schema_validation.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/table/append_only_file_store_table.h" +#include "paimon/core/table/file_store.h" +#include "paimon/core/table/primary_key_file_store_table.h" +#include "paimon/core/table/source/data_evolution_batch_scan.h" +#include "paimon/core/table/source/data_table_batch_scan.h" +#include "paimon/core/table/source/data_table_stream_scan.h" +#include "paimon/core/table/source/fallback_table_read.h" +#include "paimon/core/table/source/primary_key_index_batch_scan.h" +#include "paimon/core/table/source/read_optimized_scan_options.h" +#include "paimon/core/table/source/realtime_table_scan.h" +#include "paimon/core/table/source/snapshot/snapshot_reader.h" +#include "paimon/core/table/system/system_table.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/core/utils/field_mapping.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/primary_key_table_utils.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/file_store_commit.h" +#include "paimon/file_store_write.h" +#include "paimon/format/file_format.h" +#include "paimon/fs/file_system.h" +#include "paimon/predicate/predicate_utils.h" +#include "paimon/read_context.h" +#include "paimon/realtime/realtime_context.h" +#include "paimon/result.h" +#include "paimon/scan_context.h" +#include "paimon/schema/schema.h" +#include "paimon/status.h" +#include "paimon/table/format/format_table.h" +#include "paimon/table/source/table_read.h" +#include "paimon/table/source/table_scan.h" +#include "paimon/write_context.h" + +namespace arrow { +class Schema; +} // namespace arrow + +namespace paimon { + +class Executor; +class MemoryPool; + +namespace { + +/// The identifier a table path names when no catalog name goes with it, the C++ reading of +/// the Java `SchemaManagerUtils#identifierFromPath` over `ignoreIfUnknownDatabase = true`: the +/// directory above the table's names the database when it carries the `.db` suffix, the way a +/// catalog lays its databases out, and a path that names no database - a table sitting directly +/// under the root, say - leaves the database unknown, which is all a caller that only knows the +/// directory can say. +Identifier IdentifierFromPath(const std::string& table_path, const std::string& branch) { + const std::vector<std::string> paths = StringUtils::Split(table_path, "/"); + if (paths.size() < 2) { + return Identifier(Identifier::kUnknownDatabase, + paths.empty() ? std::string() : paths.front()); + } + const std::string& database = paths[paths.size() - 2]; + const std::string::size_type index = database.rfind(Catalog::DB_SUFFIX); + if (index == std::string::npos) { + return Identifier(Identifier::kUnknownDatabase, paths.back(), branch); + } + return Identifier(database.substr(0, index), paths.back(), branch); +} + +/// Recognises a format table behind `table_path`, or returns null when it is not one. +/// +/// The scan, read, write and commit entry points below all take a path and have to answer the +/// same question before they can dispatch, so one place answers it and they cannot disagree. +/// Null is also the answer when no schema was found under the path at all: what to say about a +/// table that is not there is the managed path's to decide, and it says it in more detail. +/// +/// The table type is read from the schema, never from `options`: `type` is structural, and one +/// call's options must not decide what kind of table this is. `branch` names the branch the +/// schema was read from: anything but the main branch is refused once the schema turns out to be +/// a format table's, which has no branches to read. +Result<std::shared_ptr<FormatTable>> TryLoadFormatTable( + const std::shared_ptr<FileSystem>& file_system, const std::string& table_path, + const std::string& branch, const std::map<std::string, std::string>& options, + const std::shared_ptr<TableSchema>& table_schema) { + if (table_schema == nullptr) { + return std::shared_ptr<FormatTable>(); + } + // From the schema alone: see `FormatTable::Create()`. + PAIMON_ASSIGN_OR_RAISE(bool is_format_table, + TableTypeDefine::IsFormatTable(table_schema->Options())); + if (!is_format_table) { + return std::shared_ptr<FormatTable>(); + } + // A format table has no branches, so it is refused rather than served from the table path + // under another branch's name. See `SchemaValidation::ValidateFormatTableSchema()`. + if (BranchManager::NormalizeBranch(branch) != BranchManager::DEFAULT_MAIN_BRANCH) { + return Status::Invalid( + fmt::format("a format table has no branches, so {} cannot be read on branch '{}'", + table_path, branch)); + } + // A context names a path, never a catalog identifier, so the table is known by the + // directory it sits in - which names the database too, when it carries the `.db` suffix a + // catalog lays its databases out with. + Identifier identifier = IdentifierFromPath(table_path, branch); + return FormatTable::Create(file_system, table_path, identifier, + checked_pointer_cast<DataSchema>(table_schema), + /*location_carries_paimon_metadata=*/true, options); +} + +/// Whether an option of a copy names the file system its store resolves: the backend +/// `file-system` picks, or the access configuration a `fs.` key carries. A copy whose dynamic +/// options change one of these resolves to another file system, so the managers of the table it +/// was copied from - built over that table's resolution - are not its to inherit: they would +/// keep reading metadata through the backend the copy just left. +bool FileSystemResolutionMayChange(const TableSchema& schema, + const std::map<std::string, std::string>& dynamic_options) { + const std::map<std::string, std::string>& schema_options = schema.Options(); + for (const auto& [key, value] : dynamic_options) { + if (key != Options::FILE_SYSTEM && key.rfind("fs.", 0) != 0) { + continue; Review Comment: @lucasfang, recognize native S3 options when rebuilding managers. This check ignores supported options such as `s3.endpoint`, `s3.access-key` and `s3.session.token`. When the filesystem is resolved from options, `Copy` reparses these into a new filesystem but retains the old schema and snapshot managers, leaving metadata requests on the old endpoint or credentials while data requests use the new settings. Please include native S3 options in the invalidation check. An isolated check of this helper confirmed that changing `s3.endpoint` returns false, whereas changing `fs.s3.endpoint` returns true. -- 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]
