wangyong9999 commented on code in PR #194: URL: https://github.com/apache/paimon-cpp/pull/194#discussion_r3791281390
########## src/paimon/core/index/pksorted/pk_sorted_index_file.cpp: ########## @@ -0,0 +1,115 @@ +/* + * 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_file.h" + +#include <cstddef> +#include <optional> +#include <utility> + +#include "arrow/c/bridge.h" +#include "arrow/c/helpers.h" +#include "fmt/format.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/index/pk/primary_key_index_source_meta.h" +#include "paimon/global_index/global_index_io_meta.h" +#include "paimon/global_index/global_index_writer.h" +#include "paimon/global_index/global_indexer.h" +#include "paimon/global_index/global_indexer_factory.h" + +namespace paimon { +Result<std::shared_ptr<IndexFileMeta>> PkSortedIndexFile::Build( + 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, + const std::shared_ptr<arrow::Array>& sorted_values, std::vector<int64_t> sorted_ordinals, + const std::shared_ptr<GlobalIndexFileWriter>& file_writer, bool is_external_path, + const std::shared_ptr<MemoryPool>& pool) { + PAIMON_ASSIGN_OR_RAISE(PrimaryKeyIndexSourceMeta source_meta, + PrimaryKeyIndexSourceMeta::Create(data_level, source_files)); + int64_t source_row_count = 0; + for (const PrimaryKeyIndexSourceFile& source_file : source_files) { + if (__builtin_add_overflow(source_row_count, source_file.row_count, &source_row_count)) { + return Status::Invalid("Source row count overflows in sorted index build."); + } + } + if (source_row_count <= 0) { + return Status::Invalid("A sorted index group must reference at least one source row."); + } + if (sorted_values == nullptr || sorted_values->length() != source_row_count || + static_cast<int64_t>(sorted_ordinals.size()) != source_row_count) { + return Status::Invalid( + fmt::format("Sorted index input row count {} does not match source row count {}.", + sorted_values == nullptr ? 0 : sorted_values->length(), source_row_count)); + } + std::vector<bool> seen_ordinals(static_cast<size_t>(source_row_count), false); + for (int64_t ordinal : sorted_ordinals) { + 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)); + } + if (seen_ordinals[ordinal]) { + return Status::Invalid(fmt::format("Row id {} appears more than once.", ordinal)); + } + seen_ordinals[ordinal] = true; + } + + 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_field = DataField::ConvertDataFieldToArrowField(field); + auto arrow_schema = arrow::schema({arrow_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_FROM_ARROW(std::shared_ptr<arrow::StructArray> struct_array, + arrow::StructArray::Make({sorted_values}, {field.Name()})); + ::ArrowArray c_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*struct_array, &c_array)); + 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())); + } + const GlobalIndexIOMeta& io_meta = io_metas[0]; + + 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) { + external_path = io_meta.file_path; + } Review Comment: Updated. External payload paths are now normalized through PathUtil::ToPath and Path::ToString, with regression coverage for duplicate separators. ########## src/paimon/core/index/pksorted/pk_sorted_index_file.cpp: ########## @@ -0,0 +1,115 @@ +/* + * 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_file.h" + +#include <cstddef> +#include <optional> +#include <utility> + +#include "arrow/c/bridge.h" +#include "arrow/c/helpers.h" +#include "fmt/format.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/index/pk/primary_key_index_source_meta.h" +#include "paimon/global_index/global_index_io_meta.h" +#include "paimon/global_index/global_index_writer.h" +#include "paimon/global_index/global_indexer.h" +#include "paimon/global_index/global_indexer_factory.h" + +namespace paimon { +Result<std::shared_ptr<IndexFileMeta>> PkSortedIndexFile::Build( + 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, + const std::shared_ptr<arrow::Array>& sorted_values, std::vector<int64_t> sorted_ordinals, + const std::shared_ptr<GlobalIndexFileWriter>& file_writer, bool is_external_path, + const std::shared_ptr<MemoryPool>& pool) { + PAIMON_ASSIGN_OR_RAISE(PrimaryKeyIndexSourceMeta source_meta, Review Comment: Added an author-tagged TODO to replace the in-memory input with an external sort buffer and bounded writer batches. ########## src/paimon/core/operation/raw_file_split_read.h: ########## @@ -65,16 +64,26 @@ class RawFileSplitRead : public AbstractSplitRead { const std::shared_ptr<MemoryPool>& memory_pool, const std::shared_ptr<Executor>& executor); + /// Also accepts an `IndexedSplit` over a single-file data split, in which case its row + /// ranges narrow the read to the given file-local physical positions. Result<std::unique_ptr<BatchReader>> CreateReader(const std::shared_ptr<Split>& split) override; Result<std::unique_ptr<BatchReader>> CreateReader( const BinaryRow& partition, int32_t bucket, const std::vector<std::shared_ptr<DataFileMeta>>& files, const std::vector<std::optional<DeletionFile>>& deletion_files); + /// Reads with an optional selection of file-local row positions. The ranges apply to + /// every file of the split, so callers pass them only for single-file splits. Result<std::unique_ptr<BatchReader>> CreateReader( const BinaryRow& partition, int32_t bucket, const std::vector<std::shared_ptr<DataFileMeta>>& files, - DeletionVector::Factory dv_factory); + const std::vector<std::optional<DeletionFile>>& deletion_files, + const std::optional<std::vector<Range>>& local_row_ranges); + + Result<std::unique_ptr<BatchReader>> CreateReader( + const BinaryRow& partition, int32_t bucket, + const std::vector<std::shared_ptr<DataFileMeta>>& files, DeletionVector::Factory dv_factory, + const std::optional<std::vector<Range>>& local_row_ranges = std::nullopt); Review Comment: Updated. The existing overloads now take local_row_ranges explicitly; the forwarding overload and production default argument were removed, and callers pass std::nullopt explicitly. ########## src/paimon/core/operation/raw_file_split_read.cpp: ########## @@ -64,6 +67,21 @@ RawFileSplitRead::RawFileSplitRead(const std::shared_ptr<FileStorePathFactory>& Result<std::unique_ptr<BatchReader>> RawFileSplitRead::CreateReader( const std::shared_ptr<Split>& split) { + if (auto indexed_split = std::dynamic_pointer_cast<IndexedSplitImpl>(split)) { + PAIMON_RETURN_NOT_OK(indexed_split->Validate()); + const std::shared_ptr<DataSplit>& inner_split = indexed_split->GetDataSplit(); + auto inner_split_impl = std::dynamic_pointer_cast<DataSplitImpl>(inner_split); + if (!inner_split_impl) { + return Status::Invalid("cannot cast indexed inner split to data_split"); + } + if (inner_split_impl->DataFiles().size() != 1) { + return Status::Invalid( + "indexed splits with file-local row ranges must contain exactly one file"); + } Review Comment: Updated. Primary-key reads now fail fast for scored IndexedSplits before routing or force-keep-delete fallback; a TODO and regression coverage were added. ########## src/paimon/core/table/source/primary_key_sorted_index_result.cpp: ########## @@ -0,0 +1,146 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "paimon/core/table/source/primary_key_sorted_index_result.h" + +#include <limits> +#include <map> +#include <set> +#include <utility> + +#include "paimon/core/global_index/indexed_split_impl.h" +#include "paimon/core/table/source/deletion_file.h" + +namespace paimon { +namespace { +struct RangeConversion { + bool use_index; + std::vector<Range> ranges; +}; + +Result<std::shared_ptr<DataSplitImpl>> ToSingleFileSplit( + const PrimaryKeySortedIndexScan::FilePlan& file) { + const std::shared_ptr<DataSplitImpl>& source = file.SourceSplit(); + std::vector<std::shared_ptr<DataFileMeta>> data_files{file.DataFile()}; + DataSplitImpl::Builder builder(source->Partition(), source->Bucket(), source->BucketPath(), + std::move(data_files)); + builder.WithSnapshot(source->SnapshotId()) + .WithTotalBuckets(source->TotalBuckets()) + .IsStreaming(false) + .RawConvertible(source->RawConvertible()); Review Comment: Updated. Derived single-file splits now use rawConvertible(false), and IndexedSplits are routed explicitly to the physical-position reader while unindexed splits use the normal merge path. -- 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]
