wangyong9999 commented on code in PR #245: URL: https://github.com/apache/paimon-cpp/pull/245#discussion_r3851241407
########## src/paimon/core/index/pksorted/pk_sorted_index_builder.cpp: ########## @@ -0,0 +1,276 @@ +/* + * 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/index/pksorted/pk_sorted_index_builder.h" + +#include <algorithm> +#include <limits> +#include <map> +#include <string> +#include <utility> + +#include "arrow/api.h" +#include "arrow/array/concatenate.h" +#include "arrow/c/bridge.h" +#include "fmt/format.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/fields_comparator.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/casting/casting_utils.h" +#include "paimon/core/global_index/global_index_file_manager.h" +#include "paimon/core/index/pk/primary_key_index_source_file.h" +#include "paimon/core/index/pk/primary_key_index_source_policy.h" +#include "paimon/core/index/pksorted/pk_sorted_data_file_reader.h" +#include "paimon/core/index/pksorted/pk_sorted_index_file.h" +#include "paimon/core/io/data_file_meta.h" +#include "paimon/core/mergetree/compact/sort_merge_reader_with_min_heap.h" +#include "paimon/core/mergetree/external_sort_buffer.h" +#include "paimon/core/mergetree/in_memory_sort_buffer.h" +#include "paimon/core/mergetree/sort_buffer.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/fs/file_system.h" +#include "paimon/global_index/io/global_index_file_writer.h" +#include "paimon/record_batch.h" + +namespace paimon { +namespace { + +constexpr char kRowIdFieldName[] = "_PK_INDEX_ROW_ID"; + +class TrackingGlobalIndexFileWriter : public GlobalIndexFileWriter { + public: + explicit TrackingGlobalIndexFileWriter(const std::shared_ptr<GlobalIndexFileManager>& delegate) + : delegate_(delegate) {} + + Result<std::string> NewFileName(const std::string& prefix) const override { + PAIMON_ASSIGN_OR_RAISE(std::string file_name, delegate_->NewFileName(prefix)); + created_file_names_.push_back(file_name); + return file_name; + } + + Result<std::unique_ptr<OutputStream>> NewOutputStream( + const std::string& file_name) const override { + return delegate_->NewOutputStream(file_name); + } + + Result<int64_t> GetFileSize(const std::string& file_name) const override { + return delegate_->GetFileSize(file_name); + } + + std::string ToPath(const std::string& file_name) const override { + return delegate_->ToPath(file_name); + } + + void Cleanup(const std::shared_ptr<FileSystem>& fs) const { + for (const std::string& file_name : created_file_names_) { + [[maybe_unused]] Status status = fs->Delete(delegate_->ToPath(file_name)); + } + } + + private: + std::shared_ptr<GlobalIndexFileManager> delegate_; + mutable std::vector<std::string> created_file_names_; +}; + +} // namespace + +Result<std::unique_ptr<PkSortedIndexBuilder>> PkSortedIndexBuilder::Create( + const std::string& root_path, const std::string& branch, const BinaryRow& partition, + int32_t bucket, const std::shared_ptr<TableSchema>& table_schema, + const PrimaryKeyIndexDefinition& definition, + const std::shared_ptr<FileStorePathFactory>& path_factory, const CoreOptions& options, + const std::shared_ptr<IOManager>& io_manager, bool enable_multi_thread_spill, + const std::shared_ptr<Executor>& executor, const std::shared_ptr<MemoryPool>& pool) { + if (definition.GetFamily() != PrimaryKeyIndexDefinition::Family::BTREE) { + return Status::Invalid("PkSortedIndexBuilder only supports BTree definitions."); + } + PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema->GetField(definition.FieldId())); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<PkSortedDataFileReader> data_file_reader, + PkSortedDataFileReader::Create(root_path, table_schema, definition.FieldId(), path_factory, + branch, options, executor, pool)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<IndexPathFactory> index_path_factory, + path_factory->CreateIndexFileFactory(partition, bucket)); + return std::unique_ptr<PkSortedIndexBuilder>(new PkSortedIndexBuilder( + partition, bucket, std::move(field), definition, + std::shared_ptr<PkSortedDataFileReader>(std::move(data_file_reader)), + options.GetFileSystem(), std::shared_ptr<IndexPathFactory>(std::move(index_path_factory)), + options, io_manager, enable_multi_thread_spill, pool)); +} + +Result<std::shared_ptr<IndexFileMeta>> PkSortedIndexBuilder::Build( + const std::vector<std::shared_ptr<DataFileMeta>>& source_files) const { + if (source_files.empty()) { + return Status::Invalid("Cannot build a sorted index for an empty data level."); + } + for (const std::shared_ptr<DataFileMeta>& file : source_files) { + if (file == nullptr) { + return Status::Invalid("A sorted index source file is null."); + } + } + std::vector<std::shared_ptr<DataFileMeta>> ordered_files = source_files; + std::sort( + ordered_files.begin(), ordered_files.end(), + [](const std::shared_ptr<DataFileMeta>& left, const std::shared_ptr<DataFileMeta>& right) { + return left->file_name < right->file_name; + }); + int32_t data_level = ordered_files.front()->level; + std::vector<PrimaryKeyIndexSourceFile> source_metas; + source_metas.reserve(ordered_files.size()); + for (const std::shared_ptr<DataFileMeta>& file : ordered_files) { + if (file == nullptr || file->level != data_level || + !PrimaryKeyIndexSourcePolicy::ShouldRead(*file)) { + return Status::Invalid( + "A sorted index can only cover compacted files from one positive data level."); + } + source_metas.emplace_back(file->file_name, file->row_count); + } + + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FieldsComparator> unique_comparator, + FieldsComparator::Create({field_}, /*is_ascending_order=*/true)); + auto comparator = std::shared_ptr<FieldsComparator>(std::move(unique_comparator)); + DataField row_id_field(std::numeric_limits<int32_t>::max(), Review Comment: Kept to match the Java (value, rowId) sort contract and preserve row-id quota accounting. Removing it is an optimization rather than a correctness fix. ########## src/paimon/core/operation/abstract_file_store_write.cpp: ########## @@ -208,6 +211,29 @@ Result<std::vector<std::shared_ptr<CommitMessage>>> AbstractFileStoreWrite::Prep compact_increment.AddNewIndexFiles({dv_index_file_meta.value()}); } } + if (writer_container.primary_key_index_maintainer) { + Status index_status = + writer_container.primary_key_index_maintainer->PrepareCommit(&increment); + if (!index_status.ok()) { + if (compact_deletion_file) { + const auto& new_index_files = + increment.GetCompactIncrement().NewIndexFiles(); + for (const std::shared_ptr<IndexFileMeta>& index_file : new_index_files) { + if (index_file != nullptr && + index_file->IndexType() == + DeletionVectorsIndexFile::DELETION_VECTORS_INDEX) { + PAIMON_ASSIGN_OR_RAISE( + std::string index_path, + dv_maintainer_factory_->GetIndexFileHandler()->FilePath( + partition, bucket, index_file)); + [[maybe_unused]] Status cleanup_status = + options_.GetFileSystem()->Delete(index_path); + } + } + } + return index_status; Review Comment: Fixed. BTree build failures now keep the drained data transition, clean payloads created by that attempt, and leave the level uncovered for fallback and later rebuild. Structural increment errors still propagate. Added a forced-failure regression test. ########## src/paimon/core/operation/expire_snapshots.cpp: ########## @@ -290,6 +295,50 @@ Status ExpireSnapshots::CleanUnusedManifests(const std::string& manifest_list_na return Status::OK(); } +Status ExpireSnapshots::CleanUnusedIndexManifest(const std::optional<std::string>& index_manifest, + std::set<std::string>* skipping_manifest_set) { + if (!index_manifest || index_manifest->empty() || + skipping_manifest_set->count(index_manifest.value()) > 0) { + return Status::OK(); + } + if (index_manifest_file_ == nullptr) { + return Status::Invalid("index manifest file is null"); + } + + std::vector<IndexManifestEntry> entries; + Status read_status = + index_manifest_file_->ReadIfFileExist(index_manifest.value(), /*filter=*/nullptr, &entries); + if (read_status.IsNotExist()) { + return Status::OK(); + } + PAIMON_RETURN_NOT_OK(read_status); + + std::vector<std::pair<std::string, std::string>> index_files_to_delete; + std::set<std::string> planned_file_names; + for (const IndexManifestEntry& entry : entries) { + const std::string& file_name = entry.index_file->FileName(); + if (skipping_manifest_set->count(file_name) > 0 || + !planned_file_names.insert(file_name).second) { + continue; + } + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<IndexPathFactory> index_path_factory, + path_factory_->CreateIndexFileFactory(entry.partition, entry.bucket)); + index_files_to_delete.emplace_back(file_name, index_path_factory->ToPath(entry.index_file)); + } + + for (const auto& [file_name, file_path] : index_files_to_delete) { + skipping_manifest_set->insert(file_name); + auto delete_status = fs_->Delete(file_path); + // Index payload cleanup is best effort, consistent with data file expiration. + (void)delete_status; + } + + skipping_manifest_set->insert(index_manifest.value()); + index_manifest_file_->DeleteQuietly(index_manifest.value()); Review Comment: This matches Java expiration: payload deletion is best-effort, followed by index-manifest cleanup. Retaining the manifest as a retry journal would change that lifecycle contract, so this remains unchanged. ########## src/paimon/core/index/pksorted/pk_sorted_data_file_reader.cpp: ########## @@ -0,0 +1,173 @@ +/* + * 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/index/pksorted/pk_sorted_data_file_reader.h" + +#include <limits> +#include <map> +#include <optional> +#include <string> +#include <utility> + +#include "arrow/api.h" +#include "arrow/c/bridge.h" +#include "fmt/format.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/io/data_file_meta.h" +#include "paimon/core/operation/internal_read_context.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/read_context.h" +#include "paimon/reader/batch_reader.h" +#include "paimon/reader/file_batch_reader.h" + +namespace paimon { + +Result<std::unique_ptr<PkSortedDataFileReader>> PkSortedDataFileReader::Create( + const std::string& root_path, const std::shared_ptr<TableSchema>& table_schema, + int32_t field_id, const std::shared_ptr<FileStorePathFactory>& path_factory, + const std::string& branch, const CoreOptions& options, + const std::shared_ptr<Executor>& executor, const std::shared_ptr<MemoryPool>& pool) { + std::map<std::string, std::string> read_options = options.ToMap(); + read_options[Options::BRANCH] = branch; + ReadContextBuilder builder(root_path); + builder.SetReadFieldIds({field_id}) + .SetOptions(read_options) + .WithBranch(branch) + .WithFileSystem(options.GetFileSystem()) + .WithExecutor(executor) + .WithMemoryPool(pool) + .EnablePrefetch(false) + .EnablePredicateFilter(false); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> read_context, builder.Finish()); + auto shared_read_context = std::shared_ptr<ReadContext>(std::move(read_context)); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr<InternalReadContext> internal_context, + InternalReadContext::Create(shared_read_context, table_schema, read_options)); + auto shared_internal_context = + std::shared_ptr<InternalReadContext>(std::move(internal_context)); + return std::unique_ptr<PkSortedDataFileReader>( + new PkSortedDataFileReader(path_factory, shared_internal_context, pool, executor)); +} + +PkSortedDataFileReader::PkSortedDataFileReader( + const std::shared_ptr<FileStorePathFactory>& path_factory, + const std::shared_ptr<InternalReadContext>& context, const std::shared_ptr<MemoryPool>& pool, + const std::shared_ptr<Executor>& executor) + : RawFileSplitRead(path_factory, context, pool, executor) {} + +Status PkSortedDataFileReader::ReadFile(const BinaryRow& partition, int32_t bucket, + const std::shared_ptr<DataFileMeta>& file, + const BatchConsumer& consumer) const { + if (file == nullptr) { + return Status::Invalid("Primary-key sorted-index source file is null."); + } + if (file->row_count < 0) { + return Status::Invalid(fmt::format("Source file {} has negative row count {}.", + file->file_name, file->row_count)); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFilePathFactory> data_file_path_factory, + path_factory_->CreateDataFilePathFactory(partition, bucket)); + PAIMON_ASSIGN_OR_RAISE( + std::vector<std::unique_ptr<FileBatchReader>> readers, + CreateRawFileReaders(partition, {file}, raw_read_schema_, /*predicate=*/nullptr, + /*dv_factory=*/{}, /*row_ranges=*/std::nullopt, data_file_path_factory, + /*extra_format_options=*/{})); + if (readers.size() != 1) { + return Status::Invalid( + fmt::format("Expected one physical reader for source file {}, but got {}.", + file->file_name, readers.size())); + } + std::unique_ptr<FileBatchReader> reader = std::move(readers[0]); + ScopeGuard close_guard([&]() { reader->Close(); }); + PAIMON_ASSIGN_OR_RAISE(uint64_t physical_row_count, reader->GetNumberOfRows()); + if (physical_row_count > static_cast<uint64_t>(std::numeric_limits<int64_t>::max()) || + static_cast<int64_t>(physical_row_count) != file->row_count) { + return Status::Invalid(fmt::format( + "Physical row count {} of source file {} does not match metadata row count {}.", + physical_row_count, file->file_name, file->row_count)); + } + + int64_t rows_read = 0; + while (true) { + PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatchWithBitmap batch_with_bitmap, + reader->NextBatchWithBitmap()); + if (BatchReader::IsEofBatch(batch_with_bitmap)) { + break; + } + auto& [batch, bitmap] = batch_with_bitmap; + auto& [c_array, c_schema] = batch; + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array, + arrow::ImportArray(c_array.get(), c_schema.get())); + if (array == nullptr || array->type_id() != arrow::Type::STRUCT) { + return Status::Invalid( + fmt::format("Source file {} did not return a struct batch.", file->file_name)); + } + auto struct_array = checked_pointer_cast<arrow::StructArray>(array); + if (struct_array->num_fields() != 1) { + return Status::Invalid( + fmt::format("Source file {} returned {} fields for a single-column index build.", + file->file_name, struct_array->num_fields())); + } + if (static_cast<int64_t>(bitmap.Cardinality()) != struct_array->length()) { + return Status::Invalid( + fmt::format("Source file {} was filtered while building a physical-row index.", + file->file_name)); + } + std::vector<int64_t> positions; + positions.reserve(static_cast<size_t>(struct_array->length())); + for (int64_t index = 0; index < struct_array->length(); ++index) { + PAIMON_ASSIGN_OR_RAISE(uint64_t physical_position, + reader->GetPreviousBatchFileRowId(static_cast<uint64_t>(index))); + if (physical_position > static_cast<uint64_t>(std::numeric_limits<int64_t>::max()) || + static_cast<int64_t>(physical_position) != rows_read + index) { + return Status::Invalid(fmt::format( + "Source file {} returned non-contiguous physical row position {} at row {}.", + file->file_name, physical_position, rows_read + index)); + } + positions.push_back(static_cast<int64_t>(physical_position)); + } + PAIMON_RETURN_NOT_OK(consumer(struct_array, positions)); Review Comment: Fixed. The reader keeps the contiguous-position validation locally, and the unused position vector and callback parameter are removed. ########## src/paimon/core/index/pksorted/pk_sorted_index_file.cpp: ########## @@ -95,24 +139,95 @@ Result<std::shared_ptr<IndexFileMeta>> PkSortedIndexFile::Build( ScopeGuard array_guard([&]() { ArrowArrayRelease(&c_array); }); PAIMON_RETURN_NOT_OK(writer->AddBatch(&c_array, std::move(sorted_ordinals))); PAIMON_ASSIGN_OR_RAISE(std::vector<GlobalIndexIOMeta> io_metas, writer->Finish()); - if (io_metas.size() != 1) { - return Status::Invalid(fmt::format( - "Sorted index build must produce exactly one payload file, but produced {}.", - io_metas.size())); + return FinishIndexFile(field.Id(), index_type, source_row_count, source_meta, io_metas, + is_external_path, pool); +} + +Result<std::shared_ptr<IndexFileMeta>> PkSortedIndexFile::BuildFromSortedReader( + const DataField& field, const std::string& index_type, + const std::map<std::string, std::string>& options, int32_t data_level, + const std::vector<PrimaryKeyIndexSourceFile>& source_files, + std::unique_ptr<SortMergeReader>&& sorted_reader, + const std::shared_ptr<GlobalIndexFileWriter>& file_writer, bool is_external_path, + int32_t write_batch_size, const std::shared_ptr<MemoryPool>& pool) { + if (sorted_reader == nullptr) { + return Status::Invalid("Sorted index reader is null."); } - const GlobalIndexIOMeta& io_meta = io_metas[0]; + if (write_batch_size <= 0) { + return Status::Invalid("Sorted index write batch size must be positive."); + } + PAIMON_ASSIGN_OR_RAISE(PrimaryKeyIndexSourceMeta source_meta, + PrimaryKeyIndexSourceMeta::Create(data_level, source_files)); + PAIMON_ASSIGN_OR_RAISE(int64_t source_row_count, ValidateAndCountSourceRows(source_files)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<GlobalIndexer> indexer, + GlobalIndexerFactory::Get(index_type, options)); + if (indexer == nullptr) { + return Status::Invalid(fmt::format("Index type {} is not registered.", index_type)); + } + auto arrow_schema = arrow::schema({DataField::ConvertDataFieldToArrowField(field)}); + ArrowSchema c_arrow_schema; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*arrow_schema, &c_arrow_schema)); + ScopeGuard schema_guard([&]() { ArrowSchemaRelease(&c_arrow_schema); }); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexWriter> writer, + indexer->CreateWriter(field.Name(), &c_arrow_schema, file_writer, pool)); - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> source_meta_bytes, source_meta.Serialize(pool)); - std::optional<std::string> external_path; - if (is_external_path) { - PAIMON_ASSIGN_OR_RAISE(Path path, PathUtil::ToPath(io_meta.file_path)); - external_path = path.ToString(); + auto projection_schema = SpecialFields::CompleteSequenceAndValueKindField(arrow_schema); + auto create_consumer = + [projection_schema, + pool]() -> Result<std::unique_ptr<RowToArrowArrayConverter<KeyValue, KeyValueBatch>>> { + return KeyValueMetaProjectionConsumer::Create(projection_schema, pool); + }; + auto producer = std::make_unique<AsyncKeyValueProducerAndConsumer<KeyValue, KeyValueBatch>>( + std::move(sorted_reader), create_consumer, write_batch_size, + /*projection_thread_num=*/1, pool); + ScopeGuard close_guard([&]() { producer->Close(); }); + int64_t rows_written = 0; + while (true) { + PAIMON_ASSIGN_OR_RAISE(KeyValueBatch key_value_batch, producer->NextBatch()); + if (key_value_batch.batch == nullptr) { + break; + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::RecordBatch> record_batch, + arrow::ImportRecordBatch(key_value_batch.batch.get(), projection_schema)); + if (record_batch->num_columns() != 3 || + record_batch->column(0)->type_id() != arrow::Type::INT64) { + return Status::Invalid("Sorted index projection produced an invalid batch."); + } + auto sequence_numbers = checked_pointer_cast<arrow::Int64Array>(record_batch->column(0)); + std::vector<int64_t> ordinals; + ordinals.reserve(static_cast<size_t>(sequence_numbers->length())); + for (int64_t index = 0; index < sequence_numbers->length(); ++index) { + if (sequence_numbers->IsNull(index)) { + return Status::Invalid("Sorted index row id must not be null."); + } + int64_t ordinal = sequence_numbers->Value(index); + if (ordinal < 0 || ordinal >= source_row_count) { + return Status::Invalid( + fmt::format("Row id {} is outside sorted index group row range [0, {}).", + ordinal, source_row_count)); + } + ordinals.push_back(ordinal); + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::StructArray> values, + arrow::StructArray::Make({record_batch->column(2)}, {field.Name()})); + ArrowArray c_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*values, &c_array)); + ScopeGuard array_guard([&]() { ArrowArrayRelease(&c_array); }); + PAIMON_RETURN_NOT_OK(writer->AddBatch(&c_array, std::move(ordinals))); Review Comment: Confirmed: spill does not bound one posting list. Java uses the same one-key/one-posting-list BTree format; chunking or a limit would change the shared format or behavior, so this parity PR keeps it unchanged. ########## src/paimon/core/io/key_value_in_memory_record_reader.cpp: ########## @@ -110,6 +115,21 @@ void KeyValueInMemoryRecordReader::Close() { Result<std::shared_ptr<arrow::NumericArray<arrow::UInt64Type>>> KeyValueInMemoryRecordReader::SortBatch() const { + if (sort_comparator_ != nullptr) { + std::vector<uint64_t> indices(static_cast<size_t>(value_struct_array_->length())); + std::iota(indices.begin(), indices.end(), 0); + std::stable_sort(indices.begin(), indices.end(), [&](uint64_t left, uint64_t right) { + ColumnarRowRef left_row(value_ctx_, left); Review Comment: The refcount traffic is real, but prebuilding owning row refs adds O(n) retained row-view memory to the spill path. This is a performance tradeoff rather than a correctness issue, so it remains unchanged here. ########## src/paimon/core/schema/schema_validation.cpp: ########## @@ -341,6 +403,71 @@ Status SchemaValidation::ValidateForDeletionVectors(const CoreOptions& options) "no deletion of old data in this merge engine."); } +Status SchemaValidation::ValidatePrimaryKeyBTreeIndexes(const TableSchema& schema, + const CoreOptions& options) { + std::vector<std::string> index_columns = PrimaryKeyBTreeIndexColumns(schema.Options()); + if (index_columns.empty()) { + return Status::OK(); + } + + PAIMON_ASSIGN_OR_RAISE(PrimaryKeyIndexDefinitions definitions, + PrimaryKeyIndexDefinitions::Create(schema)); + if (!options.DeletionVectorsEnabled()) { + return Status::Invalid( + "Primary-key BTree indexes require deletion-vectors.enabled = true."); + } + if (schema.PrimaryKeys().empty()) { + return Status::Invalid("Primary-key BTree indexes require a primary-key table."); + } + if (options.GetBucket() <= 0 && !IsPostponeBucketTable(schema, options.GetBucket())) { + return Status::Invalid( + fmt::format("Primary-key BTree indexes require fixed or postpone bucket mode " + "(bucket > 0 or bucket = -2), but bucket is {}.", + options.GetBucket())); + } + PAIMON_ASSIGN_OR_RAISE( + bool deletion_vectors_merge_on_read, + OptionsUtils::GetValueFromMap<bool>(schema.Options(), kDeletionVectorsMergeOnRead, false)); + if (deletion_vectors_merge_on_read) { + return Status::Invalid( + "Primary-key BTree indexes require deletion-vectors.merge-on-read = false."); + } + PAIMON_ASSIGN_OR_RAISE( + bool pk_clustering_override, + OptionsUtils::GetValueFromMap<bool>(schema.Options(), kPkClusteringOverride, false)); + if (pk_clustering_override) { + return Status::Invalid("Primary-key BTree indexes do not support pk-clustering-override."); + } + + for (const std::string& column : index_columns) { + auto field_iter = + std::find_if(schema.Fields().begin(), schema.Fields().end(), + [&column](const DataField& field) { return field.Name() == column; }); + if (field_iter == schema.Fields().end()) { + return Status::Invalid(fmt::format("{} entry '{}' must reference an existing column.", + Options::PK_BTREE_INDEX_COLUMNS, column)); + } + if (!IsSupportedBTreeIndexType(field_iter->Type())) { + return Status::Invalid(fmt::format("{} entry '{}' has unsupported type {}.", + Options::PK_BTREE_INDEX_COLUMNS, column, + field_iter->Type()->ToString())); + } + + auto definition_iter = std::find_if( + definitions.Definitions().begin(), definitions.Definitions().end(), + [&column](const PrimaryKeyIndexDefinition& definition) { + return definition.GetFamily() == PrimaryKeyIndexDefinition::Family::BTREE && + definition.Column() == column; + }); + if (definition_iter == definitions.Definitions().end()) { + return Status::Invalid( + fmt::format("Failed to resolve primary-key BTree index column '{}'.", column)); Review Comment: Fixed. The unreachable definition lookup is removed, resolved BTree definitions are validated directly, and pk-clustering-override is centralized under Options. ########## src/paimon/core/mergetree/in_memory_sort_buffer.h: ########## @@ -57,7 +57,8 @@ class InMemorySortBuffer : public SortBuffer { const std::vector<std::string>& user_defined_sequence_fields, bool sequence_fields_ascending, const std::shared_ptr<FieldsComparator>& key_comparator, - uint64_t write_buffer_size, const std::shared_ptr<MemoryPool>& pool); + uint64_t write_buffer_size, const std::shared_ptr<MemoryPool>& pool, + const std::shared_ptr<FieldsComparator>& sort_comparator = nullptr); Review Comment: Fixed. Java floating-point ordering is now scoped to the PK BTree in-memory and spill/merge comparators. The generic merge-tree comparator keeps its previous behavior; tests cover NaN and signed zero. ########## src/paimon/core/operation/file_system_write_restore.h: ########## @@ -84,9 +84,19 @@ class FileSystemWriteRestore : public WriteRestore { partition, bucket)); } + std::vector<std::shared_ptr<IndexFileMeta>> primary_key_index_payloads; + if (scan_primary_key_indexes) { + if (index_file_handler_ == nullptr) { + return Status::Invalid("Primary-key index restore requires an index file handler."); + } + PAIMON_ASSIGN_OR_RAISE( + primary_key_index_payloads, + index_file_handler_->Scan(snapshot.value(), "btree", partition, bucket)); Review Comment: Fixed. Restore now scans source-backed entries by source_meta instead of a hard-coded index type, virtual defaults are removed, and BtreeDefs::kIdentifier is canonical. ########## src/paimon/core/operation/expire_snapshots.cpp: ########## @@ -185,6 +189,7 @@ Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t earliest_snapshot_id, continue; } PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id)); + PAIMON_RETURN_NOT_OK(CleanUnusedIndexManifest(snapshot.IndexManifest(), &skipping_sets)); Review Comment: Fixed for the supported scope. Expiration now uses current-branch live tags to retain their data files, manifests, and index payloads, matching Java. Cross-branch traversal remains unchanged. -- 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]
