diff --git a/include/paimon/api.h b/include/paimon/api.h index 81f236bc4..f188085d5 100644 --- a/include/paimon/api.h +++ b/include/paimon/api.h @@ -20,24 +20,25 @@ #pragma once -#include "paimon/commit_context.h" // IWYU pragma: export -#include "paimon/defs.h" // IWYU pragma: export -#include "paimon/factories/factory.h" // IWYU pragma: export -#include "paimon/file_store_commit.h" // IWYU pragma: export -#include "paimon/file_store_write.h" // IWYU pragma: export -#include "paimon/fs/file_system_factory.h" // IWYU pragma: export -#include "paimon/memory/memory_pool.h" // IWYU pragma: export -#include "paimon/predicate/predicate.h" // IWYU pragma: export -#include "paimon/read_context.h" // IWYU pragma: export -#include "paimon/reader/batch_reader.h" // IWYU pragma: export -#include "paimon/record_batch.h" // IWYU pragma: export -#include "paimon/result.h" // IWYU pragma: export -#include "paimon/scan_context.h" // IWYU pragma: export -#include "paimon/statistics_mode.h" // IWYU pragma: export -#include "paimon/status.h" // IWYU pragma: export -#include "paimon/table/source/table_read.h" // IWYU pragma: export -#include "paimon/table/source/table_scan.h" // IWYU pragma: export -#include "paimon/write_context.h" // IWYU pragma: export +#include "paimon/commit_context.h" // IWYU pragma: export +#include "paimon/defs.h" // IWYU pragma: export +#include "paimon/factories/factory.h" // IWYU pragma: export +#include "paimon/file_store_commit.h" // IWYU pragma: export +#include "paimon/file_store_write.h" // IWYU pragma: export +#include "paimon/fs/file_system_factory.h" // IWYU pragma: export +#include "paimon/memory/memory_pool.h" // IWYU pragma: export +#include "paimon/predicate/predicate.h" // IWYU pragma: export +#include "paimon/read_context.h" // IWYU pragma: export +#include "paimon/reader/batch_reader.h" // IWYU pragma: export +#include "paimon/record_batch.h" // IWYU pragma: export +#include "paimon/result.h" // IWYU pragma: export +#include "paimon/scan_context.h" // IWYU pragma: export +#include "paimon/snapshot/snapshot_file_scan.h" // IWYU pragma: export +#include "paimon/statistics_mode.h" // IWYU pragma: export +#include "paimon/status.h" // IWYU pragma: export +#include "paimon/table/source/table_read.h" // IWYU pragma: export +#include "paimon/table/source/table_scan.h" // IWYU pragma: export +#include "paimon/write_context.h" // IWYU pragma: export // IWYU pragma: begin_exports #include "paimon/realtime/realtime_context.h" diff --git a/include/paimon/snapshot/snapshot_file_scan.h b/include/paimon/snapshot/snapshot_file_scan.h new file mode 100644 index 000000000..7283b2cb4 --- /dev/null +++ b/include/paimon/snapshot/snapshot_file_scan.h @@ -0,0 +1,76 @@ +/* + * 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 +#include +#include +#include +#include +#include + +#include "paimon/result.h" +#include "paimon/visibility.h" + +namespace paimon { +class Executor; +class FileSystem; +class MemoryPool; +class ScanFilter; + +/// Lists the physical files required by a table snapshot. +class PAIMON_EXPORT SnapshotFileScan { + public: + SnapshotFileScan() = delete; + ~SnapshotFileScan() = delete; + + /// List the physical files required to materialize a snapshot. + /// + /// The result includes the snapshot and schema files, manifest lists and manifest files, + /// live data and changelog files, external file indexes, the index manifest and its live index + /// files, statistics, and real-time offset files. Mutable snapshot hints such as `LATEST` and + /// `EARLIEST` are not included. + /// + /// Partition and bucket filters apply only to files which belong to a partition or bucket. + /// Snapshot-level shared metadata is always returned. A manifest file is returned when it + /// cannot be pruned by manifest-level partition and bucket statistics, even if entry-level + /// filtering later removes all of its entries. Predicate filters are not supported. + /// + /// @param table_path Root path of the table. + /// @param branch Branch to list. An empty value selects the main branch. This parameter takes + /// precedence over a branch configured in options. + /// @param snapshot_id Snapshot to list, or `std::nullopt` to use the latest snapshot. + /// @param scan_filter Optional partition and bucket filters. Partition maps use AND semantics, + /// while the vector uses OR semantics. Predicate filters are rejected. + /// @param options User options overriding options stored in the latest table schema. + /// @param file_system Optional file system implementation. If null, it is resolved from + /// options. + /// @param executor Optional executor used to read manifest files in parallel. + /// @param memory_pool Optional memory pool. If null, the default pool is used. + /// @return Physical file paths required by the selected snapshot and filters. + static Result> ListFiles( + const std::string& table_path, const std::string& branch, + const std::optional& snapshot_id, const std::shared_ptr& scan_filter, + const std::map& options, + const std::shared_ptr& file_system, const std::shared_ptr& executor, + const std::shared_ptr& memory_pool); +}; + +} // namespace paimon diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 4aca72abc..e8a47d99e 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -399,6 +399,7 @@ set(PAIMON_CORE_SRCS core/schema/schema_validation.cpp core/schema/table_schema.cpp core/snapshot.cpp + core/snapshot_file_scan.cpp core/snapshot_info.cpp core/stats/simple_stats_collector.cpp core/stats/simple_stats_converter.cpp @@ -896,6 +897,7 @@ if(PAIMON_BUILD_TESTS) core/schema/schema_validation_test.cpp core/schema/arrow_schema_validator_test.cpp core/schema/table_schema_test.cpp + core/snapshot_file_scan_test.cpp core/snapshot_test.cpp core/stats/simple_stats_evolution_test.cpp core/stats/simple_stats_collector_test.cpp diff --git a/src/paimon/core/operation/file_store_scan.h b/src/paimon/core/operation/file_store_scan.h index ff53e0c9e..ea963a716 100644 --- a/src/paimon/core/operation/file_store_scan.h +++ b/src/paimon/core/operation/file_store_scan.h @@ -213,6 +213,13 @@ class FileStoreScan { Result> ReadPartitionEntries() const; + /// Merge raw manifest entries into the set of currently-live files. + /// + /// Entries are deduplicated by identifier. A Delete cancels the corresponding Add, and + /// lingering Delete entries are omitted from the result. + static Status MergeLiveEntries(const std::vector& unmerged_entries, + std::vector* live_entries); + protected: /// @note Keep this thread-safe. virtual Result FilterByStats(const ManifestEntry& entry) const = 0; @@ -273,13 +280,6 @@ class FileStoreScan { int32_t bucket, std::vector* merged_entries) const; - /// Merge raw manifest entries into the set of currently-live files. Entries are deduplicated - /// by identifier (matching Add cancels a prior or following Delete), and lingering Delete - /// entries are dropped so the caller receives Add-only output, matching the semantics of - /// `ReadAndMergeFileEntries`. - static Status MergeLiveEntries(const std::vector& unmerged_entries, - std::vector* live_entries); - Status ReadAndMergeFileEntries(const std::vector& manifest_metas, std::vector* merged_entries) const; diff --git a/src/paimon/core/snapshot_file_scan.cpp b/src/paimon/core/snapshot_file_scan.cpp new file mode 100644 index 000000000..700f20288 --- /dev/null +++ b/src/paimon/core/snapshot_file_scan.cpp @@ -0,0 +1,412 @@ +/* + * 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/snapshot/snapshot_file_scan.h" + +#include +#include +#include +#include +#include + +#include "paimon/common/executor/future.h" +#include "paimon/common/predicate/predicate_filter.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/core/core_options.h" +#include "paimon/core/index/index_file_handler.h" +#include "paimon/core/io/data_file_path_factory.h" +#include "paimon/core/manifest/file_kind.h" +#include "paimon/core/manifest/index_manifest_file.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/operation/commit/realtime_commit_properties.h" +#include "paimon/core/operation/file_store_scan.h" +#include "paimon/core/schema/schema_manager.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/core/utils/field_mapping.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/index_file_path_factories.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/defs.h" +#include "paimon/executor.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/scan_context.h" +#include "paimon/status.h" + +namespace paimon { +namespace { + +struct ManifestReadResult { + ManifestFileMeta meta; + std::vector entries; +}; + +class SnapshotFileCollector { + public: + SnapshotFileCollector(const std::shared_ptr& path_factory, + const std::shared_ptr& manifest_list, + const std::shared_ptr& manifest_file, + std::unique_ptr&& index_file_handler, + const std::shared_ptr& schema_manager, + int64_t latest_schema_id, + const std::shared_ptr& snapshot_manager, + const std::shared_ptr& executor, + const std::shared_ptr& partition_schema, + const std::shared_ptr& partition_filter, + const std::optional& bucket_id) + : path_factory_(path_factory), + manifest_list_(manifest_list), + manifest_file_(manifest_file), + index_file_handler_(std::move(index_file_handler)), + schema_manager_(schema_manager), + latest_schema_id_(latest_schema_id), + snapshot_manager_(snapshot_manager), + executor_(executor), + partition_schema_(partition_schema), + partition_filter_(partition_filter), + bucket_id_(bucket_id) {} + + Result> Collect(const Snapshot& snapshot) { + AddPath(snapshot_manager_->SnapshotPath(snapshot.Id())); + schema_ids_.insert(snapshot.SchemaId()); + schema_ids_.insert(latest_schema_id_); + + AddManifestList(snapshot.BaseManifestList()); + AddManifestList(snapshot.DeltaManifestList()); + if (snapshot.ChangelogManifestList()) { + AddManifestList(snapshot.ChangelogManifestList().value()); + } + if (snapshot.IndexManifest()) { + AddPath(path_factory_->ToManifestFilePath(snapshot.IndexManifest().value())); + } + if (snapshot.Statistics()) { + AddPath(path_factory_->ToStatsFilePath(snapshot.Statistics().value())); + } + std::optional offsets_path = + RealtimeCommitProperties::GetOffsetsPath(snapshot); + if (offsets_path) { + AddPath(offsets_path.value()); + } + + PAIMON_RETURN_NOT_OK(CollectDataFiles(snapshot)); + PAIMON_RETURN_NOT_OK(CollectChangelogFiles(snapshot)); + PAIMON_RETURN_NOT_OK(CollectIndexFiles(snapshot)); + for (int64_t schema_id : schema_ids_) { + AddPath(PathUtil::JoinPath(schema_manager_->SchemaDirectory(), + "schema-" + std::to_string(schema_id))); + } + + return std::move(paths_); + } + + private: + void AddPath(const std::string& path) { + if (!path.empty()) { + paths_.insert(path); + } + } + + void AddManifestList(const std::string& file_name) { + if (!file_name.empty()) { + AddPath(path_factory_->ToManifestListPath(file_name)); + } + } + + bool MayContainBucket(const ManifestFileMeta& meta) const { + if (!bucket_id_) { + return true; + } + const std::optional& min_bucket = meta.MinBucket(); + const std::optional& max_bucket = meta.MaxBucket(); + return !min_bucket || !max_bucket || + (bucket_id_.value() >= min_bucket.value() && + bucket_id_.value() <= max_bucket.value()); + } + + Result ShouldReadManifest(const ManifestFileMeta& meta) const { + if (!MayContainBucket(meta)) { + return false; + } + if (!partition_filter_) { + return true; + } + SimpleStats stats = meta.PartitionStats(); + return partition_filter_->Test(partition_schema_, + /*row_count=*/meta.NumAddedFiles() + meta.NumDeletedFiles(), + stats.MinValues(), stats.MaxValues(), stats.NullCounts()); + } + + Status ApplyPartitionFilter(std::vector* entries) const { + if (!partition_filter_) { + return Status::OK(); + } + std::vector filtered_entries; + filtered_entries.reserve(entries->size()); + for (ManifestEntry& entry : *entries) { + PAIMON_ASSIGN_OR_RAISE(bool saved, + partition_filter_->Test(partition_schema_, entry.Partition())); + if (saved) { + filtered_entries.push_back(std::move(entry)); + } + } + *entries = std::move(filtered_entries); + return Status::OK(); + } + + Result> ReadManifests( + const std::vector& metas) const { + std::unordered_set submitted_files; + std::vector>> futures; + for (const ManifestFileMeta& meta : metas) { + PAIMON_ASSIGN_OR_RAISE(bool should_read, ShouldReadManifest(meta)); + if (!should_read || !submitted_files.insert(meta.FileName()).second) { + continue; + } + futures.push_back(Via(executor_.get(), [this, meta]() -> Result { + std::vector entries; + if (bucket_id_) { + PAIMON_RETURN_NOT_OK(manifest_file_->ReadBucketEntries( + meta.FileName(), bucket_id_.value(), &entries)); + } else { + PAIMON_RETURN_NOT_OK( + manifest_file_->Read(meta.FileName(), /*filter=*/nullptr, &entries)); + } + PAIMON_RETURN_NOT_OK(ApplyPartitionFilter(&entries)); + return ManifestReadResult{meta, std::move(entries)}; + })); + } + + std::vector results; + std::vector> read_results = CollectAll(futures); + results.reserve(read_results.size()); + for (Result& result : read_results) { + PAIMON_RETURN_NOT_OK(result); + results.push_back(std::move(result).value()); + } + return results; + } + + Status CollectDataFiles(const Snapshot& snapshot) { + std::vector metas; + PAIMON_RETURN_NOT_OK(manifest_list_->ReadDataManifests(snapshot, &metas)); + PAIMON_ASSIGN_OR_RAISE(std::vector manifest_results, + ReadManifests(metas)); + + std::vector unmerged_entries; + for (ManifestReadResult& result : manifest_results) { + AddPath(path_factory_->ToManifestFilePath(result.meta.FileName())); + unmerged_entries.insert(unmerged_entries.end(), + std::make_move_iterator(result.entries.begin()), + std::make_move_iterator(result.entries.end())); + } + + std::vector live_entries; + PAIMON_RETURN_NOT_OK(FileStoreScan::MergeLiveEntries(unmerged_entries, &live_entries)); + return CollectEntryFiles(live_entries, /*only_add=*/false); + } + + Status CollectChangelogFiles(const Snapshot& snapshot) { + if (!snapshot.ChangelogManifestList()) { + return Status::OK(); + } + std::vector metas; + PAIMON_RETURN_NOT_OK(manifest_list_->ReadChangelogManifests(snapshot, &metas)); + PAIMON_ASSIGN_OR_RAISE(std::vector manifest_results, + ReadManifests(metas)); + + for (ManifestReadResult& result : manifest_results) { + AddPath(path_factory_->ToManifestFilePath(result.meta.FileName())); + PAIMON_RETURN_NOT_OK(CollectEntryFiles(result.entries, /*only_add=*/true)); + } + return Status::OK(); + } + + Status CollectEntryFiles(const std::vector& entries, bool only_add) { + for (const ManifestEntry& entry : entries) { + if (only_add && !(entry.Kind() == FileKind::Add())) { + continue; + } + schema_ids_.insert(entry.File()->schema_id); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr data_file_path_factory, + path_factory_->CreateDataFilePathFactory(entry.Partition(), entry.Bucket())); + for (const std::string& path : data_file_path_factory->CollectFiles(entry.File())) { + AddPath(path); + } + } + return Status::OK(); + } + + Status CollectIndexFiles(const Snapshot& snapshot) { + if (!snapshot.IndexManifest()) { + return Status::OK(); + } + auto filter = [this](const IndexManifestEntry& entry) -> Result { + if (!(entry.kind == FileKind::Add()) || + (bucket_id_ && entry.bucket != bucket_id_.value())) { + return false; + } + if (!partition_filter_) { + return true; + } + return partition_filter_->Test(partition_schema_, entry.partition); + }; + PAIMON_ASSIGN_OR_RAISE(std::vector entries, + index_file_handler_->Scan(snapshot, filter)); + for (const IndexManifestEntry& entry : entries) { + PAIMON_ASSIGN_OR_RAISE( + std::string path, + index_file_handler_->FilePath(entry.partition, entry.bucket, entry.index_file)); + AddPath(path); + } + return Status::OK(); + } + + private: + std::shared_ptr path_factory_; + std::shared_ptr manifest_list_; + std::shared_ptr manifest_file_; + std::unique_ptr index_file_handler_; + std::shared_ptr schema_manager_; + int64_t latest_schema_id_; + std::shared_ptr snapshot_manager_; + std::shared_ptr executor_; + std::shared_ptr partition_schema_; + std::shared_ptr partition_filter_; + std::optional bucket_id_; + std::set schema_ids_; + std::set paths_; +}; + +} // namespace + +Result> SnapshotFileScan::ListFiles( + const std::string& table_path, const std::string& branch, + const std::optional& snapshot_id, const std::shared_ptr& scan_filter, + const std::map& options, + const std::shared_ptr& file_system, const std::shared_ptr& executor, + const std::shared_ptr& memory_pool) { + if (table_path.empty()) { + return Status::Invalid("table path is empty"); + } + if (snapshot_id && snapshot_id.value() < Snapshot::FIRST_SNAPSHOT_ID) { + return Status::Invalid("snapshot id must be greater than or equal to 1"); + } + if (scan_filter && scan_filter->GetPredicate()) { + return Status::Invalid("snapshot file scan does not support predicate filter"); + } + + std::shared_ptr pool = memory_pool ? memory_pool : GetDefaultPool(); + std::shared_ptr final_executor = executor; + if (!final_executor) { + final_executor = CreateDefaultExecutor(); + } + PAIMON_ASSIGN_OR_RAISE(CoreOptions temporary_options, + CoreOptions::FromMap(options, file_system)); + std::string normalized_branch = BranchManager::NormalizeBranch(branch); + auto snapshot_manager = std::make_shared(temporary_options.GetFileSystem(), + table_path, normalized_branch); + + std::optional snapshot; + if (snapshot_id) { + PAIMON_ASSIGN_OR_RAISE(Snapshot loaded_snapshot, + snapshot_manager->LoadSnapshot(snapshot_id.value())); + snapshot = std::move(loaded_snapshot); + } else { + PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager->LatestSnapshot()); + } + if (!snapshot) { + return std::set(); + } + + auto schema_manager = std::make_shared(temporary_options.GetFileSystem(), + table_path, normalized_branch); + PAIMON_ASSIGN_OR_RAISE(std::optional> latest_schema, + schema_manager->Latest()); + if (!latest_schema) { + return Status::Invalid("not found latest schema"); + } + + std::map final_options = latest_schema.value()->Options(); + for (const auto& [key, value] : options) { + final_options[key] = value; + } + final_options[Options::BRANCH] = normalized_branch; + PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, + CoreOptions::FromMap(final_options, file_system)); + + std::shared_ptr arrow_schema = + DataField::ConvertDataFieldsToArrowSchema(latest_schema.value()->Fields()); + PAIMON_ASSIGN_OR_RAISE(std::vector external_paths, + core_options.CreateExternalPaths()); + PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, + core_options.CreateGlobalIndexExternalPath()); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr path_factory, + FileStorePathFactory::Create( + table_path, arrow_schema, latest_schema.value()->PartitionKeys(), + core_options.GetPartitionDefaultName(), core_options.GetFileFormat()->Identifier(), + core_options.DataFilePrefix(), core_options.LegacyPartitionNameEnabled(), + external_paths, global_index_external_path, core_options.IndexFileInDataFileDir(), + pool)); + + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr manifest_list, + ManifestList::Create(core_options.GetFileSystem(), core_options.GetManifestFormat(), + core_options.GetManifestCompression(), path_factory, + core_options.GetCache(), pool)); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr partition_schema, + FieldMapping::GetPartitionSchema(arrow_schema, latest_schema.value()->PartitionKeys())); + std::shared_ptr partition_filter; + std::optional bucket_id; + if (scan_filter) { + bucket_id = scan_filter->GetBucketFilter(); + PAIMON_ASSIGN_OR_RAISE( + partition_filter, + FileStoreScan::CreatePartitionPredicate( + latest_schema.value()->PartitionKeys(), core_options.GetPartitionDefaultName(), + arrow_schema, scan_filter->GetPartitionFilters())); + } + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr manifest_file, + ManifestFile::Create(core_options.GetFileSystem(), core_options.GetManifestFormat(), + core_options.GetManifestCompression(), path_factory, + core_options.GetManifestTargetFileSize(), pool, core_options, + partition_schema)); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr index_manifest_file, + IndexManifestFile::Create(core_options.GetFileSystem(), core_options.GetManifestFormat(), + core_options.GetManifestCompression(), path_factory, + core_options.GetBucket(), pool, core_options)); + auto index_file_handler = std::make_unique( + core_options.GetFileSystem(), std::move(index_manifest_file), + std::make_shared(path_factory), + core_options.DeletionVectorsBitmap64(), pool); + + SnapshotFileCollector collector(path_factory, manifest_list, manifest_file, + std::move(index_file_handler), schema_manager, + latest_schema.value()->Id(), snapshot_manager, final_executor, + partition_schema, partition_filter, bucket_id); + return collector.Collect(snapshot.value()); +} + +} // namespace paimon diff --git a/src/paimon/core/snapshot_file_scan_test.cpp b/src/paimon/core/snapshot_file_scan_test.cpp new file mode 100644 index 000000000..751b39aaa --- /dev/null +++ b/src/paimon/core/snapshot_file_scan_test.cpp @@ -0,0 +1,317 @@ +/* + * 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/snapshot/snapshot_file_scan.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/defs.h" +#include "paimon/predicate/predicate_builder.h" +#include "paimon/scan_context.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { +namespace { + +Result> ListFiles(const std::string& table_path, + const std::optional& snapshot_id = std::nullopt, + const std::shared_ptr& scan_filter = nullptr, + const std::string& branch = "main") { + return SnapshotFileScan::ListFiles(table_path, branch, snapshot_id, scan_filter, /*options=*/{}, + /*file_system=*/nullptr, /*executor=*/nullptr, + /*memory_pool=*/nullptr); +} + +std::shared_ptr CreateFilter( + const std::vector>& partition_filters, + const std::optional& bucket_filter = std::nullopt, + const std::shared_ptr& predicate = nullptr) { + return std::make_shared(predicate, partition_filters, bucket_filter); +} + +std::set ExpectedFiles(const std::string& table_path, + std::initializer_list relative_paths) { + std::set paths; + for (const std::string& relative_path : relative_paths) { + paths.insert(table_path + "/" + relative_path); + } + return paths; +} + +} // namespace + +TEST(SnapshotFileScanTest, TestLatestSnapshotAndBucketFilter) { + std::string table_path = GetDataDir() + "/orc/append_09.db/append_09"; + + ASSERT_OK_AND_ASSIGN(std::set all_files, ListFiles(table_path)); + ASSERT_EQ( + ExpectedFiles(table_path, {"f1=10/bucket-0/data-d41fd7d1-b3e4-4905-aad9-b20a780e90a2-0.orc", + "f1=10/bucket-1/data-b9e7c41f-66e8-4dad-b25a-e6e1963becc4-0.orc", + "f1=20/bucket-0/data-b913a160-a4d1-4084-af2a-18333c35668e-0.orc", + "f1=20/bucket-0/data-db2b44c0-0d73-449d-82a0-4075bd2cb6e3-0.orc", + "manifest/manifest-3a44a0da-1008-463c-914e-28d271375e24-0", + "manifest/manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-0", + "manifest/manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-1", + "manifest/manifest-c5904353-0236-46a2-891f-62a326dd8e5e-0", + "manifest/manifest-f8b15cfc-437a-4d21-a6a0-e45b639ae7ed-0", + "manifest/manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-2", + "manifest/manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-3", + "schema/schema-0", "snapshot/snapshot-5"}), + all_files); + + std::shared_ptr bucket_zero_filter = + CreateFilter(/*partition_filters=*/{}, /*bucket_filter=*/0); + ASSERT_OK_AND_ASSIGN(std::set bucket_zero_files, + ListFiles(table_path, std::nullopt, bucket_zero_filter)); + ASSERT_EQ( + ExpectedFiles(table_path, {"f1=10/bucket-0/data-d41fd7d1-b3e4-4905-aad9-b20a780e90a2-0.orc", + "f1=20/bucket-0/data-b913a160-a4d1-4084-af2a-18333c35668e-0.orc", + "f1=20/bucket-0/data-db2b44c0-0d73-449d-82a0-4075bd2cb6e3-0.orc", + "manifest/manifest-3a44a0da-1008-463c-914e-28d271375e24-0", + "manifest/manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-0", + "manifest/manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-1", + "manifest/manifest-c5904353-0236-46a2-891f-62a326dd8e5e-0", + "manifest/manifest-f8b15cfc-437a-4d21-a6a0-e45b639ae7ed-0", + "manifest/manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-2", + "manifest/manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-3", + "schema/schema-0", "snapshot/snapshot-5"}), + bucket_zero_files); + + std::shared_ptr bucket_one_filter = + CreateFilter(/*partition_filters=*/{}, /*bucket_filter=*/1); + ASSERT_OK_AND_ASSIGN(std::set bucket_one_files, + ListFiles(table_path, std::nullopt, bucket_one_filter)); + ASSERT_EQ( + ExpectedFiles(table_path, {"f1=10/bucket-1/data-b9e7c41f-66e8-4dad-b25a-e6e1963becc4-0.orc", + "manifest/manifest-3a44a0da-1008-463c-914e-28d271375e24-0", + "manifest/manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-0", + "manifest/manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-1", + "manifest/manifest-c5904353-0236-46a2-891f-62a326dd8e5e-0", + "manifest/manifest-f8b15cfc-437a-4d21-a6a0-e45b639ae7ed-0", + "manifest/manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-2", + "manifest/manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-3", + "schema/schema-0", "snapshot/snapshot-5"}), + bucket_one_files); +} + +TEST(SnapshotFileScanTest, TestExplicitSnapshot) { + std::string table_path = GetDataDir() + "/orc/append_09.db/append_09"; + + ASSERT_OK_AND_ASSIGN(std::set files, ListFiles(table_path, /*snapshot_id=*/1)); + ASSERT_EQ( + ExpectedFiles(table_path, {"f1=10/bucket-0/data-d41fd7d1-b3e4-4905-aad9-b20a780e90a2-0.orc", + "f1=10/bucket-1/data-4e30d6c0-f109-4300-a010-4ba03047dd9d-0.orc", + "f1=20/bucket-0/data-db2b44c0-0d73-449d-82a0-4075bd2cb6e3-0.orc", + "manifest/manifest-f8b15cfc-437a-4d21-a6a0-e45b639ae7ed-0", + "manifest/manifest-list-616d1847-a02c-495f-9cca-2c8b7def0fec-0", + "manifest/manifest-list-616d1847-a02c-495f-9cca-2c8b7def0fec-1", + "schema/schema-0", "snapshot/snapshot-1"}), + files); +} + +TEST(SnapshotFileScanTest, TestExplicitSnapshotIncludesLatestSchema) { + std::string table_path = + GetDataDir() + "/orc/append_table_with_alter_table.db/append_table_with_alter_table"; + + ASSERT_OK_AND_ASSIGN(std::set files, ListFiles(table_path, /*snapshot_id=*/1)); + ASSERT_EQ( + ExpectedFiles(table_path, + {"key0=0/key1=1/bucket-0/data-2190cec3-ce87-4175-8d19-9268becf4440-0.orc", + "key0=1/key1=1/bucket-0/data-492ed5ab-4740-4e93-8a0a-79a6893b1770-0.orc", + "manifest/manifest-f2299c3d-c3f1-400f-ad3d-124e3a342389-0", + "manifest/manifest-list-83964df9-8a98-4f91-a4e9-05f3e07be3f9-0", + "manifest/manifest-list-83964df9-8a98-4f91-a4e9-05f3e07be3f9-1", + "schema/schema-0", "schema/schema-1", "snapshot/snapshot-1"}), + files); +} + +TEST(SnapshotFileScanTest, TestExternalFileIndex) { + std::string table_path = + GetDataDir() + "/orc/append_with_bloomfilter.db/append_with_bloomfilter"; + + ASSERT_OK_AND_ASSIGN(std::set files, ListFiles(table_path)); + ASSERT_EQ( + ExpectedFiles(table_path, {"bucket-0/data-34e8acb2-110b-4c32-9dc6-d6435178d0ad-0.orc", + "bucket-0/data-34e8acb2-110b-4c32-9dc6-d6435178d0ad-0.orc.index", + "manifest/manifest-7a45ff6a-e225-4f4c-8141-dc5c88a2a81f-0", + "manifest/manifest-list-13cc1371-7074-42aa-83c4-94ce8b2819c3-0", + "manifest/manifest-list-13cc1371-7074-42aa-83c4-94ce8b2819c3-1", + "schema/schema-0", "snapshot/snapshot-1"}), + files); +} + +TEST(SnapshotFileScanTest, TestExternalDataPath) { + std::string table_path = GetDataDir() + + "/orc/pk_dv_index_not_in_data_with_external.db/" + "pk_dv_index_not_in_data_with_external"; + std::shared_ptr scan_filter = + CreateFilter(/*partition_filters=*/{{{"f1", "20"}}}, /*bucket_filter=*/0); + + ASSERT_OK_AND_ASSIGN(std::set files, + ListFiles(table_path, /*snapshot_id=*/1, scan_filter)); + const std::string external_file_suffix = + "/external/f1=20/bucket-0/data-8b1ebe94-3177-4805-b74b-ae5e1bb1086f-0.orc"; + auto external_file = + std::find_if(files.begin(), files.end(), [&external_file_suffix](const std::string& path) { + return StringUtils::EndsWith(path, external_file_suffix); + }); + ASSERT_NE(files.end(), external_file); + files.erase(external_file); + ASSERT_EQ( + ExpectedFiles(table_path, {"manifest/manifest-0dff6454-a796-469a-8452-81a7601d1b34-0", + "manifest/manifest-list-0ef45e5c-7b8b-42f6-b3a1-6d16bc9f522a-0", + "manifest/manifest-list-0ef45e5c-7b8b-42f6-b3a1-6d16bc9f522a-1", + "schema/schema-0", "snapshot/snapshot-1"}), + files); +} + +TEST(SnapshotFileScanTest, TestPartitionAndBucketFilter) { + std::string table_path = GetDataDir() + "/orc/append_09.db/append_09"; + std::shared_ptr scan_filter = + CreateFilter(/*partition_filters=*/{{{"f1", "10"}}}, /*bucket_filter=*/0); + + ASSERT_OK_AND_ASSIGN(std::set files, + ListFiles(table_path, std::nullopt, scan_filter)); + ASSERT_EQ( + ExpectedFiles(table_path, {"f1=10/bucket-0/data-d41fd7d1-b3e4-4905-aad9-b20a780e90a2-0.orc", + "manifest/manifest-3a44a0da-1008-463c-914e-28d271375e24-0", + "manifest/manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-0", + "manifest/manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-1", + "manifest/manifest-c5904353-0236-46a2-891f-62a326dd8e5e-0", + "manifest/manifest-f8b15cfc-437a-4d21-a6a0-e45b639ae7ed-0", + "manifest/manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-2", + "manifest/manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-3", + "schema/schema-0", "snapshot/snapshot-5"}), + files); +} + +TEST(SnapshotFileScanTest, TestPartitionFilterWithoutMatchingEntries) { + std::string table_path = GetDataDir() + "/orc/append_09.db/append_09"; + std::shared_ptr scan_filter = CreateFilter(/*partition_filters=*/{{{"f1", "999"}}}); + + ASSERT_OK_AND_ASSIGN(std::set files, + ListFiles(table_path, std::nullopt, scan_filter)); + ASSERT_EQ( + ExpectedFiles(table_path, {"manifest/manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-2", + "manifest/manifest-list-f2d59cb8-3ec6-4860-b34b-050b1a533416-3", + "schema/schema-0", "snapshot/snapshot-5"}), + files); +} + +TEST(SnapshotFileScanTest, TestDeletionVectorIndexBucketFilter) { + std::string table_path = GetDataDir() + + "/orc/pk_table_with_dv_cardinality.db/" + "pk_table_with_dv_cardinality"; + + std::shared_ptr bucket_zero_filter = + CreateFilter(/*partition_filters=*/{}, /*bucket_filter=*/0); + ASSERT_OK_AND_ASSIGN(std::set bucket_zero_files, + ListFiles(table_path, /*snapshot_id=*/4, bucket_zero_filter)); + ASSERT_EQ( + ExpectedFiles(table_path, {"f1=10/bucket-0/data-0d0f29cc-63c6-4fab-a594-71bd7d06fcde-0.orc", + "f1=10/bucket-0/data-0d0f29cc-63c6-4fab-a594-71bd7d06fcde-1.orc", + "index/index-86356766-3238-46e6-990b-656cd7409eaa-0", + "manifest/index-manifest-59bcb792-830f-4b57-b838-cfcc50d29266-0", + "manifest/manifest-1d9dbf4f-667b-4a05-ad5a-f7fed8908aa4-0", + "manifest/manifest-1d9dbf4f-667b-4a05-ad5a-f7fed8908aa4-1", + "manifest/manifest-1d9dbf4f-667b-4a05-ad5a-f7fed8908aa4-2", + "manifest/manifest-1d9dbf4f-667b-4a05-ad5a-f7fed8908aa4-3", + "manifest/manifest-list-540cf68e-698e-4b66-af30-eb558d8db43d-6", + "manifest/manifest-list-540cf68e-698e-4b66-af30-eb558d8db43d-7", + "schema/schema-0", "snapshot/snapshot-4"}), + bucket_zero_files); + + std::shared_ptr bucket_one_filter = + CreateFilter(/*partition_filters=*/{}, /*bucket_filter=*/1); + ASSERT_OK_AND_ASSIGN(std::set bucket_one_files, + ListFiles(table_path, /*snapshot_id=*/4, bucket_one_filter)); + ASSERT_EQ( + ExpectedFiles(table_path, {"f1=10/bucket-1/data-2ffe7ae9-2cf7-41e9-944b-2065585cde31-0.orc", + "index/index-86356766-3238-46e6-990b-656cd7409eaa-1", + "manifest/index-manifest-59bcb792-830f-4b57-b838-cfcc50d29266-0", + "manifest/manifest-1d9dbf4f-667b-4a05-ad5a-f7fed8908aa4-0", + "manifest/manifest-1d9dbf4f-667b-4a05-ad5a-f7fed8908aa4-1", + "manifest/manifest-1d9dbf4f-667b-4a05-ad5a-f7fed8908aa4-2", + "manifest/manifest-1d9dbf4f-667b-4a05-ad5a-f7fed8908aa4-3", + "manifest/manifest-list-540cf68e-698e-4b66-af30-eb558d8db43d-6", + "manifest/manifest-list-540cf68e-698e-4b66-af30-eb558d8db43d-7", + "schema/schema-0", "snapshot/snapshot-4"}), + bucket_one_files); +} + +TEST(SnapshotFileScanTest, TestGlobalIndex) { + std::string table_path = + GetDataDir() + "/orc/append_with_global_index.db/append_with_global_index"; + + ASSERT_OK_AND_ASSIGN(std::set files, ListFiles(table_path, /*snapshot_id=*/4)); + ASSERT_EQ(ExpectedFiles(table_path, + {"bucket-0/data-2430f01c-b947-48dc-82a8-7c60aaa348e4-0.orc", + "index/bitmap-global-index-0c950cb6-e6e9-46cd-a9a9-cfcd55f870d3.index", + "index/bitmap-global-index-21ac35d9-200a-489d-b649-ec241f832345.index", + "index/bitmap-global-index-8b1bed37-31c5-4288-8c46-3d0d30fbd302.index", + "manifest/index-manifest-795a9d8a-60bf-401d-ab02-6f13f8ca1098-0", + "manifest/manifest-65b0d403-a1bc-4157-b242-bff73c46596d-0", + "manifest/manifest-list-2bccccf8-9f5e-48f2-b706-5b33f8c3bfc0-0", + "manifest/manifest-list-2bccccf8-9f5e-48f2-b706-5b33f8c3bfc0-1", + "schema/schema-0", "snapshot/snapshot-4"}), + files); +} + +TEST(SnapshotFileScanTest, TestBranch) { + std::string table_path = GetDataDir() + + "/orc/append_table_with_append_pt_branch.db/" + "append_table_with_append_pt_branch"; + + ASSERT_OK_AND_ASSIGN(std::set files, + ListFiles(table_path, std::nullopt, /*scan_filter=*/nullptr, "test")); + ASSERT_EQ( + ExpectedFiles(table_path, + {"branch/branch-test/schema/schema-0", "branch/branch-test/schema/schema-1", + "branch/branch-test/snapshot/snapshot-2", + "manifest/manifest-4e72f3a9-4ad2-4ce3-a387-febef513ee24-0", + "manifest/manifest-529f38e6-ae67-4a61-83fa-ede384f720e4-0", + "manifest/manifest-list-10f6b65d-7580-42b7-9c56-c9c46bf2a3bd-0", + "manifest/manifest-list-10f6b65d-7580-42b7-9c56-c9c46bf2a3bd-1", + "pt=2/bucket-0/data-150a7c45-2972-4d8c-983e-a1ab2e382e6a-0.orc", + "pt=2/bucket-0/data-ea107991-e78c-445a-b225-7af07bcdf8c3-0.orc"}), + files); +} + +TEST(SnapshotFileScanTest, TestInvalidArguments) { + ASSERT_NOK_WITH_MSG(ListFiles(""), "table path is empty"); + ASSERT_NOK_WITH_MSG(ListFiles("unused", /*snapshot_id=*/0), + "snapshot id must be greater than or equal to 1"); + std::shared_ptr predicate = + PredicateBuilder::IsNull(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING); + std::shared_ptr scan_filter = + CreateFilter(/*partition_filters=*/{}, /*bucket_filter=*/std::nullopt, predicate); + ASSERT_NOK_WITH_MSG(ListFiles("unused", std::nullopt, scan_filter), + "snapshot file scan does not support predicate filter"); +} + +} // namespace paimon::test