Copilot commented on code in PR #120: URL: https://github.com/apache/paimon-cpp/pull/120#discussion_r3472344240
########## src/paimon/catalog/snapshot_commit.h: ########## @@ -0,0 +1,40 @@ +/* + * 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 <string> +#include <vector> + +#include "paimon/core/partition/partition_statistics.h" +#include "paimon/core/snapshot.h" + Review Comment: `SnapshotCommit` uses `Result` but this header does not include the definition of `Result`, so any translation unit that includes it before `paimon/result.h` (e.g. `file_store_commit_impl.h`) will fail to compile. ########## src/paimon/catalog/file_system_catalog.cpp: ########## @@ -0,0 +1,559 @@ +/* + * 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/catalog/file_system_catalog.h" + +#include <algorithm> +#include <cstring> +#include <optional> +#include <set> +#include <utility> + +#include "arrow/c/bridge.h" +#include "fmt/format.h" +#include "fmt/ranges.h" +#include "paimon/catalog/identifier.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/core_options.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/table/system/system_table.h" +#include "paimon/core/table/system/system_table_schema.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/defs.h" +#include "paimon/fs/file_system.h" +#include "paimon/logging.h" +#include "paimon/result.h" + +namespace arrow { +class Schema; +} // namespace arrow +struct ArrowSchema; + +namespace paimon { +FileSystemCatalog::FileSystemCatalog(const std::shared_ptr<FileSystem>& fs, + const std::string& warehouse) + : fs_(fs), warehouse_(warehouse), logger_(Logger::GetLogger("FileSystemCatalog")) {} + +Status FileSystemCatalog::CreateDatabase(const std::string& db_name, + const std::map<std::string, std::string>& options, + bool ignore_if_exists) { + if (IsSystemDatabase(db_name)) { + return Status::Invalid( + fmt::format("Cannot create database for system database {}.", db_name)); + } + PAIMON_ASSIGN_OR_RAISE(bool exist, DatabaseExists(db_name)); + if (exist) { + if (ignore_if_exists) { + return Status::OK(); + } else { + return Status::Invalid(fmt::format("database {} already exist", db_name)); + } + } + return CreateDatabaseImpl(db_name, options); +} + +Status FileSystemCatalog::CreateDatabaseImpl(const std::string& db_name, + const std::map<std::string, std::string>& options) { + if (options.find(Catalog::DB_LOCATION_PROP) != options.end()) { + return Status::Invalid( + "Cannot specify location for a database when using fileSystem catalog."); + } + if (!options.empty()) { + std::string log_msg = fmt::format( + "Currently filesystem catalog can't store database properties, discard properties: " + "{{{}}}", + fmt::join(options, ", ")); + PAIMON_LOG_DEBUG(logger_, "%s", log_msg.c_str()); + } + std::string db_path = NewDatabasePath(warehouse_, db_name); + PAIMON_RETURN_NOT_OK(fs_->Mkdirs(db_path)); + return Status::OK(); +} + +Result<bool> FileSystemCatalog::DatabaseExists(const std::string& db_name) const { + if (IsSystemDatabase(db_name)) { + return Status::NotImplemented( + "do not support checking DatabaseExists for system database."); + } + return fs_->Exists(NewDatabasePath(warehouse_, db_name)); +} + +Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); + if (is_system_table) { + PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table_name, + identifier.GetSystemTableName()); + if (!system_table_name || !SystemTableLoader::IsSupported(system_table_name.value())) { + return false; + } + PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName()); + Identifier data_identifier(identifier.GetDatabaseName(), data_table_name); + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(data_identifier)); + return latest_schema != std::nullopt; + } + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(identifier)); + return latest_schema != std::nullopt; +} + +std::string FileSystemCatalog::GetDatabaseLocation(const std::string& db_name) const { + return NewDatabasePath(warehouse_, db_name); +} + +std::string FileSystemCatalog::GetTableLocation(const Identifier& identifier) const { + return NewDataTablePath(warehouse_, identifier); +} + +Status FileSystemCatalog::CreateTable(const Identifier& identifier, ArrowSchema* c_schema, + const std::vector<std::string>& partition_keys, + const std::vector<std::string>& primary_keys, + const std::map<std::string, std::string>& options, + bool ignore_if_exists) { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, IsSystemTable(identifier)); + if (is_system_table) { + return Status::Invalid( + fmt::format("Cannot create table for system table {}, please use data table.", + identifier.ToString())); + } + PAIMON_ASSIGN_OR_RAISE(bool db_exist, DatabaseExists(identifier.GetDatabaseName())); + if (!db_exist) { + return Status::Invalid( + fmt::format("database {} is not exist", identifier.GetDatabaseName())); + } + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(identifier)); + bool table_exist = (latest_schema != std::nullopt); + if (table_exist) { + if (ignore_if_exists) { + return Status::OK(); + } else { + return Status::Invalid( + fmt::format("table {} already exist", identifier.GetTableName())); + } + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> schema, + arrow::ImportSchema(c_schema)); + PAIMON_ASSIGN_OR_RAISE(bool is_object_store, FileSystem::IsObjectStore(warehouse_)); + if (is_object_store && + options.find("enable-object-store-catalog-in-inte-test") == options.end()) { + return Status::NotImplemented( + "create table operation does not support object store file system for now"); + } + SchemaManager schema_manager(fs_, NewDataTablePath(warehouse_, identifier)); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<TableSchema> table_schema, + schema_manager.CreateTable(schema, partition_keys, primary_keys, options)); + return Status::OK(); Review Comment: `table_schema` is assigned but never used. With `-Werror` this can fail builds on compilers that warn on unused local variables (e.g. Clang with `-Wall`). If the return value is intentionally ignored, explicitly mark it used. ########## src/paimon/catalog/renaming_snapshot_commit.h: ########## @@ -0,0 +1,77 @@ +/* + * 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 <memory> +#include <string> +#include <utility> +#include <vector> + +#include "paimon/core/catalog/snapshot_commit.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/fs/file_system.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { +class PartitionStatistics; + +/// A `SnapshotCommit` using file renaming to commit. +/// +/// @note When the file system is local or hdfs, rename is atomic. But if the file system is object +/// storage, we need additional lock protection. +// TODO(jinli.zjw): add additional lock protection for object storage. +class RenamingSnapshotCommit : public SnapshotCommit { + public: + RenamingSnapshotCommit(const std::shared_ptr<FileSystem>& fs, + const std::shared_ptr<SnapshotManager>& snapshot_manager) + : fs_(fs), snapshot_manager_(snapshot_manager) {} + + Result<bool> Commit(const Snapshot& snapshot, + const std::vector<PartitionStatistics>& statistics) override { + PAIMON_ASSIGN_OR_RAISE(std::string json_str, snapshot.ToJsonString()); Review Comment: `statistics` is unused in this inline method definition, which will trigger `-Wunused-parameter` (and the build uses `-Werror` for GCC). If the parameter is intentionally unused for this implementation, omit the name (or mark it `[[maybe_unused]]`). ########## include/paimon/catalog/identifier.h: ########## @@ -0,0 +1,62 @@ +/* + * 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 <optional> +#include <string> + +#include "paimon/result.h" +#include "paimon/type_fwd.h" +#include "paimon/visibility.h" + +namespace paimon { + +/// An identifier for a table containing database and table name. +class PAIMON_EXPORT Identifier { + public: + static const char kUnknownDatabase[]; + static const char kSystemTableSplitter[]; + static const char kSystemBranchPrefix[]; + static const char kDefaultMainBranch[]; + + explicit Identifier(const std::string& table); + Identifier(const std::string& database, const std::string& table); + + bool operator==(const Identifier& other); + const std::string& GetDatabaseName() const; + const std::string& GetTableName() const; Review Comment: `operator==` is missing the `const` qualifier. As a value type, `Identifier` comparisons should be callable on const instances (e.g. when used as keys or passed as const refs). ########## src/paimon/catalog/renaming_snapshot_commit_test.cpp: ########## @@ -0,0 +1,67 @@ +/* + * 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/catalog/renaming_snapshot_commit.h" + Review Comment: The new `renaming_snapshot_commit.h` file is under `src/paimon/catalog/`, but this test includes `paimon/core/catalog/renaming_snapshot_commit.h`. This path mismatch will prevent compilation unless the headers are moved or the include paths are updated consistently. ########## src/paimon/catalog/commit_table_request_test.cpp: ########## @@ -0,0 +1,106 @@ +/* + * 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/catalog/commit_table_request.h" + Review Comment: The header `commit_table_request.h` added in this PR is located under `src/paimon/catalog/`, but this test includes it via `paimon/core/catalog/...`. This include path mismatch will break compilation unless files are moved or includes are updated consistently. ########## src/paimon/catalog/file_system_catalog_test.cpp: ########## @@ -0,0 +1,1072 @@ +/* + * 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/catalog/file_system_catalog.h" + Review Comment: The new `file_system_catalog.h` in this PR is under `src/paimon/catalog/`, but this test includes `paimon/core/catalog/file_system_catalog.h`. This mismatch will break compilation unless files are moved or the include paths are updated consistently. ########## src/paimon/catalog/file_system_catalog.cpp: ########## @@ -0,0 +1,559 @@ +/* + * 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/catalog/file_system_catalog.h" + +#include <algorithm> +#include <cstring> +#include <optional> +#include <set> +#include <utility> + +#include "arrow/c/bridge.h" +#include "fmt/format.h" +#include "fmt/ranges.h" +#include "paimon/catalog/identifier.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/core_options.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/table/system/system_table.h" +#include "paimon/core/table/system/system_table_schema.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/defs.h" +#include "paimon/fs/file_system.h" +#include "paimon/logging.h" +#include "paimon/result.h" + +namespace arrow { +class Schema; +} // namespace arrow +struct ArrowSchema; + +namespace paimon { +FileSystemCatalog::FileSystemCatalog(const std::shared_ptr<FileSystem>& fs, + const std::string& warehouse) + : fs_(fs), warehouse_(warehouse), logger_(Logger::GetLogger("FileSystemCatalog")) {} + +Status FileSystemCatalog::CreateDatabase(const std::string& db_name, + const std::map<std::string, std::string>& options, + bool ignore_if_exists) { + if (IsSystemDatabase(db_name)) { + return Status::Invalid( + fmt::format("Cannot create database for system database {}.", db_name)); + } + PAIMON_ASSIGN_OR_RAISE(bool exist, DatabaseExists(db_name)); + if (exist) { + if (ignore_if_exists) { + return Status::OK(); + } else { + return Status::Invalid(fmt::format("database {} already exist", db_name)); + } + } + return CreateDatabaseImpl(db_name, options); +} + +Status FileSystemCatalog::CreateDatabaseImpl(const std::string& db_name, + const std::map<std::string, std::string>& options) { + if (options.find(Catalog::DB_LOCATION_PROP) != options.end()) { + return Status::Invalid( + "Cannot specify location for a database when using fileSystem catalog."); + } + if (!options.empty()) { + std::string log_msg = fmt::format( + "Currently filesystem catalog can't store database properties, discard properties: " + "{{{}}}", + fmt::join(options, ", ")); + PAIMON_LOG_DEBUG(logger_, "%s", log_msg.c_str()); + } + std::string db_path = NewDatabasePath(warehouse_, db_name); + PAIMON_RETURN_NOT_OK(fs_->Mkdirs(db_path)); + return Status::OK(); +} + +Result<bool> FileSystemCatalog::DatabaseExists(const std::string& db_name) const { + if (IsSystemDatabase(db_name)) { + return Status::NotImplemented( + "do not support checking DatabaseExists for system database."); + } + return fs_->Exists(NewDatabasePath(warehouse_, db_name)); +} + +Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); + if (is_system_table) { + PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table_name, + identifier.GetSystemTableName()); + if (!system_table_name || !SystemTableLoader::IsSupported(system_table_name.value())) { + return false; + } + PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName()); + Identifier data_identifier(identifier.GetDatabaseName(), data_table_name); + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(data_identifier)); + return latest_schema != std::nullopt; + } + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(identifier)); + return latest_schema != std::nullopt; +} + +std::string FileSystemCatalog::GetDatabaseLocation(const std::string& db_name) const { + return NewDatabasePath(warehouse_, db_name); +} + +std::string FileSystemCatalog::GetTableLocation(const Identifier& identifier) const { + return NewDataTablePath(warehouse_, identifier); +} + +Status FileSystemCatalog::CreateTable(const Identifier& identifier, ArrowSchema* c_schema, + const std::vector<std::string>& partition_keys, + const std::vector<std::string>& primary_keys, + const std::map<std::string, std::string>& options, + bool ignore_if_exists) { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, IsSystemTable(identifier)); + if (is_system_table) { + return Status::Invalid( + fmt::format("Cannot create table for system table {}, please use data table.", + identifier.ToString())); + } + PAIMON_ASSIGN_OR_RAISE(bool db_exist, DatabaseExists(identifier.GetDatabaseName())); + if (!db_exist) { + return Status::Invalid( + fmt::format("database {} is not exist", identifier.GetDatabaseName())); + } + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(identifier)); + bool table_exist = (latest_schema != std::nullopt); + if (table_exist) { + if (ignore_if_exists) { + return Status::OK(); + } else { + return Status::Invalid( + fmt::format("table {} already exist", identifier.GetTableName())); + } + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> schema, + arrow::ImportSchema(c_schema)); + PAIMON_ASSIGN_OR_RAISE(bool is_object_store, FileSystem::IsObjectStore(warehouse_)); + if (is_object_store && + options.find("enable-object-store-catalog-in-inte-test") == options.end()) { + return Status::NotImplemented( + "create table operation does not support object store file system for now"); + } + SchemaManager schema_manager(fs_, NewDataTablePath(warehouse_, identifier)); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<TableSchema> table_schema, + schema_manager.CreateTable(schema, partition_keys, primary_keys, options)); + return Status::OK(); +} + +Result<std::optional<std::shared_ptr<TableSchema>>> FileSystemCatalog::TableSchemaExists( + const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, IsSystemTable(identifier)); + if (is_system_table) { + return Status::NotImplemented( + "do not support checking TableSchemaExists for system table."); + } + SchemaManager schema_manager(fs_, NewDataTablePath(warehouse_, identifier)); + return schema_manager.Latest(); +} + +std::string FileSystemCatalog::GetRootPath() const { + return warehouse_; +} + +std::shared_ptr<FileSystem> FileSystemCatalog::GetFileSystem() const { + return fs_; +} + +bool FileSystemCatalog::IsSystemDatabase(const std::string& db_name) { + return db_name == SYSTEM_DATABASE_NAME; +} + +Result<bool> FileSystemCatalog::IsSpecifiedSystemTable(const Identifier& identifier) { + return identifier.IsSystemTable(); +} + +Result<bool> FileSystemCatalog::IsSystemTable(const Identifier& identifier) { + if (IsSystemDatabase(identifier.GetDatabaseName())) { + return true; + } + return IsSpecifiedSystemTable(identifier); +} + +std::string FileSystemCatalog::NewDatabasePath(const std::string& warehouse, + const std::string& db_name) { + return PathUtil::JoinPath(warehouse, db_name + DB_SUFFIX); +} + +std::string FileSystemCatalog::NewDataTablePath(const std::string& warehouse, + const Identifier& identifier) { + return PathUtil::JoinPath(NewDatabasePath(warehouse, identifier.GetDatabaseName()), + identifier.GetTableName()); +} + +Result<std::vector<std::string>> FileSystemCatalog::ListDatabases() const { + std::vector<std::unique_ptr<BasicFileStatus>> file_status_list; + PAIMON_RETURN_NOT_OK(fs_->ListDir(warehouse_, &file_status_list)); + std::vector<std::string> db_names; + for (const auto& file_status : file_status_list) { + if (file_status->IsDir()) { + std::string name = PathUtil::GetName(file_status->GetPath()); + if (StringUtils::EndsWith(name, DB_SUFFIX)) { + db_names.push_back(name.substr(0, name.length() - std::strlen(DB_SUFFIX))); + } + } + } + return db_names; +} + +Result<std::vector<std::string>> FileSystemCatalog::ListTables(const std::string& db_name) const { + if (IsSystemDatabase(db_name)) { + return Status::NotImplemented("do not support listing tables for system database."); + } + std::string database_path = NewDatabasePath(warehouse_, db_name); + std::vector<std::unique_ptr<BasicFileStatus>> file_status_list; + PAIMON_RETURN_NOT_OK(fs_->ListDir(database_path, &file_status_list)); + std::vector<std::string> table_names; + for (const auto& file_status : file_status_list) { + if (file_status->IsDir()) { + std::string table_path = file_status->GetPath(); + PAIMON_ASSIGN_OR_RAISE(bool table_exist, TableExistsInFileSystem(table_path)); + if (table_exist) { + table_names.push_back(PathUtil::GetName(table_path)); + } + } + } + return table_names; +} + +Result<bool> FileSystemCatalog::TableExistsInFileSystem(const std::string& table_path) const { + SchemaManager schema_manager(fs_, table_path); + // in order to improve the performance, check the schema-0 firstly. + PAIMON_ASSIGN_OR_RAISE(bool schema_zero_exists, schema_manager.SchemaExists(0)); + if (schema_zero_exists) { + return true; + } else { + // if schema-0 not exists, fallback to check other schemas + PAIMON_ASSIGN_OR_RAISE(std::vector<int64_t> schema_ids, schema_manager.ListAllIds()); + return !schema_ids.empty(); + } +} + +Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema( + const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); + if (is_system_table) { + PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table_name, + identifier.GetSystemTableName()); + if (!system_table_name || !SystemTableLoader::IsSupported(system_table_name.value())) { + return Status::NotExist(fmt::format("{} not exist", identifier.ToString())); + } + PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName()); + Identifier data_identifier(identifier.GetDatabaseName(), data_table_name); + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(data_identifier)); + if (!latest_schema) { + return Status::NotExist(fmt::format("{} not exist", data_identifier.ToString())); + } + std::map<std::string, std::string> dynamic_options; + PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> branch, identifier.GetBranchName()); + if (branch) { + dynamic_options[Options::BRANCH] = branch.value(); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<SystemTable> system_table, + SystemTableLoader::Load(system_table_name.value(), fs_, + GetTableLocation(data_identifier), + latest_schema.value(), dynamic_options)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> arrow_schema, + system_table->ArrowSchema()); + return std::make_shared<SystemTableSchema>(std::move(arrow_schema)); + } + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(identifier)); + if (!latest_schema) { + return Status::NotExist(fmt::format("{} not exist", identifier.ToString())); + } + return std::static_pointer_cast<Schema>(*latest_schema); +} + +Result<std::shared_ptr<Table>> FileSystemCatalog::GetTable(const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); + if (is_system_table) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Schema> schema, LoadTableSchema(identifier)); Review Comment: `GetTable` uses `Identifier::IsSystemTable()` (only `$`-syntax) instead of `FileSystemCatalog::IsSystemTable()` (also reserves the system database). This can make `sys.*` behave like a normal data table here while being rejected elsewhere. ########## src/paimon/catalog/file_system_catalog.cpp: ########## @@ -0,0 +1,559 @@ +/* + * 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/catalog/file_system_catalog.h" + +#include <algorithm> +#include <cstring> +#include <optional> +#include <set> +#include <utility> + +#include "arrow/c/bridge.h" +#include "fmt/format.h" +#include "fmt/ranges.h" +#include "paimon/catalog/identifier.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/core_options.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/table/system/system_table.h" +#include "paimon/core/table/system/system_table_schema.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/defs.h" +#include "paimon/fs/file_system.h" +#include "paimon/logging.h" +#include "paimon/result.h" + +namespace arrow { +class Schema; +} // namespace arrow +struct ArrowSchema; + +namespace paimon { +FileSystemCatalog::FileSystemCatalog(const std::shared_ptr<FileSystem>& fs, + const std::string& warehouse) + : fs_(fs), warehouse_(warehouse), logger_(Logger::GetLogger("FileSystemCatalog")) {} + +Status FileSystemCatalog::CreateDatabase(const std::string& db_name, + const std::map<std::string, std::string>& options, + bool ignore_if_exists) { + if (IsSystemDatabase(db_name)) { + return Status::Invalid( + fmt::format("Cannot create database for system database {}.", db_name)); + } + PAIMON_ASSIGN_OR_RAISE(bool exist, DatabaseExists(db_name)); + if (exist) { + if (ignore_if_exists) { + return Status::OK(); + } else { + return Status::Invalid(fmt::format("database {} already exist", db_name)); + } + } + return CreateDatabaseImpl(db_name, options); +} + +Status FileSystemCatalog::CreateDatabaseImpl(const std::string& db_name, + const std::map<std::string, std::string>& options) { + if (options.find(Catalog::DB_LOCATION_PROP) != options.end()) { + return Status::Invalid( + "Cannot specify location for a database when using fileSystem catalog."); + } + if (!options.empty()) { + std::string log_msg = fmt::format( + "Currently filesystem catalog can't store database properties, discard properties: " + "{{{}}}", + fmt::join(options, ", ")); + PAIMON_LOG_DEBUG(logger_, "%s", log_msg.c_str()); + } + std::string db_path = NewDatabasePath(warehouse_, db_name); + PAIMON_RETURN_NOT_OK(fs_->Mkdirs(db_path)); + return Status::OK(); +} + +Result<bool> FileSystemCatalog::DatabaseExists(const std::string& db_name) const { + if (IsSystemDatabase(db_name)) { + return Status::NotImplemented( + "do not support checking DatabaseExists for system database."); + } + return fs_->Exists(NewDatabasePath(warehouse_, db_name)); +} + +Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); + if (is_system_table) { Review Comment: This method treats system-database identifiers (`db == "sys"`) as *non*-system tables because it calls `Identifier::IsSystemTable()` (which only checks the `$` syntax). Other methods in this class use `FileSystemCatalog::IsSystemTable()` to also reserve the system database name, so this inconsistency can allow operations on `sys.*` to proceed unexpectedly. ########## src/paimon/catalog/identifier.cpp: ########## @@ -0,0 +1,124 @@ +/* + * 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/catalog/identifier.h" + +#include <cstring> +#include <optional> +#include <vector> + +#include "fmt/format.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +const char Identifier::kUnknownDatabase[] = "unknown"; +const char Identifier::kSystemTableSplitter[] = "$"; +const char Identifier::kSystemBranchPrefix[] = "branch_"; +const char Identifier::kDefaultMainBranch[] = "main"; + +Identifier::Identifier(const std::string& table) + : Identifier(std::string(kUnknownDatabase), table) {} + +Identifier::Identifier(const std::string& database, const std::string& table) + : database_(database), table_(table) {} + +bool Identifier::operator==(const Identifier& other) { + return (database_ == other.database_ && table_ == other.table_); +} Review Comment: Definition of `Identifier::operator==` must match the header change to be `const`, otherwise it won’t compile / won’t be callable on const objects. ########## src/paimon/catalog/file_system_catalog.cpp: ########## @@ -0,0 +1,559 @@ +/* + * 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/catalog/file_system_catalog.h" + +#include <algorithm> +#include <cstring> +#include <optional> +#include <set> +#include <utility> + +#include "arrow/c/bridge.h" +#include "fmt/format.h" +#include "fmt/ranges.h" +#include "paimon/catalog/identifier.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/core_options.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/table/system/system_table.h" +#include "paimon/core/table/system/system_table_schema.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/defs.h" +#include "paimon/fs/file_system.h" +#include "paimon/logging.h" +#include "paimon/result.h" + +namespace arrow { +class Schema; +} // namespace arrow +struct ArrowSchema; + +namespace paimon { +FileSystemCatalog::FileSystemCatalog(const std::shared_ptr<FileSystem>& fs, + const std::string& warehouse) + : fs_(fs), warehouse_(warehouse), logger_(Logger::GetLogger("FileSystemCatalog")) {} + +Status FileSystemCatalog::CreateDatabase(const std::string& db_name, + const std::map<std::string, std::string>& options, + bool ignore_if_exists) { + if (IsSystemDatabase(db_name)) { + return Status::Invalid( + fmt::format("Cannot create database for system database {}.", db_name)); + } + PAIMON_ASSIGN_OR_RAISE(bool exist, DatabaseExists(db_name)); + if (exist) { + if (ignore_if_exists) { + return Status::OK(); + } else { + return Status::Invalid(fmt::format("database {} already exist", db_name)); + } + } + return CreateDatabaseImpl(db_name, options); +} + +Status FileSystemCatalog::CreateDatabaseImpl(const std::string& db_name, + const std::map<std::string, std::string>& options) { + if (options.find(Catalog::DB_LOCATION_PROP) != options.end()) { + return Status::Invalid( + "Cannot specify location for a database when using fileSystem catalog."); + } + if (!options.empty()) { + std::string log_msg = fmt::format( + "Currently filesystem catalog can't store database properties, discard properties: " + "{{{}}}", + fmt::join(options, ", ")); + PAIMON_LOG_DEBUG(logger_, "%s", log_msg.c_str()); + } + std::string db_path = NewDatabasePath(warehouse_, db_name); + PAIMON_RETURN_NOT_OK(fs_->Mkdirs(db_path)); + return Status::OK(); +} + +Result<bool> FileSystemCatalog::DatabaseExists(const std::string& db_name) const { + if (IsSystemDatabase(db_name)) { + return Status::NotImplemented( + "do not support checking DatabaseExists for system database."); + } + return fs_->Exists(NewDatabasePath(warehouse_, db_name)); +} + +Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); + if (is_system_table) { + PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table_name, + identifier.GetSystemTableName()); + if (!system_table_name || !SystemTableLoader::IsSupported(system_table_name.value())) { + return false; + } + PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName()); + Identifier data_identifier(identifier.GetDatabaseName(), data_table_name); + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(data_identifier)); + return latest_schema != std::nullopt; + } + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(identifier)); + return latest_schema != std::nullopt; +} + +std::string FileSystemCatalog::GetDatabaseLocation(const std::string& db_name) const { + return NewDatabasePath(warehouse_, db_name); +} + +std::string FileSystemCatalog::GetTableLocation(const Identifier& identifier) const { + return NewDataTablePath(warehouse_, identifier); +} + +Status FileSystemCatalog::CreateTable(const Identifier& identifier, ArrowSchema* c_schema, + const std::vector<std::string>& partition_keys, + const std::vector<std::string>& primary_keys, + const std::map<std::string, std::string>& options, + bool ignore_if_exists) { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, IsSystemTable(identifier)); + if (is_system_table) { + return Status::Invalid( + fmt::format("Cannot create table for system table {}, please use data table.", + identifier.ToString())); + } + PAIMON_ASSIGN_OR_RAISE(bool db_exist, DatabaseExists(identifier.GetDatabaseName())); + if (!db_exist) { + return Status::Invalid( + fmt::format("database {} is not exist", identifier.GetDatabaseName())); + } + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema, + TableSchemaExists(identifier)); + bool table_exist = (latest_schema != std::nullopt); + if (table_exist) { + if (ignore_if_exists) { + return Status::OK(); + } else { + return Status::Invalid( + fmt::format("table {} already exist", identifier.GetTableName())); + } + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> schema, + arrow::ImportSchema(c_schema)); + PAIMON_ASSIGN_OR_RAISE(bool is_object_store, FileSystem::IsObjectStore(warehouse_)); + if (is_object_store && + options.find("enable-object-store-catalog-in-inte-test") == options.end()) { + return Status::NotImplemented( + "create table operation does not support object store file system for now"); + } + SchemaManager schema_manager(fs_, NewDataTablePath(warehouse_, identifier)); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<TableSchema> table_schema, + schema_manager.CreateTable(schema, partition_keys, primary_keys, options)); + return Status::OK(); +} + +Result<std::optional<std::shared_ptr<TableSchema>>> FileSystemCatalog::TableSchemaExists( + const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, IsSystemTable(identifier)); + if (is_system_table) { + return Status::NotImplemented( + "do not support checking TableSchemaExists for system table."); + } + SchemaManager schema_manager(fs_, NewDataTablePath(warehouse_, identifier)); + return schema_manager.Latest(); +} + +std::string FileSystemCatalog::GetRootPath() const { + return warehouse_; +} + +std::shared_ptr<FileSystem> FileSystemCatalog::GetFileSystem() const { + return fs_; +} + +bool FileSystemCatalog::IsSystemDatabase(const std::string& db_name) { + return db_name == SYSTEM_DATABASE_NAME; +} + +Result<bool> FileSystemCatalog::IsSpecifiedSystemTable(const Identifier& identifier) { + return identifier.IsSystemTable(); +} + +Result<bool> FileSystemCatalog::IsSystemTable(const Identifier& identifier) { + if (IsSystemDatabase(identifier.GetDatabaseName())) { + return true; + } + return IsSpecifiedSystemTable(identifier); +} + +std::string FileSystemCatalog::NewDatabasePath(const std::string& warehouse, + const std::string& db_name) { + return PathUtil::JoinPath(warehouse, db_name + DB_SUFFIX); +} + +std::string FileSystemCatalog::NewDataTablePath(const std::string& warehouse, + const Identifier& identifier) { + return PathUtil::JoinPath(NewDatabasePath(warehouse, identifier.GetDatabaseName()), + identifier.GetTableName()); +} + +Result<std::vector<std::string>> FileSystemCatalog::ListDatabases() const { + std::vector<std::unique_ptr<BasicFileStatus>> file_status_list; + PAIMON_RETURN_NOT_OK(fs_->ListDir(warehouse_, &file_status_list)); + std::vector<std::string> db_names; + for (const auto& file_status : file_status_list) { + if (file_status->IsDir()) { + std::string name = PathUtil::GetName(file_status->GetPath()); + if (StringUtils::EndsWith(name, DB_SUFFIX)) { + db_names.push_back(name.substr(0, name.length() - std::strlen(DB_SUFFIX))); + } + } + } + return db_names; +} + +Result<std::vector<std::string>> FileSystemCatalog::ListTables(const std::string& db_name) const { + if (IsSystemDatabase(db_name)) { + return Status::NotImplemented("do not support listing tables for system database."); + } + std::string database_path = NewDatabasePath(warehouse_, db_name); + std::vector<std::unique_ptr<BasicFileStatus>> file_status_list; + PAIMON_RETURN_NOT_OK(fs_->ListDir(database_path, &file_status_list)); + std::vector<std::string> table_names; + for (const auto& file_status : file_status_list) { + if (file_status->IsDir()) { + std::string table_path = file_status->GetPath(); + PAIMON_ASSIGN_OR_RAISE(bool table_exist, TableExistsInFileSystem(table_path)); + if (table_exist) { + table_names.push_back(PathUtil::GetName(table_path)); + } + } + } + return table_names; +} + +Result<bool> FileSystemCatalog::TableExistsInFileSystem(const std::string& table_path) const { + SchemaManager schema_manager(fs_, table_path); + // in order to improve the performance, check the schema-0 firstly. + PAIMON_ASSIGN_OR_RAISE(bool schema_zero_exists, schema_manager.SchemaExists(0)); + if (schema_zero_exists) { + return true; + } else { + // if schema-0 not exists, fallback to check other schemas + PAIMON_ASSIGN_OR_RAISE(std::vector<int64_t> schema_ids, schema_manager.ListAllIds()); + return !schema_ids.empty(); + } +} + +Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema( + const Identifier& identifier) const { + PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); + if (is_system_table) { + PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table_name, Review Comment: `LoadTableSchema` also uses `Identifier::IsSystemTable()` which ignores the reserved system database name (`sys`). This is inconsistent with other paths (e.g. `CreateTable`/`DropTable`) that treat `sys` as system-only via `FileSystemCatalog::IsSystemTable()`. ########## src/paimon/catalog/catalog.cpp: ########## @@ -0,0 +1,40 @@ +/* + * 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/catalog/catalog.h" + +#include <utility> + +#include "paimon/core/catalog/file_system_catalog.h" +#include "paimon/core/core_options.h" + Review Comment: This include path (`paimon/core/catalog/file_system_catalog.h`) doesn’t match the file layout in this PR (the header lives under `src/paimon/catalog/`). This will fail to compile unless the file is moved or includes are updated consistently. ########## src/paimon/catalog/file_system_catalog.cpp: ########## @@ -0,0 +1,559 @@ +/* + * 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/catalog/file_system_catalog.h" + Review Comment: This include path doesn’t match the file layout in this PR (`file_system_catalog.h` is under `src/paimon/catalog/`, not `src/paimon/core/catalog/`). As-is, this source file won’t compile because the header cannot be found on the include path. ########## src/paimon/catalog/catalog_test.cpp: ########## @@ -0,0 +1,44 @@ +/* + * 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/catalog/catalog.h" + +#include "gtest/gtest.h" +#include "paimon/defs.h" +#include "paimon/testing/mock/mock_file_system.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +TEST(CatalogTest, Create) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create("path", options)); +} Review Comment: `catalog` is assigned but never used in this test, which can trigger `-Wunused-variable` (and the build uses `-Werror` on some compilers). ########## src/paimon/catalog/renaming_snapshot_commit.h: ########## @@ -0,0 +1,77 @@ +/* + * 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 <memory> +#include <string> +#include <utility> +#include <vector> + +#include "paimon/core/catalog/snapshot_commit.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/snapshot_manager.h" Review Comment: This include path does not match the file layout in this PR: `snapshot_commit.h` is added under `src/paimon/catalog/`, but here it is included via `paimon/core/catalog/...`. This will fail to compile unless the headers are moved or include paths are updated consistently. ########## src/paimon/catalog/catalog_snapshot_commit.h: ########## @@ -0,0 +1,50 @@ +/* + * 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 <string> +#include <vector> + +#include "paimon/core/catalog/commit_table_request.h" +#include "paimon/core/catalog/snapshot_commit.h" + Review Comment: This header uses `std::optional` and `Status::Invalid` but does not include `<optional>` or `paimon/status.h`, and it also includes the new catalog headers via a `paimon/core/catalog/...` path that doesn’t match their location under `src/paimon/catalog/` in this PR. ########## src/paimon/catalog/file_system_catalog_test.cpp: ########## @@ -0,0 +1,1072 @@ +/* + * 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/catalog/file_system_catalog.h" + +#include "arrow/api.h" +#include "arrow/c/abi.h" +#include "arrow/c/bridge.h" +#include "arrow/c/helpers.h" +#include "gtest/gtest.h" +#include "paimon/catalog/identifier.h" +#include "paimon/common/data/blob_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/core/core_options.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/table/system/system_table_schema.h" +#include "paimon/defs.h" +#include "paimon/fs/file_system.h" +#include "paimon/fs/file_system_factory.h" +#include "paimon/snapshot/snapshot_info.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +TEST(FileSystemCatalogTest, TestDatabaseExists) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + + ASSERT_OK_AND_ASSIGN(auto exist, catalog.DatabaseExists("db1")); + ASSERT_FALSE(exist); + + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/false)); + ASSERT_NOK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/false)); + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + + ASSERT_OK_AND_ASSIGN(exist, catalog.DatabaseExists("db1")); + ASSERT_TRUE(exist); + ASSERT_OK_AND_ASSIGN(std::vector<std::string> db_names, catalog.ListDatabases()); + ASSERT_EQ(1, db_names.size()); + ASSERT_EQ(db_names[0], "db1"); + ASSERT_EQ(catalog.GetDatabaseLocation("db1"), PathUtil::JoinPath(dir->Str(), "db1.db")); +} + +TEST(FileSystemCatalogTest, TestInvalidCreateDatabase) { + { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + options[Catalog::DB_LOCATION_PROP] = "loc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + + ASSERT_NOK_WITH_MSG( + catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true), + "Cannot specify location for a database when using fileSystem catalog."); + } +} + +TEST(FileSystemCatalogTest, TestCreateSystemDatabaseAndTable) { + // do not support create system database + { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_NOK_WITH_MSG(catalog.CreateDatabase(Catalog::SYSTEM_DATABASE_NAME, options, + /*ignore_if_exists=*/true), + "Cannot create database for system database"); + } + // do not support create system table + { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + arrow::FieldVector fields = { + arrow::field("f0", arrow::boolean()), + arrow::field("f1", arrow::int8()), + arrow::field("f2", arrow::int8()), + arrow::field("f3", arrow::int16()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + ASSERT_NOK_WITH_MSG( + catalog.CreateTable(Identifier("db1", "ta$ble"), &schema, {"f1"}, {}, options, false), + "Cannot create table for system table"); + ArrowSchemaRelease(&schema); + } +} + +TEST(FileSystemCatalogTest, TestCreateTable) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + + arrow::FieldVector fields = { + arrow::field("f0", arrow::boolean()), arrow::field("f1", arrow::int8()), + arrow::field("f2", arrow::int8()), arrow::field("f3", arrow::int16()), + arrow::field("f4", arrow::int16()), arrow::field("f5", arrow::int32()), + arrow::field("f6", arrow::int32()), arrow::field("f7", arrow::int64()), + arrow::field("f8", arrow::int64()), arrow::field("f9", arrow::float32()), + arrow::field("f10", arrow::float64()), arrow::field("f11", arrow::utf8()), + arrow::field("f12", arrow::binary()), arrow::field("non-partition-field", arrow::int32())}; + + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + Identifier identifier("db1", "tbl1"); + ASSERT_OK_AND_ASSIGN(auto exist, catalog.TableExists(identifier)); + ASSERT_FALSE(exist); + + ASSERT_OK(catalog.CreateTable(identifier, &schema, + /*partition_keys=*/{"f1", "f2"}, /*primary_keys=*/{"f3"}, options, + false)); + + ASSERT_OK_AND_ASSIGN(exist, catalog.TableExists(identifier)); + ASSERT_TRUE(exist); + ArrowSchemaRelease(&schema); +} + +TEST(FileSystemCatalogTest, TestOptionsSystemTableCatalog) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + options["custom.option"] = "custom-value"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + + auto typed_schema = arrow::schema({arrow::field("f0", arrow::int32())}); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &schema).ok()); + ASSERT_OK(catalog.CreateTable(Identifier("db1", "tbl1"), &schema, + /*partition_keys=*/{}, /*primary_keys=*/{}, options, + /*ignore_if_exists=*/false)); + ArrowSchemaRelease(&schema); + + Identifier options_identifier("db1", "tbl1$options"); + ASSERT_OK_AND_ASSIGN(bool exists, catalog.TableExists(options_identifier)); + ASSERT_TRUE(exists); + ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("db1", "tbl1$unknown"))); + ASSERT_FALSE(exists); + ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("db1", "missing$options"))); + ASSERT_FALSE(exists); + ASSERT_EQ(catalog.GetTableLocation(options_identifier), + PathUtil::JoinPath(PathUtil::JoinPath(dir->Str(), "db1.db"), "tbl1$options")); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> system_schema, + catalog.LoadTableSchema(options_identifier)); + ASSERT_TRUE(std::dynamic_pointer_cast<SystemSchema>(system_schema) != nullptr); + ASSERT_TRUE(std::dynamic_pointer_cast<SystemTableSchema>(system_schema) != nullptr); + ASSERT_OK_AND_ASSIGN(auto c_schema, system_schema->GetArrowSchema()); + auto loaded_schema_result = arrow::ImportSchema(c_schema.get()); + ASSERT_TRUE(loaded_schema_result.ok()) << loaded_schema_result.status().ToString(); + auto loaded_schema = loaded_schema_result.ValueUnsafe(); + ASSERT_EQ(loaded_schema->field_names(), (std::vector<std::string>{"key", "value"})); + ASSERT_EQ(loaded_schema->field(0)->type()->id(), arrow::Type::STRING); + ASSERT_EQ(loaded_schema->field(1)->type()->id(), arrow::Type::STRING); + ASSERT_FALSE(loaded_schema->field(0)->nullable()); + ASSERT_FALSE(loaded_schema->field(1)->nullable()); + + ASSERT_OK_AND_ASSIGN(auto system_table, catalog.GetTable(options_identifier)); + ASSERT_EQ(system_table->Name(), "tbl1$options"); + ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db1", "tbl1$unknown")), "not exist"); + ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db1", "missing$options")), "not exist"); + + ::ArrowSchema system_create_schema; + ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &system_create_schema).ok()); + ASSERT_NOK_WITH_MSG( + catalog.CreateTable(options_identifier, &system_create_schema, {}, {}, options, false), + "Cannot create table for system table"); + ArrowSchemaRelease(&system_create_schema); + ASSERT_NOK_WITH_MSG(catalog.DropTable(options_identifier, false), "Cannot drop system table"); + ASSERT_NOK_WITH_MSG(catalog.RenameTable(options_identifier, Identifier("db1", "tbl2"), false), + "Cannot rename system table"); +} + +TEST(FileSystemCatalogTest, TestAuditLogAndBinlogSystemTableCatalog) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + + auto typed_schema = + arrow::schema({arrow::field("pk", arrow::utf8()), arrow::field("v", arrow::int32(), true)}); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &schema).ok()); + ASSERT_OK(catalog.CreateTable(Identifier("db1", "tbl1"), &schema, + /*partition_keys=*/{}, /*primary_keys=*/{"pk"}, options, + /*ignore_if_exists=*/false)); + ArrowSchemaRelease(&schema); + + Identifier audit_log_identifier("db1", "tbl1$audit_log"); + Identifier binlog_identifier("db1", "tbl1$binlog"); + ASSERT_OK_AND_ASSIGN(bool exists, catalog.TableExists(audit_log_identifier)); + ASSERT_TRUE(exists); + ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(binlog_identifier)); + ASSERT_TRUE(exists); + ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("db1", "tbl1$unknown"))); + ASSERT_FALSE(exists); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> audit_log_system_schema, + catalog.LoadTableSchema(audit_log_identifier)); + ASSERT_TRUE(std::dynamic_pointer_cast<SystemTableSchema>(audit_log_system_schema) != nullptr); + ASSERT_OK_AND_ASSIGN(auto audit_log_c_schema, audit_log_system_schema->GetArrowSchema()); + auto audit_log_schema_result = arrow::ImportSchema(audit_log_c_schema.get()); + ASSERT_TRUE(audit_log_schema_result.ok()) << audit_log_schema_result.status().ToString(); + auto audit_log_schema = audit_log_schema_result.ValueUnsafe(); + ASSERT_EQ(audit_log_schema->field_names(), (std::vector<std::string>{"rowkind", "pk", "v"})); + ASSERT_EQ(audit_log_schema->field(0)->type()->id(), arrow::Type::STRING); + ASSERT_EQ(audit_log_schema->field(1)->type()->id(), arrow::Type::STRING); + ASSERT_EQ(audit_log_schema->field(2)->type()->id(), arrow::Type::INT32); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> binlog_system_schema, + catalog.LoadTableSchema(binlog_identifier)); + ASSERT_TRUE(std::dynamic_pointer_cast<SystemTableSchema>(binlog_system_schema) != nullptr); + ASSERT_OK_AND_ASSIGN(auto binlog_c_schema, binlog_system_schema->GetArrowSchema()); + auto binlog_schema_result = arrow::ImportSchema(binlog_c_schema.get()); + ASSERT_TRUE(binlog_schema_result.ok()) << binlog_schema_result.status().ToString(); + auto binlog_schema = binlog_schema_result.ValueUnsafe(); + ASSERT_EQ(binlog_schema->field_names(), (std::vector<std::string>{"rowkind", "pk", "v"})); + ASSERT_EQ(binlog_schema->field(0)->type()->id(), arrow::Type::STRING); + ASSERT_EQ(binlog_schema->field(1)->type()->id(), arrow::Type::LIST); + ASSERT_EQ(binlog_schema->field(2)->type()->id(), arrow::Type::LIST); + auto binlog_pk_type = + std::dynamic_pointer_cast<arrow::ListType>(binlog_schema->field(1)->type()); + auto binlog_v_type = + std::dynamic_pointer_cast<arrow::ListType>(binlog_schema->field(2)->type()); + ASSERT_TRUE(binlog_pk_type); + ASSERT_TRUE(binlog_v_type); + ASSERT_EQ(binlog_pk_type->value_type()->id(), arrow::Type::STRING); + ASSERT_EQ(binlog_v_type->value_type()->id(), arrow::Type::INT32); + + ::ArrowSchema system_create_schema; + ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &system_create_schema).ok()); + ASSERT_NOK_WITH_MSG( + catalog.CreateTable(audit_log_identifier, &system_create_schema, {}, {}, options, false), + "Cannot create table for system table"); + ArrowSchemaRelease(&system_create_schema); + ASSERT_NOK_WITH_MSG(catalog.DropTable(binlog_identifier, false), "Cannot drop system table"); + ASSERT_NOK_WITH_MSG(catalog.RenameTable(audit_log_identifier, Identifier("db1", "tbl2"), false), + "Cannot rename system table"); +} + +TEST(FileSystemCatalogTest, TestCreateTableWithBlob) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + options[Options::DATA_EVOLUTION_ENABLED] = "true"; + options[Options::ROW_TRACKING_ENABLED] = "true"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + + arrow::FieldVector fields = {arrow::field("f0", arrow::boolean()), + arrow::field("f1", arrow::int8()), + arrow::field("f2", arrow::int8()), + arrow::field("f3", arrow::int16()), + arrow::field("f4", arrow::int16()), + arrow::field("f5", arrow::int32()), + arrow::field("f6", arrow::int32()), + arrow::field("f7", arrow::int64()), + arrow::field("f8", arrow::int64()), + arrow::field("f9", arrow::float32()), + arrow::field("f10", arrow::float64()), + arrow::field("f11", arrow::utf8()), + arrow::field("f12", arrow::binary()), + arrow::field("non-partition-field", arrow::int32()), + BlobUtils::ToArrowField("f13", /*nullable=*/false)}; + + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + ASSERT_OK(catalog.CreateTable(Identifier("db1", "tbl1"), &schema, + /*partition_keys=*/{"f1", "f2"}, /*primary_keys=*/{}, options, + false)); + ASSERT_OK_AND_ASSIGN(std::vector<std::string> table_names, catalog.ListTables("db1")); + ASSERT_EQ(1, table_names.size()); + ASSERT_EQ(table_names[0], "tbl1"); + ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> table_schema, + catalog.LoadTableSchema(Identifier("db1", "tbl1"))); + ASSERT_TRUE(std::dynamic_pointer_cast<DataSchema>(table_schema) != nullptr); + ASSERT_TRUE(std::dynamic_pointer_cast<TableSchema>(table_schema) != nullptr); + ASSERT_OK_AND_ASSIGN(auto arrow_schema, table_schema->GetArrowSchema()); + auto loaded_schema = arrow::ImportSchema(arrow_schema.get()).ValueOrDie(); + ASSERT_TRUE(typed_schema.Equals(loaded_schema)); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr<Table> table, catalog.GetTable(Identifier("db1", "tbl1"))); + ASSERT_OK_AND_ASSIGN(auto arrow_schema_from_get_table, table->LatestSchema()->GetArrowSchema()); + auto schema_from_get_table = + arrow::ImportSchema(arrow_schema_from_get_table.get()).ValueOrDie(); + ASSERT_TRUE(typed_schema.Equals(schema_from_get_table)); + ASSERT_EQ(table->FullName(), "db1.tbl1"); + + ASSERT_NOK_WITH_MSG(catalog.GetTable(Identifier("db1", "table_xaxa")), + "Identifier{database='db1', table='table_xaxa'} not exist"); + + ArrowSchemaRelease(&schema); +} + +TEST(FileSystemCatalogTest, TestInvalidCreateTable) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + arrow::FieldVector fields = { + arrow::field("f0", arrow::boolean()), arrow::field("f1", arrow::int8()), + arrow::field("f2", arrow::struct_({arrow::field("s1", arrow::int8()), + arrow::field("s1", arrow::int16())}))}; + + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + ASSERT_NOK_WITH_MSG( + catalog.CreateTable(Identifier("db1", "tbl1"), &schema, {"f0"}, {"f1"}, options, false), + "validate schema failed: read schema has duplicate field s1"); + ASSERT_OK_AND_ASSIGN(std::vector<std::string> table_names, catalog.ListTables("db1")); + ASSERT_EQ(0, table_names.size()); + ArrowSchemaRelease(&schema); +} + +TEST(FileSystemCatalogTest, TestCreateTableWhileDbNotExist) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + arrow::FieldVector fields = { + arrow::field("f0", arrow::boolean()), + arrow::field("f1", arrow::int8()), + arrow::field("f2", arrow::int8()), + arrow::field("f3", arrow::int16()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + ASSERT_NOK_WITH_MSG( + catalog.CreateTable(Identifier("db1", "table"), &schema, {"f1"}, {}, options, false), + "database db1 is not exist"); + ArrowSchemaRelease(&schema); +} + +TEST(FileSystemCatalogTest, TestCreateTableWhileTableExist) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + { + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + arrow::FieldVector fields = { + arrow::field("f0", arrow::boolean()), + arrow::field("f1", arrow::int8()), + arrow::field("f2", arrow::int8()), + arrow::field("f3", arrow::int16()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + ASSERT_OK(catalog.CreateTable(Identifier("db1", "tbl1"), &schema, {"f1"}, {}, options, + /*ignore_if_exists=*/false)); + ASSERT_OK(catalog.CreateTable(Identifier("db1", "tbl1"), &schema, {"f1"}, {}, options, + /*ignore_if_exists=*/true)); + ASSERT_OK_AND_ASSIGN(std::vector<std::string> table_names, catalog.ListTables("db1")); + ASSERT_EQ(1, table_names.size()); + ASSERT_EQ(table_names[0], "tbl1"); + } + { + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + arrow::FieldVector fields = { + arrow::field("f0", arrow::boolean()), + arrow::field("f1", arrow::int8()), + arrow::field("f2", arrow::int8()), + arrow::field("f3", arrow::int16()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + ASSERT_OK(catalog.CreateTable(Identifier("db1", "tbl1"), &schema, {"f1"}, {}, options, + /*ignore_if_exists=*/true)); + ASSERT_NOK_WITH_MSG( + catalog.CreateTable(Identifier("db1", "tbl1"), &schema, {"f1"}, {}, options, + /*ignore_if_exists=*/false), + "already exist"); + ArrowSchemaRelease(&schema); + } + { + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + arrow::FieldVector fields = { + arrow::field("f0", arrow::boolean()), + arrow::field("f1", arrow::int8()), + arrow::field("f2", arrow::int8()), + arrow::field("f3", arrow::int16()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + Identifier identifier("db1", "tbl1"); + ASSERT_OK(catalog.CreateTable(identifier, &schema, {"f1"}, {}, options, + /*ignore_if_exists=*/true)); + ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", dir->Str(), {})); + ASSERT_OK(fs->Delete( + PathUtil::JoinPath(catalog.GetTableLocation(identifier), "schema/schema-0"))); + ASSERT_OK(catalog.CreateTable(identifier, &schema, {"f1"}, {}, options, + /*ignore_if_exists=*/false)); + } +} + +TEST(FileSystemCatalogTest, TestInvalidList) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_NOK_WITH_MSG(catalog.ListTables("sys"), + "do not support listing tables for system database."); +} + +TEST(FileSystemCatalogTest, TestValidateTableSchema) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); + arrow::FieldVector fields = { + arrow::field("f0", arrow::utf8()), + arrow::field("f1", arrow::int32()), + arrow::field("f2", arrow::int32()), + arrow::field("f3", arrow::float64()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + Identifier identifier("db1", "tbl1"); + ASSERT_OK(catalog.CreateTable(identifier, &schema, {"f1"}, {}, options, + /*ignore_if_exists=*/false)); + + ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db0", "tbl0")), + "Identifier{database=\'db0\', table=\'tbl0\'} not exist"); + ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> table_schema, catalog.LoadTableSchema(identifier)); + auto data_schema = std::dynamic_pointer_cast<DataSchema>(table_schema); + ASSERT_TRUE(data_schema != nullptr); + ASSERT_EQ(0, data_schema->Id()); + ASSERT_EQ(3, data_schema->HighestFieldId()); + ASSERT_EQ(1, data_schema->PartitionKeys().size()); + ASSERT_EQ(0, data_schema->PrimaryKeys().size()); + ASSERT_EQ(-1, data_schema->NumBuckets()); + ASSERT_FALSE(table_schema->Comment().has_value()); + std::vector<std::string> field_names = table_schema->FieldNames(); + std::vector<std::string> expected_field_names = {"f0", "f1", "f2", "f3"}; + ASSERT_EQ(field_names, expected_field_names); + + FieldType type; + ASSERT_OK_AND_ASSIGN(type, table_schema->GetFieldType("f0")); + ASSERT_EQ(type, FieldType::STRING); + ASSERT_OK_AND_ASSIGN(type, table_schema->GetFieldType("f1")); + ASSERT_EQ(type, FieldType::INT); + ASSERT_OK_AND_ASSIGN(type, table_schema->GetFieldType("f2")); + ASSERT_EQ(type, FieldType::INT); + ASSERT_OK_AND_ASSIGN(type, table_schema->GetFieldType("f3")); + ASSERT_EQ(type, FieldType::DOUBLE); + ASSERT_NOK(table_schema->GetFieldType("f4")); + + ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", dir->Str(), {})); + std::string schema_path = + PathUtil::JoinPath(catalog.GetTableLocation(identifier), "schema/schema-0"); + std::string expected_json_schema; + ASSERT_OK(fs->ReadFile(schema_path, &expected_json_schema)); + + ASSERT_OK_AND_ASSIGN(auto json_schema, table_schema->GetJsonSchema()); + ASSERT_EQ(expected_json_schema, json_schema); + + ASSERT_OK_AND_ASSIGN(auto arrow_schema, table_schema->GetArrowSchema()); + auto loaded_schema = arrow::ImportSchema(arrow_schema.get()).ValueOrDie(); + ASSERT_TRUE(typed_schema.Equals(loaded_schema)); + + ASSERT_OK(fs->Delete(schema_path)); + ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(identifier), + "Identifier{database=\'db1\', table=\'tbl1\'} not exist"); + + ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db1", "tbl$11")), "not exist"); + ArrowSchemaRelease(&schema); +} + +TEST(FileSystemCatalogTest, TestDropDatabase) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + + // Test 1: Drop non-existent database with ignore_if_not_exists=true + ASSERT_OK(catalog.DropDatabase("non_existent_db", /*ignore_if_not_exists=*/true, + /*cascade=*/false)); + + // Test 2: Drop non-existent database with ignore_if_not_exists=false + ASSERT_NOK_WITH_MSG(catalog.DropDatabase("non_existent_db", /*ignore_if_not_exists=*/false, + /*cascade=*/false), + "database non_existent_db does not exist"); + + // Test 3: Drop empty database + ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); + ASSERT_OK(catalog.DropDatabase("test_db", /*ignore_if_not_exists=*/false, + /*cascade=*/false)); + ASSERT_OK_AND_ASSIGN(bool exist, catalog.DatabaseExists("test_db")); + ASSERT_FALSE(exist); + + // Test 4: Drop non-empty database without cascade + ASSERT_OK(catalog.CreateDatabase("test_db2", options, /*ignore_if_exists=*/false)); + arrow::FieldVector fields = { + arrow::field("f0", arrow::int32()), + arrow::field("f1", arrow::utf8()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + ASSERT_OK(catalog.CreateTable(Identifier("test_db2", "tbl1"), &schema, {}, {}, options, false)); + ASSERT_NOK_WITH_MSG(catalog.DropDatabase("test_db2", /*ignore_if_not_exists=*/false, + /*cascade=*/false), + "Cannot drop non-empty database test_db2. Use cascade=true to force."); + + // Test 5: Drop non-empty database with cascade + ASSERT_OK(catalog.DropDatabase("test_db2", /*ignore_if_not_exists=*/false, + /*cascade=*/true)); + ASSERT_OK_AND_ASSIGN(exist, catalog.DatabaseExists("test_db2")); + ASSERT_FALSE(exist); + ASSERT_OK_AND_ASSIGN(std::vector<std::string> tables, catalog.ListTables("test_db2")); + ASSERT_TRUE(tables.empty()); + + // Test 6: Drop system database + ASSERT_NOK_WITH_MSG(catalog.DropDatabase(Catalog::SYSTEM_DATABASE_NAME, + /*ignore_if_not_exists=*/false, + /*cascade=*/false), + "Cannot drop system database sys."); + + ArrowSchemaRelease(&schema); +} + +TEST(FileSystemCatalogTest, TestDropTable) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); + + // Test 1: Drop non-existent table with ignore_if_not_exists=true + ASSERT_OK(catalog.DropTable(Identifier("test_db", "non_existent_tbl"), + /*ignore_if_not_exists=*/true)); + + // Test 2: Drop non-existent table with ignore_if_not_exists=false + ASSERT_NOK_WITH_MSG( + catalog.DropTable(Identifier("test_db", "non_existent_tbl"), + /*ignore_if_not_exists=*/false), + "table Identifier{database='test_db', table='non_existent_tbl'} does not exist"); + + // Test 3: Drop existing table + arrow::FieldVector fields = { + arrow::field("f0", arrow::int32()), + arrow::field("f1", arrow::utf8()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + ASSERT_OK(catalog.CreateTable(Identifier("test_db", "tbl1"), &schema, {}, {}, options, false)); + ASSERT_OK(catalog.DropTable(Identifier("test_db", "tbl1"), /*ignore_if_not_exists=*/false)); + ASSERT_OK_AND_ASSIGN(bool exist, catalog.TableExists(Identifier("test_db", "tbl1"))); + ASSERT_FALSE(exist); + + // Test 4: Drop system table + ASSERT_NOK_WITH_MSG( + catalog.DropTable(Identifier("test_db", "tbl$system"), + /*ignore_if_not_exists=*/false), + "Cannot drop system table Identifier{database='test_db', table='tbl$system'}."); + + ArrowSchemaRelease(&schema); +} + +TEST(FileSystemCatalogTest, TestRenameTable) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); + + // Test 1: Rename non-existent table with ignore_if_not_exists=true + ASSERT_OK(catalog.RenameTable(Identifier("test_db", "non_existent_tbl"), + Identifier("test_db", "new_tbl"), + /*ignore_if_not_exists=*/true)); + + // Test 2: Rename non-existent table with ignore_if_not_exists=false + ASSERT_NOK_WITH_MSG( + catalog.RenameTable(Identifier("test_db", "non_existent_tbl"), + Identifier("test_db", "new_tbl"), + /*ignore_if_not_exists=*/false), + "source table Identifier{database='test_db', table='non_existent_tbl'} does not exist"); + + // Test 3: Normal rename + arrow::FieldVector fields = { + arrow::field("f0", arrow::int32()), + arrow::field("f1", arrow::utf8()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema1; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema1).ok()); + ASSERT_OK( + catalog.CreateTable(Identifier("test_db", "old_tbl"), &schema1, {}, {}, options, false)); + ASSERT_OK(catalog.RenameTable(Identifier("test_db", "old_tbl"), + Identifier("test_db", "new_tbl"), + /*ignore_if_not_exists=*/false)); + ASSERT_OK_AND_ASSIGN(bool old_exist, catalog.TableExists(Identifier("test_db", "old_tbl"))); + ASSERT_FALSE(old_exist); + ASSERT_OK_AND_ASSIGN(bool new_exist, catalog.TableExists(Identifier("test_db", "new_tbl"))); + ASSERT_TRUE(new_exist); + + // Test 4: Rename to existing table + ::ArrowSchema schema2; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema2).ok()); + ASSERT_OK(catalog.CreateTable(Identifier("test_db", "tbl2"), &schema2, {}, {}, options, false)); + ASSERT_NOK_WITH_MSG( + catalog.RenameTable(Identifier("test_db", "new_tbl"), Identifier("test_db", "tbl2"), + /*ignore_if_not_exists=*/false), + "target table Identifier{database='test_db', table='tbl2'} already exists"); + + // Test 5: Cross-database rename + ASSERT_OK(catalog.CreateDatabase("test_db2", options, /*ignore_if_exists=*/false)); + ASSERT_NOK_WITH_MSG( + catalog.RenameTable(Identifier("test_db", "new_tbl"), + Identifier("test_db2", "cross_db_tbl"), + /*ignore_if_not_exists=*/false), + "Cannot rename table across databases. Cross-database rename is not supported."); + + // Test 6: Rename system table + ASSERT_NOK_WITH_MSG(catalog.RenameTable(Identifier("test_db", "tbl$system"), + Identifier("test_db", "new_system_tbl"), + /*ignore_if_not_exists=*/false), + "Cannot rename system table"); + + ArrowSchemaRelease(&schema1); + ArrowSchemaRelease(&schema2); +} + +TEST(FileSystemCatalogTest, TestDropTableWithExternalPath) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); + + // Create external path directory + auto external_dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(external_dir); + std::string external_path = external_dir->Str(); + external_path = "FILE://" + external_path; + + // Create a file in external path to simulate external data + ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", dir->Str(), {})); + std::string external_data_file = PathUtil::JoinPath(external_path, "data-file.parquet"); + ASSERT_OK(fs->WriteFile(external_data_file, "test data", /*overwrite=*/true)); + + // Verify external data file exists + ASSERT_OK_AND_ASSIGN(bool external_file_exists, fs->Exists(external_data_file)); + ASSERT_TRUE(external_file_exists); + + // Create table with external path + arrow::FieldVector fields = { + arrow::field("f0", arrow::int32()), + arrow::field("f1", arrow::utf8()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + + std::map<std::string, std::string> table_options = options; + table_options[Options::DATA_FILE_EXTERNAL_PATHS] = external_path; + table_options[Options::DATA_FILE_EXTERNAL_PATHS_STRATEGY] = "round-robin"; + + ASSERT_OK(catalog.CreateTable(Identifier("test_db", "tbl_with_external"), &schema, {}, {}, + table_options, false)); + + // Verify table exists + ASSERT_OK_AND_ASSIGN(bool table_exists, + catalog.TableExists(Identifier("test_db", "tbl_with_external"))); + ASSERT_TRUE(table_exists); + + // Drop the table + ASSERT_OK(catalog.DropTable(Identifier("test_db", "tbl_with_external"), + /*ignore_if_not_exists=*/false)); + + // Verify table is dropped + ASSERT_OK_AND_ASSIGN(table_exists, + catalog.TableExists(Identifier("test_db", "tbl_with_external"))); + ASSERT_FALSE(table_exists); + + // Verify external path is also cleaned up + ASSERT_OK_AND_ASSIGN(external_file_exists, fs->Exists(external_path)); + ASSERT_FALSE(external_file_exists); + + ArrowSchemaRelease(&schema); +} + +TEST(FileSystemCatalogTest, TestDropTableWithMultipleExternalPaths) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); + + // Create multiple external path directories + auto external_dir1 = UniqueTestDirectory::Create(); + auto external_dir2 = UniqueTestDirectory::Create(); + ASSERT_TRUE(external_dir1); + ASSERT_TRUE(external_dir2); + + std::string external_path1 = external_dir1->Str(); + external_path1 = "FILE://" + external_path1; + std::string external_path2 = external_dir2->Str(); + external_path2 = "FILE://" + external_path2; + + // Create files in external paths + ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", dir->Str(), {})); + std::string external_data_file1 = PathUtil::JoinPath(external_path1, "data-1.parquet"); + std::string external_data_file2 = PathUtil::JoinPath(external_path2, "data-2.parquet"); + ASSERT_OK(fs->WriteFile(external_data_file1, "test data 1", /*overwrite=*/true)); + ASSERT_OK(fs->WriteFile(external_data_file2, "test data 2", /*overwrite=*/true)); + + // Create table with multiple external paths + arrow::FieldVector fields = { + arrow::field("f0", arrow::int32()), + arrow::field("f1", arrow::utf8()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + + std::map<std::string, std::string> table_options = options; + table_options[Options::DATA_FILE_EXTERNAL_PATHS] = external_path1 + "," + external_path2; + table_options[Options::DATA_FILE_EXTERNAL_PATHS_STRATEGY] = "round-robin"; + + ASSERT_OK(catalog.CreateTable(Identifier("test_db", "tbl_multi_external"), &schema, {}, {}, + table_options, false)); + + // Drop the table + ASSERT_OK(catalog.DropTable(Identifier("test_db", "tbl_multi_external"), + /*ignore_if_not_exists=*/false)); + + // Verify both external paths are cleaned up + ASSERT_OK_AND_ASSIGN(bool exists1, fs->Exists(external_path1)); + ASSERT_OK_AND_ASSIGN(bool exists2, fs->Exists(external_path2)); + ASSERT_FALSE(exists1); + ASSERT_FALSE(exists2); + + ArrowSchemaRelease(&schema); +} + +TEST(FileSystemCatalogTest, TestDropTableWithGlobalIndexExternalPathOnMainBranch) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); + + // Create external path directory for global index + auto global_index_dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(global_index_dir); + std::string global_index_path = global_index_dir->Str(); + global_index_path = "FILE://" + global_index_path; + + // Create a file in external path to simulate global index data + ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", dir->Str(), {})); + std::string index_file = PathUtil::JoinPath(global_index_path, "index-file.bin"); + ASSERT_OK(fs->WriteFile(index_file, "index data", /*overwrite=*/true)); + + // Verify index file exists + ASSERT_OK_AND_ASSIGN(bool index_file_exists, fs->Exists(index_file)); + ASSERT_TRUE(index_file_exists); + + // Create table with global index external path on main branch + arrow::FieldVector fields = { + arrow::field("f0", arrow::int32()), + arrow::field("f1", arrow::utf8()), + }; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + + std::map<std::string, std::string> table_options = options; + table_options[Options::GLOBAL_INDEX_EXTERNAL_PATH] = global_index_path; + + ASSERT_OK(catalog.CreateTable(Identifier("test_db", "tbl_global_index"), &schema, {}, {}, + table_options, false)); + + // Verify table exists + ASSERT_OK_AND_ASSIGN(bool table_exists, + catalog.TableExists(Identifier("test_db", "tbl_global_index"))); + ASSERT_TRUE(table_exists); + + // Drop the table + ASSERT_OK(catalog.DropTable(Identifier("test_db", "tbl_global_index"), + /*ignore_if_not_exists=*/false)); + + // Verify table is dropped + ASSERT_OK_AND_ASSIGN(table_exists, + catalog.TableExists(Identifier("test_db", "tbl_global_index"))); + ASSERT_FALSE(table_exists); + + // Verify global index external path is also cleaned up + ASSERT_OK_AND_ASSIGN(bool global_index_exists, fs->Exists(global_index_path)); + ASSERT_FALSE(global_index_exists); + + ArrowSchemaRelease(&schema); +} + +TEST(FileSystemCatalogTest, TestDropTableWithGlobalIndexExternalPathOnBranch) { + // Copy the real append_table_with_rt_branch.db test data, then patch the + // branch-rt schema-1 to include a GLOBAL_INDEX_EXTERNAL_PATH option. + // DropTable should discover and clean up that external path. + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + + // Copy the real test_data db into a temp directory + std::string test_data_path = GetDataDir() + "/orc/append_table_with_rt_branch.db"; + std::string db_path = dir->Str() + "/test_db.db"; + ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, db_path)); + + // Create a temporary external directory that represents branch global index data + auto branch_external_dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(branch_external_dir); + std::string branch_external_path = branch_external_dir->Str(); + branch_external_path = "FILE://" + branch_external_path; + + ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", dir->Str(), {})); + ASSERT_OK(fs->WriteFile(PathUtil::JoinPath(branch_external_path, "index.bin"), + "branch global index data", /*overwrite=*/true)); + + // Patch branch-rt/schema/schema-1: add GLOBAL_INDEX_EXTERNAL_PATH to options + std::string branch_schema_path = + PathUtil::JoinPath(db_path, "append_table_with_rt_branch/branch/branch-rt/schema/schema-1"); + std::string schema_content; + ASSERT_OK(fs->ReadFile(branch_schema_path, &schema_content)); + + // Insert the global index external path option into the JSON options block. + // Original: "file.format" : "orc" + // Patched: "file.format" : "orc", + // "global-index.external-path" : "<branch_external_path>" + std::string search_str = R"("file.format" : "orc")"; + std::string replace_str = R"("file.format" : "orc", + "global-index.external-path" : ")" + + branch_external_path + R"(")"; + auto pos = schema_content.rfind(search_str); + ASSERT_NE(pos, std::string::npos); + schema_content.replace(pos, search_str.length(), replace_str); + ASSERT_OK(fs->WriteFile(branch_schema_path, schema_content, /*overwrite=*/true)); + + // Verify external path exists before drop + ASSERT_OK_AND_ASSIGN(bool external_exists, fs->Exists(branch_external_path)); + ASSERT_TRUE(external_exists); + + // Drop the table via catalog + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + Identifier identifier("test_db", "append_table_with_rt_branch"); + ASSERT_OK_AND_ASSIGN(bool table_exists, catalog.TableExists(identifier)); + ASSERT_TRUE(table_exists); + + ASSERT_OK(catalog.DropTable(identifier, /*ignore_if_not_exists=*/false)); + + // Verify table is dropped + ASSERT_OK_AND_ASSIGN(table_exists, catalog.TableExists(identifier)); + ASSERT_FALSE(table_exists); + + // Verify the branch global index external path is cleaned up by DropTable + ASSERT_OK_AND_ASSIGN(external_exists, fs->Exists(branch_external_path)); + ASSERT_FALSE(external_exists); +} + +TEST(FileSystemCatalogTest, TestListSnapshots) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + + std::string test_data_path = GetDataDir() + "/append_table_with_multiple_file_format.db"; + std::string db_path = dir->Str() + "/test_db.db"; + ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, db_path)); + + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + + Identifier id("test_db", "append_table_with_multiple_file_format"); + ASSERT_OK_AND_ASSIGN(std::vector<SnapshotInfo> snapshots, catalog.ListSnapshots(id, "")); + + ASSERT_EQ(snapshots.size(), 2); + + ASSERT_EQ(snapshots[0].snapshot_id, 1); + ASSERT_EQ(snapshots[0].schema_id, 0); + ASSERT_EQ(snapshots[0].commit_kind, SnapshotInfo::CommitKind::APPEND); + ASSERT_EQ(snapshots[0].time_millis, 1755671728191); + ASSERT_EQ(snapshots[0].total_record_count, 5); + ASSERT_EQ(snapshots[0].delta_record_count, 5); + ASSERT_FALSE(snapshots[0].watermark.has_value()); + + ASSERT_EQ(snapshots[1].snapshot_id, 2); + ASSERT_EQ(snapshots[1].schema_id, 1); + ASSERT_EQ(snapshots[1].commit_kind, SnapshotInfo::CommitKind::APPEND); + ASSERT_EQ(snapshots[1].time_millis, 1755671956423); + ASSERT_EQ(snapshots[1].total_record_count, 7); + ASSERT_EQ(snapshots[1].delta_record_count, 2); + ASSERT_FALSE(snapshots[1].watermark.has_value()); + + // Verify ascending order by snapshot_id + ASSERT_LT(snapshots[0].snapshot_id, snapshots[1].snapshot_id); +} + +TEST(FileSystemCatalogTest, TestListSnapshotsTableNotExist) { + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + + ASSERT_NOK_WITH_MSG( + catalog.ListSnapshots(Identifier("non_existent_db", "non_existent_table"), ""), + "does not exist"); +} + +TEST(FileSystemCatalogTest, TestDropTableWithBranchExternalPaths) { + // Copy the real append_table_with_rt_branch.db test data, then patch the + // branch-rt schema-1 to include a DATA_FILE_EXTERNAL_PATHS option. + // DropTable should discover and clean up that external path. + std::map<std::string, std::string> options; + options[Options::FILE_SYSTEM] = "local"; + options[Options::FILE_FORMAT] = "orc"; + ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + + // Copy the real test_data db into a temp directory + std::string test_data_path = GetDataDir() + "orc/append_table_with_rt_branch.db"; + std::string db_path = dir->Str() + "/test_db.db"; + ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, db_path)); Review Comment: `test_data_path` is missing a path separator between `GetDataDir()` and `orc/...`, so the directory copy will fail (and the test will fail) unless `GetDataDir()` already ends with `/`. -- 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]
