| /* | |
| * Licensed to the Apache Software Foundation (ASF) under one | |
| * or more contributor license agreements. See the NOTICE file | |
| * distributed with this work for additional information | |
| * regarding copyright ownership. The ASF licenses this file | |
| * to you under the Apache License, Version 2.0 (the | |
| * "License"); you may not use this file except in compliance | |
| * with the License. You may obtain a copy of the License at | |
| * | |
| * http://www.apache.org/licenses/LICENSE-2.0 | |
| * | |
| * Unless required by applicable law or agreed to in writing, | |
| * software distributed under the License is distributed on an | |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | |
| * KIND, either express or implied. See the License for the | |
| * specific language governing permissions and limitations | |
| * under the License. | |
| */ | |
| #include "paimon/table/source/table_scan.h" | |
| #include <map> | |
| #include <memory> | |
| #include <optional> | |
| #include <string> | |
| #include <utility> | |
| #include <vector> | |
| #include "fmt/format.h" | |
| #include "paimon/common/metrics/metrics_impl.h" | |
| #include "paimon/common/predicate/predicate_validator.h" | |
| #include "paimon/common/types/data_field.h" | |
| #include "paimon/common/utils/fields_comparator.h" | |
| #include "paimon/common/utils/options_utils.h" | |
| #include "paimon/core/core_options.h" | |
| #include "paimon/core/index/index_file_handler.h" | |
| #include "paimon/core/index/pk/primary_key_index_definitions.h" | |
| #include "paimon/core/manifest/index_manifest_file.h" | |
| #include "paimon/core/manifest/manifest_file.h" | |
| #include "paimon/core/manifest/manifest_list.h" | |
| #include "paimon/core/operation/append_only_file_store_scan.h" | |
| #include "paimon/core/operation/data_evolution_file_store_scan.h" | |
| #include "paimon/core/operation/file_store_scan.h" | |
| #include "paimon/core/operation/key_value_file_store_scan.h" | |
| #include "paimon/core/realtime/realtime_context_impl.h" | |
| #include "paimon/core/schema/schema_manager.h" | |
| #include "paimon/core/schema/schema_validation.h" | |
| #include "paimon/core/schema/table_schema.h" | |
| #include "paimon/core/table/bucket_mode.h" | |
| #include "paimon/core/table/format/format_table_loader.h" | |
| #include "paimon/core/table/format/format_table_scan.h" | |
| #include "paimon/core/table/source/abstract_table_scan.h" | |
| #include "paimon/core/table/source/append_only_split_generator.h" | |
| #include "paimon/core/table/source/data_evolution_batch_scan.h" | |
| #include "paimon/core/table/source/data_evolution_split_generator.h" | |
| #include "paimon/core/table/source/data_table_batch_scan.h" | |
| #include "paimon/core/table/source/data_table_stream_scan.h" | |
| #include "paimon/core/table/source/merge_tree_split_generator.h" | |
| #include "paimon/core/table/source/primary_key_index_batch_scan.h" | |
| #include "paimon/core/table/source/read_optimized_scan_options.h" | |
| #include "paimon/core/table/source/realtime_table_scan.h" | |
| #include "paimon/core/table/source/snapshot/snapshot_reader.h" | |
| #include "paimon/core/table/source/split_generator.h" | |
| #include "paimon/core/table/system/system_table.h" | |
| #include "paimon/core/utils/branch_manager.h" | |
| #include "paimon/core/utils/field_mapping.h" | |
| #include "paimon/core/utils/file_store_path_factory.h" | |
| #include "paimon/core/utils/index_file_path_factories.h" | |
| #include "paimon/core/utils/primary_key_table_utils.h" | |
| #include "paimon/core/utils/snapshot_manager.h" | |
| #include "paimon/format/file_format.h" | |
| #include "paimon/realtime/realtime_context.h" | |
| #include "paimon/result.h" | |
| #include "paimon/scan_context.h" | |
| #include "paimon/status.h" | |
| #include "paimon/table/format/format_table.h" | |
| namespace arrow { | |
| class Schema; | |
| } // namespace arrow | |
| namespace paimon { | |
| class Executor; | |
| class MemoryPool; | |
| namespace { | |
| class TableScanImpl { | |
| public: | |
| static Result<std::unique_ptr<FileStoreScan>> CreateFileStoreScan( | |
| const std::shared_ptr<FileStorePathFactory>& path_factory, | |
| const std::shared_ptr<arrow::Schema>& arrow_schema, | |
| const std::shared_ptr<TableSchema>& table_schema, const CoreOptions& core_options, | |
| const std::shared_ptr<Executor>& executor, const std::shared_ptr<MemoryPool>& memory_pool, | |
| const ScanContext* context) { | |
| auto fs = core_options.GetFileSystem(); | |
| auto manifest_file_format = core_options.GetManifestFormat(); | |
| std::string branch = BranchManager::NormalizeBranch(core_options.GetBranch()); | |
| auto snapshot_manager = std::make_shared<SnapshotManager>(fs, context->GetPath(), branch); | |
| // TODO(liancheng.lsz): support fallback branch in scan | |
| auto schema_manager = std::make_shared<SchemaManager>(fs, context->GetPath(), branch); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<ManifestList> manifest_list, | |
| ManifestList::Create(fs, manifest_file_format, core_options.GetManifestCompression(), | |
| path_factory, core_options.GetCache(), memory_pool)); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<arrow::Schema> partition_schema, | |
| FieldMapping::GetPartitionSchema(arrow_schema, table_schema->PartitionKeys())); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<ManifestFile> manifest_file, | |
| ManifestFile::Create(fs, manifest_file_format, core_options.GetManifestCompression(), | |
| path_factory, core_options.GetManifestTargetFileSize(), | |
| memory_pool, core_options, partition_schema)); | |
| std::unique_ptr<FileStoreScan> scan; | |
| if (table_schema->PrimaryKeys().empty()) { | |
| if (core_options.DataEvolutionEnabled()) { | |
| PAIMON_ASSIGN_OR_RAISE( | |
| scan, DataEvolutionFileStoreScan::Create( | |
| snapshot_manager, schema_manager, manifest_list, manifest_file, | |
| table_schema, arrow_schema, context->GetScanFilters(), core_options, | |
| executor, memory_pool)); | |
| } else { | |
| PAIMON_ASSIGN_OR_RAISE( | |
| scan, AppendOnlyFileStoreScan::Create( | |
| snapshot_manager, schema_manager, manifest_list, manifest_file, | |
| table_schema, arrow_schema, context->GetScanFilters(), core_options, | |
| executor, memory_pool)); | |
| } | |
| } else { | |
| PAIMON_ASSIGN_OR_RAISE( | |
| scan, KeyValueFileStoreScan::Create(snapshot_manager, schema_manager, manifest_list, | |
| manifest_file, table_schema, arrow_schema, | |
| context->GetScanFilters(), core_options, | |
| executor, memory_pool)); | |
| } | |
| return WithTablePath(std::move(scan), context); | |
| } | |
| static std::unique_ptr<FileStoreScan> WithTablePath(std::unique_ptr<FileStoreScan>&& scan, | |
| const ScanContext* context) { | |
| scan->WithTablePath(context->GetPath()); | |
| return std::move(scan); | |
| } | |
| static Result<std::unique_ptr<SplitGenerator>> CreateSplitGenerator( | |
| const std::shared_ptr<TableSchema>& table_schema, const CoreOptions& core_options, | |
| const ScanContext* context) { | |
| auto source_split_target_size = core_options.GetSourceSplitTargetSize(); | |
| auto source_split_open_file_cost = core_options.GetSourceSplitOpenFileCost(); | |
| if (table_schema->PrimaryKeys().empty()) { | |
| if (core_options.DataEvolutionEnabled()) { | |
| return std::make_unique<DataEvolutionSplitGenerator>( | |
| source_split_target_size, source_split_open_file_cost, | |
| core_options.BlobSplitByFileSize()); | |
| } | |
| BucketMode bucket_mode = (core_options.GetBucket() == -1 ? BucketMode::BUCKET_UNAWARE | |
| : BucketMode::HASH_FIXED); | |
| return std::make_unique<AppendOnlySplitGenerator>( | |
| source_split_target_size, source_split_open_file_cost, bucket_mode); | |
| } else { | |
| // TODO(liancheng.lsz): support evolution | |
| PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> trimmed_primary_keys, | |
| table_schema->TrimmedPrimaryKeys()); | |
| PAIMON_ASSIGN_OR_RAISE(std::vector<DataField> trimmed_pk_fields, | |
| table_schema->GetFields(trimmed_primary_keys)); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<FieldsComparator> key_comparator, | |
| FieldsComparator::Create(trimmed_pk_fields, /*is_ascending_order=*/true)); | |
| return std::make_unique<MergeTreeSplitGenerator>( | |
| source_split_target_size, source_split_open_file_cost, | |
| core_options.DeletionVectorsEnabled(), core_options.GetMergeEngine(), | |
| key_comparator); | |
| } | |
| } | |
| static Result<std::unique_ptr<IndexFileHandler>> CreateIndexFileHandler( | |
| const CoreOptions& core_options, const std::shared_ptr<FileStorePathFactory>& path_factory, | |
| const std::shared_ptr<MemoryPool>& memory_pool) { | |
| PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<IndexManifestFile> index_manifest_file, | |
| IndexManifestFile::Create( | |
| core_options.GetFileSystem(), core_options.GetManifestFormat(), | |
| core_options.GetManifestCompression(), path_factory, | |
| core_options.GetBucket(), memory_pool, core_options)); | |
| return std::make_unique<IndexFileHandler>( | |
| core_options.GetFileSystem(), std::move(index_manifest_file), | |
| std::make_shared<IndexFilePathFactories>(path_factory), | |
| core_options.DeletionVectorsBitmap64(), memory_pool); | |
| } | |
| }; | |
| /// @param loaded_schema The table's schema, when the caller already read it, or null when it has | |
| /// to be read here. `TableScan::Create()` reads one to dispatch on its table type. | |
| Result<std::unique_ptr<TableScan>> NewDataTableScan( | |
| const std::shared_ptr<ScanContext>& context, const std::shared_ptr<TableSchema>& loaded_schema); | |
| /// Maps a `ScanContext` onto `FormatTableScan`, which takes one partition filter and a limit and | |
| /// nothing else. | |
| /// | |
| /// A format table has no buckets, so a bucket filter has nothing to act on. A predicate is | |
| /// refused too, which is narrower than Java: `FormatReadBuilder.newScan()` hands the partition | |
| /// half of one to the scan to prune directories with, which this does not do yet. | |
| Result<std::unique_ptr<TableScan>> NewFormatTableScan(const std::shared_ptr<FormatTable>& table, | |
| const std::shared_ptr<ScanContext>& context) { | |
| // Anything a `ScanContext` carries that a format table cannot honour is refused rather than | |
| // silently dropped. | |
| if (context->IsStreamingMode()) { | |
| return Status::NotImplemented( | |
| "a format table has no snapshots, so there is nothing for a streaming scan to follow"); | |
| } | |
| if (context->GetRealtimeContext() != nullptr) { | |
| return Status::NotImplemented( | |
| "a format table has no real-time store to union with what is on disk"); | |
| } | |
| if (context->GetGlobalIndexResult() != nullptr) { | |
| return Status::NotImplemented("a format table carries no global index"); | |
| } | |
| std::map<std::string, std::string> partition_filter; | |
| std::shared_ptr<ScanFilter> filters = context->GetScanFilters(); | |
| if (filters != nullptr) { | |
| if (filters->GetPredicate() != nullptr) { | |
| return Status::NotImplemented( | |
| "a format table scan does not take a predicate: its files carry no statistics to " | |
| "skip by, and pruning by the partition columns a predicate names is not " | |
| "implemented yet; give the partition values as a partition filter instead"); | |
| } | |
| if (filters->GetBucketFilter()) { | |
| return Status::NotImplemented("a format table has no buckets to filter by"); | |
| } | |
| const std::vector<std::map<std::string, std::string>>& partition_filters = | |
| filters->GetPartitionFilters(); | |
| if (partition_filters.size() > 1) { | |
| return Status::NotImplemented("a format table scan takes at most one partition filter"); | |
| } | |
| if (partition_filters.size() == 1) { | |
| partition_filter = partition_filters.front(); | |
| } | |
| } | |
| PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FormatTableScan> scan, | |
| FormatTableScan::Create(table, partition_filter, context->GetLimit())); | |
| return std::unique_ptr<TableScan>(std::move(scan)); | |
| } | |
| } // namespace | |
| std::shared_ptr<Metrics> TableScan::GetMetrics() const { | |
| return std::make_shared<MetricsImpl>(); | |
| } | |
| Result<std::vector<std::map<std::string, std::string>>> TableScan::ListPartitions() const { | |
| // A managed table's partitions live in its manifests, and reading them is a scan of its own | |
| // rather than a listing, so only the format table path answers this for now. | |
| return Status::NotImplemented("this table does not list its partitions"); | |
| } | |
| Result<std::unique_ptr<TableScan>> TableScan::Create(std::unique_ptr<ScanContext> context) { | |
| if (context == nullptr) { | |
| return Status::Invalid("scan context is null pointer"); | |
| } | |
| if (context->GetMemoryPool() == nullptr) { | |
| return Status::Invalid("memory pool is null pointer"); | |
| } | |
| if (context->GetExecutor() == nullptr) { | |
| return Status::Invalid("executor is null pointer"); | |
| } | |
| std::shared_ptr<ScanContext> shared_context = std::move(context); | |
| // A table the caller already loaded says what it is, so nothing is read to find out. | |
| if (shared_context->GetFormatTable() != nullptr) { | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<FormatTable> format_table, | |
| FormatTable::Copy(shared_context->GetFormatTable(), shared_context->GetOptions())); | |
| return NewFormatTableScan(format_table, shared_context); | |
| } | |
| // load schema | |
| PAIMON_ASSIGN_OR_RAISE(CoreOptions tmp_options, | |
| CoreOptions::FromMap(shared_context->GetOptions(), | |
| shared_context->GetSpecificFileSystem(), {})); | |
| PAIMON_ASSIGN_OR_RAISE(std::optional<SystemTablePath> system_table_path, | |
| SystemTableLoader::TryParsePath(shared_context->GetPath())); | |
| if (system_table_path) { | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<SystemTable> system_table, | |
| SystemTableLoader::LoadFromPath(tmp_options.GetFileSystem(), shared_context->GetPath(), | |
| shared_context->GetOptions())); | |
| return system_table->NewScan(shared_context); | |
| } | |
| // A format table is planned by listing directories, so it never reaches the manifest path | |
| // below. One `TableScan` interface serves both, as Java Paimon serves both through one | |
| // `ReadBuilder`. | |
| std::shared_ptr<TableSchema> latest_schema; | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<FormatTable> format_table, | |
| FormatTableLoader::TryLoad(tmp_options.GetFileSystem(), shared_context->GetPath(), | |
| BranchManager::NormalizeBranch(tmp_options.GetBranch()), | |
| shared_context->GetOptions(), | |
| shared_context->GetSpecificTableSchema(), | |
| /*schema_manager=*/nullptr, &latest_schema)); | |
| if (format_table != nullptr) { | |
| return NewFormatTableScan(format_table, shared_context); | |
| } | |
| // With the schema the dispatch already read, so the managed path does not read it again. | |
| return NewDataTableScan(shared_context, latest_schema); | |
| } | |
| namespace { | |
| Status ValidateRealtimeScan(const TableSchema& table_schema, const CoreOptions& core_options, | |
| const ScanContext& context, bool read_optimized) { | |
| if (!context.GetRealtimeContext()) { | |
| return Status::OK(); | |
| } | |
| if (!core_options.RealtimeEnabled()) { | |
| return Status::Invalid("real-time scan requires realtime.enabled=true"); | |
| } | |
| if (core_options.GetBucket() <= 0) { | |
| return Status::Invalid("real-time union read requires fixed bucket mode"); | |
| } | |
| if (core_options.DataEvolutionEnabled()) { | |
| return Status::Invalid("real-time union read does not support data evolution"); | |
| } | |
| if (!table_schema.PrimaryKeys().empty()) { | |
| if (read_optimized) { | |
| return Status::NotImplemented( | |
| "PK real-time union read does not support read-optimized scans"); | |
| } | |
| PAIMON_RETURN_NOT_OK( | |
| PrimaryKeyTableUtils::ValidateRealtimeOptions(core_options, table_schema)); | |
| } | |
| if (context.IsStreamingMode()) { | |
| return Status::Invalid("real-time union read currently supports batch scans only"); | |
| } | |
| if (context.GetLimit()) { | |
| return Status::Invalid("real-time union read does not support scan limit pushdown"); | |
| } | |
| if (context.GetGlobalIndexResult()) { | |
| return Status::Invalid("real-time union read does not support global index splits"); | |
| } | |
| StartupMode startup_mode = core_options.GetStartupMode(); | |
| const bool specified_snapshot = | |
| startup_mode == StartupMode::FromSnapshot() && core_options.GetScanSnapshotId().has_value(); | |
| if (!(startup_mode == StartupMode::LatestFull() || startup_mode == StartupMode::Latest() || | |
| specified_snapshot)) { | |
| return Status::Invalid( | |
| "real-time union read requires the latest snapshot or an explicit snapshot id"); | |
| } | |
| return Status::OK(); | |
| } | |
| Result<std::unique_ptr<TableScan>> NewDataTableScan( | |
| const std::shared_ptr<ScanContext>& context, | |
| const std::shared_ptr<TableSchema>& loaded_schema) { | |
| PAIMON_ASSIGN_OR_RAISE( | |
| CoreOptions tmp_options, | |
| CoreOptions::FromMap(context->GetOptions(), context->GetSpecificFileSystem(), {})); | |
| std::string branch = BranchManager::NormalizeBranch(tmp_options.GetBranch()); | |
| std::shared_ptr<TableSchema> table_schema; | |
| const auto& specific_table_schema = context->GetSpecificTableSchema(); | |
| if (loaded_schema != nullptr) { | |
| table_schema = loaded_schema; | |
| } else if (branch == BranchManager::DEFAULT_MAIN_BRANCH && specific_table_schema) { | |
| PAIMON_ASSIGN_OR_RAISE(table_schema, | |
| TableSchema::CreateFromJson(specific_table_schema.value())); | |
| } else { | |
| SchemaManager schema_manager(tmp_options.GetFileSystem(), context->GetPath(), branch); | |
| PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_table_schema, | |
| schema_manager.Latest()); | |
| if (latest_table_schema == std::nullopt) { | |
| return Status::Invalid("not found latest schema"); | |
| } | |
| table_schema = latest_table_schema.value(); | |
| } | |
| PAIMON_ASSIGN_OR_RAISE(bool read_optimized, | |
| OptionsUtils::GetValueFromMap<bool>(context->GetOptions(), | |
| kReadOptimizedScanOption, false)); | |
| // merge options | |
| auto options = table_schema->Options(); | |
| for (const auto& [key, value] : context->GetOptions()) { | |
| options[key] = value; | |
| } | |
| PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, | |
| CoreOptions::FromMap(options, context->GetSpecificFileSystem(), {})); | |
| core_options.WithCache(context->GetCache()); | |
| PAIMON_RETURN_NOT_OK(SchemaValidation::ValidateRealtime(*table_schema, core_options)); | |
| PAIMON_RETURN_NOT_OK( | |
| ValidateRealtimeScan(*table_schema, core_options, *context, read_optimized)); | |
| // validate options | |
| if (core_options.GetBucket() == -1) { | |
| if (!table_schema->PrimaryKeys().empty()) { | |
| return Status::NotImplemented(fmt::format( | |
| "do not support pk table bucket={} in scan process", core_options.GetBucket())); | |
| } | |
| } else if (core_options.GetBucket() < 1 && | |
| !SchemaValidation::IsPostponeBucketTable(*table_schema, core_options.GetBucket())) { | |
| return Status::Invalid( | |
| fmt::format("do not support bucket={} in scan process", core_options.GetBucket())); | |
| } | |
| // validate schema and scan filter | |
| auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields()); | |
| if (context->GetScanFilters() && context->GetScanFilters()->GetPredicate()) { | |
| PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithSchema( | |
| *arrow_schema, context->GetScanFilters()->GetPredicate(), | |
| /*validate_field_idx=*/false)); | |
| } | |
| PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> external_paths, | |
| core_options.CreateExternalPaths()); | |
| PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> global_index_external_path, | |
| core_options.CreateGlobalIndexExternalPath()); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<FileStorePathFactory> path_factory, | |
| FileStorePathFactory::Create( | |
| context->GetPath(), arrow_schema, table_schema->PartitionKeys(), | |
| core_options.GetPartitionDefaultName(), core_options.GetFileFormat()->Identifier(), | |
| core_options.DataFilePrefix(), core_options.LegacyPartitionNameEnabled(), | |
| external_paths, global_index_external_path, core_options.IndexFileInDataFileDir(), | |
| context->GetMemoryPool())); | |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStoreScan> file_store_scan, | |
| TableScanImpl::CreateFileStoreScan( | |
| path_factory, arrow_schema, table_schema, core_options, | |
| context->GetExecutor(), context->GetMemoryPool(), context.get())); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::unique_ptr<SplitGenerator> split_generator, | |
| TableScanImpl::CreateSplitGenerator(table_schema, core_options, context.get())); | |
| PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<IndexFileHandler> index_file_handler, | |
| TableScanImpl::CreateIndexFileHandler(core_options, path_factory, | |
| context->GetMemoryPool())); | |
| auto snapshot_reader = std::make_shared<SnapshotReader>( | |
| file_store_scan, path_factory, std::move(split_generator), std::move(index_file_handler)); | |
| const bool pk_table = !table_schema->PrimaryKeys().empty(); | |
| if (read_optimized && pk_table && context->IsStreamingMode()) { | |
| return Status::NotImplemented( | |
| "read-optimized system table does not support streaming scan for primary key table"); | |
| } | |
| if (context->IsStreamingMode()) { | |
| return std::make_unique<DataTableStreamScan>(core_options, snapshot_reader); | |
| } | |
| auto batch_scan = std::make_unique<DataTableBatchScan>( | |
| /*pk_table=*/pk_table, core_options, snapshot_reader, read_optimized, context->GetLimit(), | |
| /*realtime_pk_scan=*/pk_table && context->GetRealtimeContext() != nullptr); | |
| if (context->GetRealtimeContext()) { | |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RealtimeContextImpl> realtime_context, | |
| RealtimeContextImpl::Cast(context->GetRealtimeContext())); | |
| return std::make_unique<RealtimeTableScan>( | |
| std::move(batch_scan), pk_table, realtime_context, path_factory, | |
| snapshot_reader->GetSnapshotManager(), core_options.GetFileSystem(), | |
| context->GetScanFilters(), core_options.GetRealtimeReadViewTtlMillis()); | |
| } | |
| if (core_options.DataEvolutionEnabled()) { | |
| return std::make_unique<DataEvolutionBatchScan>( | |
| context->GetPath(), snapshot_reader, std::move(batch_scan), table_schema, | |
| context->GetGlobalIndexResult(), core_options, context->GetMemoryPool(), | |
| context->GetExecutor()); | |
| } | |
| if (pk_table && !read_optimized && core_options.GlobalIndexEnabled()) { | |
| PAIMON_ASSIGN_OR_RAISE(PrimaryKeyIndexDefinitions definitions, | |
| PrimaryKeyIndexDefinitions::Create(*table_schema)); | |
| if (!definitions.ScalarDefinitions().empty()) { | |
| return PrimaryKeyIndexBatchScan::Create(snapshot_reader, std::move(batch_scan), | |
| table_schema, path_factory, core_options, | |
| context->GetMemoryPool()); | |
| } | |
| } | |
| return batch_scan; | |
| } | |
| } // namespace | |
| } // namespace paimon |