Copilot commented on code in PR #112: URL: https://github.com/apache/paimon-cpp/pull/112#discussion_r3471640295
########## src/paimon/append/append_only_writer_test.cpp: ########## @@ -0,0 +1,710 @@ +/* + * 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/append/append_only_writer.h" Review Comment: The test includes `paimon/core/append/append_only_writer.h`, but the header added by this PR is located at `src/paimon/append/append_only_writer.h`. This mismatch will break the build unless the header is moved to match the include path (or all includes are updated consistently). ########## src/paimon/append/append_only_writer.cpp: ########## @@ -0,0 +1,286 @@ +/* + * 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/append/append_only_writer.h" Review Comment: The include path points to `paimon/core/append/append_only_writer.h`, but this PR adds the header at `src/paimon/append/append_only_writer.h`. As-is, this will not be found by the compiler unless the file is moved to `src/paimon/core/append/` (or all call sites are updated to include the correct path). ########## include/paimon/append/append_compact_coordinator.h: ########## @@ -0,0 +1,65 @@ +/* + * 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 <map> +#include <memory> +#include <string> +#include <vector> + +#include "paimon/result.h" Review Comment: This public header uses `PAIMON_EXPORT` but doesn't include `paimon/visibility.h` directly, relying on transitive includes via `paimon/result.h`. Other public headers typically include `paimon/visibility.h` explicitly, and relying on transitive includes is brittle. ########## src/paimon/append/append_compact_coordinator.cpp: ########## @@ -0,0 +1,360 @@ +/* + * 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/append/append_compact_coordinator.h" + +#include <algorithm> +#include <map> +#include <memory> +#include <string> +#include <utility> +#include <vector> + +#include "paimon/common/data/binary_row.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/linked_hash_map.h" +#include "paimon/core/append/append_compact_task.h" +#include "paimon/core/core_options.h" +#include "paimon/core/io/data_file_meta.h" +#include "paimon/core/manifest/file_kind.h" +#include "paimon/core/manifest/manifest_entry.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/append_only_file_store_write.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/snapshot.h" +#include "paimon/core/utils/field_mapping.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/executor.h" +#include "paimon/memory/memory_pool.h" +namespace paimon { + +namespace { + +/// A bin for packing small files into compaction groups. +class FileBin { + public: + FileBin(int64_t target_file_size, int64_t open_file_cost, int32_t min_file_num) + : target_file_size_(target_file_size), + open_file_cost_(open_file_cost), + min_file_num_(min_file_num) {} + + void AddFile(const std::shared_ptr<DataFileMeta>& file) { + total_file_size_ += file->file_size + open_file_cost_; + bin_.push_back(file); + } + + bool EnoughContent() const { + return bin_.size() > 1 && total_file_size_ >= target_file_size_ * 2; + } + + bool EnoughInputFiles() const { + return static_cast<int32_t>(bin_.size()) >= min_file_num_; + } + + std::vector<std::shared_ptr<DataFileMeta>> Drain() { + std::vector<std::shared_ptr<DataFileMeta>> result = std::move(bin_); + bin_.clear(); + total_file_size_ = 0; + return result; + } + + bool IsEmpty() const { + return bin_.empty(); + } + + private: + int64_t target_file_size_; + int64_t open_file_cost_; + int32_t min_file_num_; + std::vector<std::shared_ptr<DataFileMeta>> bin_; + int64_t total_file_size_ = 0; +}; + +/// Pack small files into compaction groups using a bin-packing algorithm. +/// Files are sorted by size ascending, then greedily packed into bins. +/// A bin is flushed when its total size >= targetFileSize * 2 (and has > 1 file), +/// or when it has >= minFileNum files. +std::vector<std::vector<std::shared_ptr<DataFileMeta>>> PackFiles( Review Comment: The comment says a bin is flushed "when it has >= minFileNum files", but the current implementation only applies the min-file-num rule to the final leftover bin after the loop. Either flush inside the loop when `EnoughInputFiles()` becomes true, or update the comment to match the actual behavior. ########## src/paimon/append/append_compact_task.cpp: ########## @@ -0,0 +1,97 @@ +/* + * 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/append/append_compact_task.h" Review Comment: The include path refers to `paimon/core/append/append_compact_task.h`, but this PR adds the header at `src/paimon/append/append_compact_task.h`. This will fail to compile unless the header is moved to the expected path (or all includes are updated consistently). ########## src/paimon/append/append_compact_coordinator.cpp: ########## @@ -0,0 +1,360 @@ +/* + * 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/append/append_compact_coordinator.h" + +#include <algorithm> +#include <map> +#include <memory> +#include <string> +#include <utility> +#include <vector> + +#include "paimon/common/data/binary_row.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/linked_hash_map.h" +#include "paimon/core/append/append_compact_task.h" Review Comment: This file includes `paimon/core/append/append_compact_task.h`, but the header added in this PR is at `src/paimon/append/append_compact_task.h`. The include path / file layout mismatch will break compilation unless the header is moved (preferred if the intent is a `paimon/core/append/...` API) or the include path is updated consistently across the repo. ########## src/paimon/append/append_only_writer.cpp: ########## @@ -0,0 +1,286 @@ +/* + * 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/append/append_only_writer.h" + +#include <functional> +#include <string> +#include <utility> + +#include "arrow/c/abi.h" +#include "arrow/c/bridge.h" +#include "arrow/c/helpers.h" +#include "arrow/type.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/types/row_kind.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/long_counter.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/io/compact_increment.h" +#include "paimon/core/io/data_file_path_factory.h" +#include "paimon/core/io/data_file_writer.h" +#include "paimon/core/io/data_increment.h" +#include "paimon/core/io/multiple_blob_file_writer.h" +#include "paimon/core/io/rolling_blob_file_writer.h" +#include "paimon/core/io/rolling_file_writer.h" +#include "paimon/core/io/single_file_writer.h" +#include "paimon/core/manifest/file_source.h" +#include "paimon/core/utils/commit_increment.h" +#include "paimon/format/file_format.h" +#include "paimon/format/file_format_factory.h" +#include "paimon/format/writer_builder.h" +#include "paimon/macros.h" +#include "paimon/metrics.h" +#include "paimon/record_batch.h" + +namespace paimon { + +class MemoryPool; +class FormatStatsExtractor; + +AppendOnlyWriter::AppendOnlyWriter(const CoreOptions& options, int64_t schema_id, + const std::shared_ptr<arrow::Schema>& write_schema, + const std::optional<std::vector<std::string>>& write_cols, + int64_t max_sequence_number, + const std::shared_ptr<DataFilePathFactory>& path_factory, + const std::shared_ptr<CompactManager>& compact_manager, + const std::shared_ptr<MemoryPool>& memory_pool) + : options_(options), + schema_id_(schema_id), + write_schema_(write_schema), + write_cols_(write_cols), + seq_num_counter_(std::make_shared<LongCounter>(max_sequence_number + 1)), + path_factory_(path_factory), + compact_manager_(compact_manager), + memory_pool_(memory_pool), + metrics_(std::make_shared<MetricsImpl>()) {} + +AppendOnlyWriter::~AppendOnlyWriter() = default; + +Status AppendOnlyWriter::Write(std::unique_ptr<RecordBatch>&& batch) { + for (const auto& row_kind : batch->GetRowKind()) { + if (PAIMON_UNLIKELY(row_kind != RecordBatch::RowKind::INSERT)) { + PAIMON_ASSIGN_OR_RAISE(const RowKind* kind, + RowKind::FromByteValue(static_cast<int8_t>(row_kind))); + return Status::Invalid("Append only writer can not accept record batch with RowKind ", + kind->Name()); + } + } + if (writer_ == nullptr) { + PAIMON_ASSIGN_OR_RAISE(writer_, CreateRollingRowWriter()); + } + return writer_->Write(batch->GetData()); +} + +Result<CommitIncrement> AppendOnlyWriter::PrepareCommit(bool wait_compaction) { + PAIMON_RETURN_NOT_OK( + Flush(/*wait_for_latest_compaction=*/false, /*forced_full_compaction=*/false)); + PAIMON_RETURN_NOT_OK(TrySyncLatestCompaction(wait_compaction || options_.CommitForceCompact())); + return DrainIncrement(); +} + +Result<CommitIncrement> AppendOnlyWriter::DrainIncrement() { + DataIncrement data_increment(std::move(new_files_), std::move(deleted_files_), {}); + CompactIncrement compact_increment(std::move(compact_before_), std::move(compact_after_), {}); + auto drain_deletion_file = compact_deletion_file_; + + new_files_.clear(); + deleted_files_.clear(); + compact_before_.clear(); + compact_after_.clear(); + compact_deletion_file_ = nullptr; + + return CommitIncrement(data_increment, compact_increment, drain_deletion_file); +} + +Status AppendOnlyWriter::TrySyncLatestCompaction(bool blocking) { + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<CompactResult>> result, + compact_manager_->GetCompactionResult(blocking)); + if (result.has_value()) { + const auto& compaction_result = result.value(); + const auto& before = compaction_result->Before(); + compact_before_.insert(compact_before_.end(), before.begin(), before.end()); + const auto& after = compaction_result->After(); + compact_after_.insert(compact_after_.end(), after.begin(), after.end()); + PAIMON_RETURN_NOT_OK(UpdateCompactDeletionFile(compaction_result->DeletionFile())); + } + return Status::OK(); +} + +Status AppendOnlyWriter::UpdateCompactDeletionFile( + const std::shared_ptr<CompactDeletionFile>& new_deletion_file) { + if (new_deletion_file) { + if (compact_deletion_file_ == nullptr) { + compact_deletion_file_ = new_deletion_file; + } else { + PAIMON_ASSIGN_OR_RAISE(compact_deletion_file_, + new_deletion_file->MergeOldFile(compact_deletion_file_)); + } + } + return Status::OK(); +} + +Status AppendOnlyWriter::Flush(bool wait_for_latest_compaction, bool forced_full_compaction) { + std::vector<std::shared_ptr<DataFileMeta>> flushed_files; + if (writer_) { + PAIMON_RETURN_NOT_OK(writer_->Close()); + PAIMON_ASSIGN_OR_RAISE(flushed_files, writer_->GetResult()); + } + // add new generated files + for (const auto& flushed_file : flushed_files) { + PAIMON_RETURN_NOT_OK(compact_manager_->AddNewFile(flushed_file)); + } + PAIMON_RETURN_NOT_OK(TrySyncLatestCompaction(wait_for_latest_compaction)); + PAIMON_RETURN_NOT_OK(compact_manager_->TriggerCompaction(forced_full_compaction)); + new_files_.insert(new_files_.end(), flushed_files.begin(), flushed_files.end()); + if (writer_) { + metrics_->Merge(writer_->GetMetrics()); + writer_.reset(); + } + return Status::OK(); +} + +AppendOnlyWriter::RollingFileWriterResult AppendOnlyWriter::CreateRollingRowWriter() const { + auto schemas = BlobUtils::SeparateBlobSchema(write_schema_); + if (schemas.blob_schema && schemas.blob_schema->num_fields() > 0) { + return CreateRollingBlobWriter(schemas); + } else { + return std::make_unique<RollingFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>( + options_.GetTargetFileSize(/*has_primary_key=*/false), + GetDataFileWriterCreator(write_schema_, write_cols_)); + } +} + +AppendOnlyWriter::SingleFileWriterCreator AppendOnlyWriter::GetDataFileWriterCreator( + const std::shared_ptr<arrow::Schema>& schema, + const std::optional<std::vector<std::string>>& write_cols) const { + return + [this, schema, write_cols]() + -> Result< + std::unique_ptr<SingleFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>> { + ::ArrowSchema arrow_schema; + ScopeGuard guard([&arrow_schema]() { ArrowSchemaRelease(&arrow_schema); }); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &arrow_schema)); + auto format = options_.GetFileFormat(); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr<WriterBuilder> writer_builder, + format->CreateWriterBuilder(&arrow_schema, options_.GetWriteBatchSize())); + writer_builder->WithMemoryPool(memory_pool_); + + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &arrow_schema)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FormatStatsExtractor> stats_extractor, + format->CreateStatsExtractor(&arrow_schema)); + auto writer = std::make_unique<DataFileWriter>( + options_.GetFileCompression(), std::function<Status(ArrowArray*, ArrowArray*)>(), + schema_id_, seq_num_counter_, FileSource::Append(), stats_extractor, + path_factory_->IsExternalPath(), write_cols, memory_pool_); + PAIMON_RETURN_NOT_OK( + writer->Init(options_.GetFileSystem(), path_factory_->NewPath(), writer_builder)); + return writer; + }; +} + +AppendOnlyWriter::SingleFileWriterCreator AppendOnlyWriter::GetBlobFileWriterCreator( + const std::shared_ptr<WriterBuilder>& writer_builder, + const std::shared_ptr<FormatStatsExtractor>& stats_extractor, + const std::optional<std::vector<std::string>>& write_cols) const { + return + [this, writer_builder, stats_extractor, write_cols]() + -> Result< + std::unique_ptr<SingleFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>> { + auto writer = std::make_unique<DataFileWriter>( + /*compression=*/"none", std::function<Status(ArrowArray*, ArrowArray*)>(), + schema_id_, seq_num_counter_, FileSource::Append(), stats_extractor, + path_factory_->IsExternalPath(), write_cols, memory_pool_); + PAIMON_RETURN_NOT_OK(writer->Init(options_.GetFileSystem(), + path_factory_->NewBlobPath(), writer_builder)); + return writer; + }; +} + +AppendOnlyWriter::RollingFileWriterResult AppendOnlyWriter::CreateRollingBlobWriter( + const BlobUtils::SeparatedSchemas& schemas) const { + // Multiple blob fields are supported. Each blob field gets its own rolling file writer + // via MultipleBlobFileWriter. + auto blob_schema = schemas.blob_schema; + auto blob_writer_creator = [this, blob_schema](const std::string& blob_field_name) + -> Result< + std::unique_ptr<RollingFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>> { + // Create a single-field schema for this blob field + auto field = blob_schema->GetFieldByName(blob_field_name); + if (!field) { + return Status::Invalid( + fmt::format("Blob field '{}' not found in blob schema", blob_field_name)); + } + auto single_field_schema = arrow::schema({field}); + ::ArrowSchema arrow_schema; + ScopeGuard guard([&arrow_schema]() { ArrowSchemaRelease(&arrow_schema); }); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*single_field_schema, &arrow_schema)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileFormat> format, + FileFormatFactory::Get("blob", options_.ToMap())); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr<WriterBuilder> writer_builder, + format->CreateWriterBuilder(&arrow_schema, options_.GetWriteBatchSize())); + writer_builder->WithMemoryPool(memory_pool_); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*single_field_schema, &arrow_schema)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FormatStatsExtractor> stats_extractor, + format->CreateStatsExtractor(&arrow_schema)); Review Comment: `ArrowSchema arrow_schema;` is not value-initialized before being released in the scope guard, which is undefined behavior if `ExportSchema` fails before setting `release`. This block also re-exports into the same `ArrowSchema` storage; explicitly resetting it avoids leaking the previous export if the consumer doesn't release it. ########## src/paimon/append/append_only_writer.cpp: ########## @@ -0,0 +1,286 @@ +/* + * 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/append/append_only_writer.h" + +#include <functional> +#include <string> +#include <utility> + +#include "arrow/c/abi.h" +#include "arrow/c/bridge.h" +#include "arrow/c/helpers.h" +#include "arrow/type.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/types/row_kind.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/long_counter.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/io/compact_increment.h" +#include "paimon/core/io/data_file_path_factory.h" +#include "paimon/core/io/data_file_writer.h" +#include "paimon/core/io/data_increment.h" +#include "paimon/core/io/multiple_blob_file_writer.h" +#include "paimon/core/io/rolling_blob_file_writer.h" +#include "paimon/core/io/rolling_file_writer.h" +#include "paimon/core/io/single_file_writer.h" +#include "paimon/core/manifest/file_source.h" +#include "paimon/core/utils/commit_increment.h" +#include "paimon/format/file_format.h" +#include "paimon/format/file_format_factory.h" +#include "paimon/format/writer_builder.h" +#include "paimon/macros.h" +#include "paimon/metrics.h" +#include "paimon/record_batch.h" + +namespace paimon { + +class MemoryPool; +class FormatStatsExtractor; + +AppendOnlyWriter::AppendOnlyWriter(const CoreOptions& options, int64_t schema_id, + const std::shared_ptr<arrow::Schema>& write_schema, + const std::optional<std::vector<std::string>>& write_cols, + int64_t max_sequence_number, + const std::shared_ptr<DataFilePathFactory>& path_factory, + const std::shared_ptr<CompactManager>& compact_manager, + const std::shared_ptr<MemoryPool>& memory_pool) + : options_(options), + schema_id_(schema_id), + write_schema_(write_schema), + write_cols_(write_cols), + seq_num_counter_(std::make_shared<LongCounter>(max_sequence_number + 1)), + path_factory_(path_factory), + compact_manager_(compact_manager), + memory_pool_(memory_pool), + metrics_(std::make_shared<MetricsImpl>()) {} + +AppendOnlyWriter::~AppendOnlyWriter() = default; + +Status AppendOnlyWriter::Write(std::unique_ptr<RecordBatch>&& batch) { + for (const auto& row_kind : batch->GetRowKind()) { + if (PAIMON_UNLIKELY(row_kind != RecordBatch::RowKind::INSERT)) { + PAIMON_ASSIGN_OR_RAISE(const RowKind* kind, + RowKind::FromByteValue(static_cast<int8_t>(row_kind))); + return Status::Invalid("Append only writer can not accept record batch with RowKind ", + kind->Name()); + } + } + if (writer_ == nullptr) { + PAIMON_ASSIGN_OR_RAISE(writer_, CreateRollingRowWriter()); + } + return writer_->Write(batch->GetData()); +} + +Result<CommitIncrement> AppendOnlyWriter::PrepareCommit(bool wait_compaction) { + PAIMON_RETURN_NOT_OK( + Flush(/*wait_for_latest_compaction=*/false, /*forced_full_compaction=*/false)); + PAIMON_RETURN_NOT_OK(TrySyncLatestCompaction(wait_compaction || options_.CommitForceCompact())); + return DrainIncrement(); +} + +Result<CommitIncrement> AppendOnlyWriter::DrainIncrement() { + DataIncrement data_increment(std::move(new_files_), std::move(deleted_files_), {}); + CompactIncrement compact_increment(std::move(compact_before_), std::move(compact_after_), {}); + auto drain_deletion_file = compact_deletion_file_; + + new_files_.clear(); + deleted_files_.clear(); + compact_before_.clear(); + compact_after_.clear(); + compact_deletion_file_ = nullptr; + + return CommitIncrement(data_increment, compact_increment, drain_deletion_file); +} + +Status AppendOnlyWriter::TrySyncLatestCompaction(bool blocking) { + PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<CompactResult>> result, + compact_manager_->GetCompactionResult(blocking)); + if (result.has_value()) { + const auto& compaction_result = result.value(); + const auto& before = compaction_result->Before(); + compact_before_.insert(compact_before_.end(), before.begin(), before.end()); + const auto& after = compaction_result->After(); + compact_after_.insert(compact_after_.end(), after.begin(), after.end()); + PAIMON_RETURN_NOT_OK(UpdateCompactDeletionFile(compaction_result->DeletionFile())); + } + return Status::OK(); +} + +Status AppendOnlyWriter::UpdateCompactDeletionFile( + const std::shared_ptr<CompactDeletionFile>& new_deletion_file) { + if (new_deletion_file) { + if (compact_deletion_file_ == nullptr) { + compact_deletion_file_ = new_deletion_file; + } else { + PAIMON_ASSIGN_OR_RAISE(compact_deletion_file_, + new_deletion_file->MergeOldFile(compact_deletion_file_)); + } + } + return Status::OK(); +} + +Status AppendOnlyWriter::Flush(bool wait_for_latest_compaction, bool forced_full_compaction) { + std::vector<std::shared_ptr<DataFileMeta>> flushed_files; + if (writer_) { + PAIMON_RETURN_NOT_OK(writer_->Close()); + PAIMON_ASSIGN_OR_RAISE(flushed_files, writer_->GetResult()); + } + // add new generated files + for (const auto& flushed_file : flushed_files) { + PAIMON_RETURN_NOT_OK(compact_manager_->AddNewFile(flushed_file)); + } + PAIMON_RETURN_NOT_OK(TrySyncLatestCompaction(wait_for_latest_compaction)); + PAIMON_RETURN_NOT_OK(compact_manager_->TriggerCompaction(forced_full_compaction)); + new_files_.insert(new_files_.end(), flushed_files.begin(), flushed_files.end()); + if (writer_) { + metrics_->Merge(writer_->GetMetrics()); + writer_.reset(); + } + return Status::OK(); +} + +AppendOnlyWriter::RollingFileWriterResult AppendOnlyWriter::CreateRollingRowWriter() const { + auto schemas = BlobUtils::SeparateBlobSchema(write_schema_); + if (schemas.blob_schema && schemas.blob_schema->num_fields() > 0) { + return CreateRollingBlobWriter(schemas); + } else { + return std::make_unique<RollingFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>( + options_.GetTargetFileSize(/*has_primary_key=*/false), + GetDataFileWriterCreator(write_schema_, write_cols_)); + } +} + +AppendOnlyWriter::SingleFileWriterCreator AppendOnlyWriter::GetDataFileWriterCreator( + const std::shared_ptr<arrow::Schema>& schema, + const std::optional<std::vector<std::string>>& write_cols) const { + return + [this, schema, write_cols]() + -> Result< + std::unique_ptr<SingleFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>> { + ::ArrowSchema arrow_schema; + ScopeGuard guard([&arrow_schema]() { ArrowSchemaRelease(&arrow_schema); }); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &arrow_schema)); + auto format = options_.GetFileFormat(); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr<WriterBuilder> writer_builder, + format->CreateWriterBuilder(&arrow_schema, options_.GetWriteBatchSize())); + writer_builder->WithMemoryPool(memory_pool_); + + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &arrow_schema)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FormatStatsExtractor> stats_extractor, + format->CreateStatsExtractor(&arrow_schema)); Review Comment: `ArrowSchema arrow_schema;` is not value-initialized before being passed to `ArrowSchemaRelease` via the scope guard. If `arrow::ExportSchema` fails early, `arrow_schema.release` may be garbage and releasing it is undefined behavior. Also, the same `ArrowSchema` storage is reused for a second `ExportSchema` call; resetting it avoids leaks if the first consumer didn't release it. -- 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]
