wangyong9999 commented on code in PR #245:
URL: https://github.com/apache/paimon-cpp/pull/245#discussion_r3850592684
##########
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:
`skipping_sets` only covers the current retained snapshot. Tags (and
branches sharing the table root) can still reference an older index manifest,
so this can delete a live payload; a tagged DV read then fails or loses its
deletion semantics. Add all live tag/branch index manifests and payloads to the
retention set, or keep expiration rejected until that traversal exists.
##########
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:
These batches do not bound writer memory: `BTreeGlobalIndexWriter` retains
every row id for the current key until the key changes. A valid low-cardinality
field such as BOOL can therefore hold a whole level across all spill batches,
and `Flush()` allocates/copies another encoded buffer, so prepare-commit can
OOM even with spill enabled. Spill/chunk each posting list, or enforce a
controlled per-key limit.
##########
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:
The payload delete result is ignored, then the only manifest that records it
is removed. Orphan cleanup does not enumerate the global-index external path,
so a transient delete failure there becomes permanent and repeated expirations
can exhaust external storage. Keep the manifest/retry record until every
external payload is deleted or confirmed absent.
--
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]
