zjw1111 commented on code in PR #393:
URL: https://github.com/apache/paimon-cpp/pull/393#discussion_r4268339236
##########
src/paimon/core/table/source/append_only_table_read.cpp:
##########
@@ -232,6 +295,85 @@ Result<std::unique_ptr<BatchReader>>
AppendOnlyTableRead::CreateDiskReader(
return Status::Invalid("create reader failed, not read match with split.");
}
+Result<std::unique_ptr<BatchReader>>
AppendOnlyTableRead::CreateDeduplicateDiskReader(
+ const std::shared_ptr<Split>& split,
+ const std::shared_ptr<const RoaringBitmap64>& offset_deletions) {
+ if (!offset_deletions) {
+ return Status::Invalid("deduplicate real-time split does not pin an
offset DV");
+ }
+ auto raw_read =
dynamic_cast<RawFileSplitRead*>(split_reads_.front().get());
+ if (!raw_read) {
+ return Status::NotImplemented(
+ "deduplicate real-time offset-DV overlay only supports raw append
reads");
+ }
+
+ std::shared_ptr<DataSplitImpl> data_split;
+ std::optional<std::vector<Range>> row_ranges;
+ if (auto indexed_split =
std::dynamic_pointer_cast<IndexedSplitImpl>(split)) {
+ PAIMON_RETURN_NOT_OK(indexed_split->Validate());
+ if (!indexed_split->Scores().empty()) {
+ return Status::NotImplemented(
+ "deduplicate real-time reads do not support scored indexed
splits yet");
+ }
+ data_split =
std::dynamic_pointer_cast<DataSplitImpl>(indexed_split->GetDataSplit());
+ row_ranges = indexed_split->RowRanges();
+ } else {
+ data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+ }
+ if (!data_split) {
+ return Status::Invalid("deduplicate real-time disk split is not a data
split");
+ }
+
+ if (offset_deletions->IsEmpty()) {
+ return raw_read->CreateReader(data_split->Partition(),
data_split->Bucket(),
+ data_split->DataFiles(),
data_split->DeletionFiles(),
+ row_ranges);
+ }
+
+ std::shared_ptr<arrow::Schema> read_schema = context_->GetReadSchema();
+ const std::string& offset_name = SpecialFields::RealtimeOffset().Name();
+ const bool expose_offset = read_schema->GetFieldIndex(offset_name) >= 0;
+ if (!expose_offset) {
+ arrow::FieldVector fields = read_schema->fields();
+ PAIMON_ASSIGN_OR_RAISE(DataField offset_field,
+
context_->GetTableSchema()->GetField(offset_name));
+
fields.push_back(DataField::ConvertDataFieldToArrowField(offset_field));
+ read_schema = arrow::schema(std::move(fields),
read_schema->metadata());
+ }
+
+ // TODO(xinyu.lxy): Build a file-specific visibility predicate for each
data file. Use the
+ // `_REALTIME_OFFSET` min/max statistics in DataFileMeta to remove deleted
offsets that cannot
+ // intersect that file, instead of pushing the complete bucket offset DV
to every file.
+ std::vector<Literal> deleted_offsets;
+ deleted_offsets.reserve(offset_deletions->Cardinality());
+ for (RoaringBitmap64::Iterator offset = offset_deletions->Begin();
+ offset != offset_deletions->End(); ++offset) {
+ deleted_offsets.emplace_back(static_cast<int64_t>(*offset));
+ }
+ const int32_t offset_position = read_schema->GetFieldIndex(offset_name);
+ std::shared_ptr<Predicate> visibility_predicate =
+ PredicateBuilder::NotIn(offset_position, offset_name,
FieldType::BIGINT, deleted_offsets);
+ std::shared_ptr<Predicate> predicate = visibility_predicate;
+ if (context_->GetPredicate()) {
+ PAIMON_ASSIGN_OR_RAISE(predicate,
PredicateBuilder::And({context_->GetPredicate(),
+
std::move(visibility_predicate)}));
+ }
+
+ DeletionVector::Factory committed_factory = DeletionVector::CreateFactory(
+ context_->GetCoreOptions().GetFileSystem(),
+ DeletionVector::CreateDeletionFileMap(data_split->DataFiles(),
data_split->DeletionFiles()),
+ context_->GetMemoryPool());
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<BatchReader> reader,
+ raw_read->CreateReader(data_split->Partition(), data_split->Bucket(),
+ data_split->DataFiles(),
std::move(committed_factory), row_ranges,
+ read_schema, predicate));
Review Comment:
[P1] Real-time deletions can be skipped when both filtering paths are
disabled
With `file-index.read.enabled=false` and the default
`EnablePredicateFilter(false)`, file-index selection is skipped and
`ApplyPredicateFilterIfNeeded()` returns the reader unchanged. Parquet metadata
pruning can retain a page containing both deleted and live offsets, so the
`NOT_IN` visibility predicate passed here does not remove every deleted row.
Disk rows overwritten or deleted in memory can consequently remain visible
before their deletion vectors are committed. The realtime option validation
accepts this configuration.
##########
src/paimon/core/realtime/realtime_key_offset_index.cpp:
##########
@@ -0,0 +1,315 @@
+/*
+ * 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/realtime/realtime_key_offset_index.h"
+
+#include <optional>
+#include <utility>
+
+#include "arrow/api.h"
+#include "paimon/common/global_index/key_serializer.h"
+#include "paimon/common/io/data_output_stream.h"
+#include "paimon/common/predicate/literal_converter.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/io/data_input_stream.h"
+#include "paimon/macros.h"
+#include "paimon/memory/bytes.h"
+
+namespace paimon {
+namespace {
+
+constexpr int64_t kOffsetIndexMagic = 0x5041494D4F464653LL;
+constexpr int32_t kOffsetIndexVersion = 1;
+
+Result<std::vector<std::string>> EncodeKeyBatch(
+ const std::shared_ptr<arrow::StructArray>& keys, const
std::shared_ptr<arrow::Field>& key_field,
+ const std::shared_ptr<KeySerializer>& key_serializer) {
+ if (!keys || keys->num_fields() != 1 ||
!keys->field(0)->type()->Equals(key_field->type())) {
+ return Status::Invalid(
+ "deduplicate key lookup must contain exactly the configured key
field");
+ }
+ if (keys->null_count() != 0 || keys->field(0)->null_count() != 0) {
+ return Status::Invalid("deduplicate key must not contain null");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<Literal> literals,
+
LiteralConverter::ConvertLiteralsFromArray(*keys->field(0),
+
/*own_data=*/true));
+ std::vector<std::string> encoded;
+ encoded.reserve(literals.size());
+ for (const Literal& literal : literals) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes,
key_serializer->Serialize(literal));
+ std::string key;
+ if (bytes->size() > 0) {
+ key.append(bytes->data(), bytes->size());
+ }
+ encoded.push_back(std::move(key));
+ }
+ return encoded;
+}
+
+Status WriteRaw(const std::shared_ptr<OutputStream>& output, const char* data,
int64_t size) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t written, output->Write(data, size));
+ if (written != size) {
+ return Status::IOError("short write for realtime .offset index");
+ }
+ return Status::OK();
+}
+
+} // namespace
+
+Result<std::shared_ptr<MutableKeyOffsetIndex>> MutableKeyOffsetIndex::Create(
+ const std::shared_ptr<arrow::Field>& key_field,
+ const std::shared_ptr<MemoryPool>& memory_pool) {
+ if (!key_field || !memory_pool) {
+ return Status::Invalid("mutable key-offset index requires a key field
and memory pool");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<KeySerializer> serializer,
+ KeySerializer::Create(key_field->type(),
memory_pool));
+ return std::shared_ptr<MutableKeyOffsetIndex>(
+ new MutableKeyOffsetIndex(key_field, std::move(serializer),
std::make_shared<OffsetMap>()));
+}
+
+MutableKeyOffsetIndex::MutableKeyOffsetIndex(std::shared_ptr<arrow::Field>
key_field,
+ std::shared_ptr<KeySerializer>
key_serializer,
+ std::shared_ptr<OffsetMap>
offsets)
+ : key_field_(std::move(key_field)),
+ key_serializer_(std::move(key_serializer)),
+ offsets_(std::move(offsets)) {}
+
+Result<std::vector<std::string>> MutableKeyOffsetIndex::EncodeKeys(
+ const std::shared_ptr<arrow::StructArray>& keys) const {
+ return EncodeKeyBatch(keys, key_field_, key_serializer_);
+}
+
+Status MutableKeyOffsetIndex::Add(const std::shared_ptr<arrow::StructArray>&
keys,
+ const std::shared_ptr<arrow::Int64Array>&
offsets) {
+ if (!offsets || offsets->null_count() != 0 || !keys || keys->length() !=
offsets->length()) {
+ return Status::Invalid("key-offset index input columns are not
aligned");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> encoded, EncodeKeys(keys));
+ // Copy on write to keep previously pinned snapshots immutable.
+ if (!offsets_.unique()) {
+ offsets_ = std::make_shared<OffsetMap>(*offsets_);
+ }
+ for (int64_t row = 0; row < offsets->length(); ++row) {
+ const int64_t offset = offsets->Value(row);
+ if (offset < 0) {
+ return Status::Invalid("realtime offset must not be negative");
+ }
+ (*offsets_)[encoded[row]] = offset;
+ }
+ return Status::OK();
+}
+
+Status MutableKeyOffsetIndex::Erase(const std::shared_ptr<arrow::StructArray>&
keys) {
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> encoded, EncodeKeys(keys));
+ if (!offsets_.unique()) {
+ offsets_ = std::make_shared<OffsetMap>(*offsets_);
+ }
+ for (const std::string& key : encoded) {
+ offsets_->erase(key);
+ }
+ return Status::OK();
+}
+
+Result<RoaringBitmap64> MutableKeyOffsetIndex::LookupOffsets(
+ const std::shared_ptr<arrow::StructArray>& keys) const {
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> encoded, EncodeKeys(keys));
+ RoaringBitmap64 result;
+ for (const std::string& key : encoded) {
+ auto iter = offsets_->find(key);
+ if (iter != offsets_->end()) {
+ result.Add(iter->second);
+ }
+ }
+ return result;
+}
+
+std::shared_ptr<const KeyOffsetLookup> MutableKeyOffsetIndex::Snapshot() const
{
+ return std::shared_ptr<const KeyOffsetLookup>(
+ new MutableKeyOffsetIndex(key_field_, key_serializer_, offsets_));
+}
+
+Result<RoaringBitmap64> CompositeKeyOffsetLookup::LookupOffsets(
+ const std::shared_ptr<arrow::StructArray>& keys) const {
+ RoaringBitmap64 result;
+ for (const std::shared_ptr<const KeyOffsetLookup>& lookup : lookups_) {
+ if (!lookup) {
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 offsets,
lookup->LookupOffsets(keys));
+ result |= offsets;
+ }
+ return result;
+}
+
+Result<std::unique_ptr<DataFileKeyOffsetIndexWriter>>
DataFileKeyOffsetIndexWriter::Create(
+ const std::shared_ptr<arrow::Field>& key_field,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::shared_ptr<FileSystem>& file_system, const
std::shared_ptr<MemoryPool>& memory_pool,
+ const std::map<std::string, std::string>& options) {
+ (void)options;
+ if (!key_field || !path_factory || !file_system || !memory_pool) {
+ return Status::Invalid("key-offset writer is missing a required
dependency");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<KeySerializer> serializer,
+ KeySerializer::Create(key_field->type(),
memory_pool));
+ return std::unique_ptr<DataFileKeyOffsetIndexWriter>(new
DataFileKeyOffsetIndexWriter(
+ key_field, path_factory, file_system, std::move(serializer)));
+}
+
+DataFileKeyOffsetIndexWriter::DataFileKeyOffsetIndexWriter(
+ std::shared_ptr<arrow::Field> key_field,
std::shared_ptr<DataFilePathFactory> path_factory,
+ std::shared_ptr<FileSystem> file_system, std::shared_ptr<KeySerializer>
key_serializer)
+ : key_field_(std::move(key_field)),
+ path_factory_(std::move(path_factory)),
+ file_system_(std::move(file_system)),
+ key_serializer_(std::move(key_serializer)) {}
+
+Status DataFileKeyOffsetIndexWriter::AddBatch(const
std::shared_ptr<arrow::StructArray>& keys,
+ const
std::shared_ptr<arrow::Int64Array>& offsets) {
+ if (!keys || !offsets || keys->length() != offsets->length() ||
offsets->null_count() != 0) {
+ return Status::Invalid("key-offset writer input columns are not
aligned");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> encoded,
+ EncodeKeyBatch(keys, key_field_, key_serializer_));
+ for (int64_t row = 0; row < offsets->length(); ++row) {
+ const int64_t offset = offsets->Value(row);
+ if (offset < 0) {
+ return Status::Invalid("realtime offset must not be negative");
+ }
+ if (!offsets_.emplace(encoded[row], offset).second) {
+ return Status::Invalid(
+ "deduplicate data file contains more than one offset for a
deduplicate key");
+ }
+ }
+ return Status::OK();
+}
+
+Result<std::string> DataFileKeyOffsetIndexWriter::Finish(
+ const std::shared_ptr<DataFileMeta>& data_file) {
+ if (!data_file || offsets_.empty()) {
+ return Status::Invalid("cannot finish an empty key-offset index");
+ }
+ const std::string file_name = data_file->file_name +
DataFileKeyOffsetIndexReader::kFileSuffix;
+ output_path_ = path_factory_->ToAlignedPath(file_name, data_file);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OutputStream> output,
+ file_system_->Create(output_path_,
/*overwrite=*/false));
+ DataOutputStream data_output(output);
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int64_t>(kOffsetIndexMagic));
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(kOffsetIndexVersion));
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int64_t>(offsets_.size()));
+ for (const auto& [key, offset] : offsets_) {
+
PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(static_cast<int32_t>(key.size())));
+ PAIMON_RETURN_NOT_OK(WriteRaw(output, key.data(), key.size()));
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int64_t>(offset));
+ }
+ PAIMON_RETURN_NOT_OK(output->Flush());
+ PAIMON_RETURN_NOT_OK(output->Close());
+ return file_name;
+}
+
+void DataFileKeyOffsetIndexWriter::Abort() {
+ if (!output_path_.empty()) {
+ [[maybe_unused]] Status status = file_system_->Delete(output_path_);
+ output_path_.clear();
+ }
+}
+
+Result<std::shared_ptr<DataFileKeyOffsetIndexReader>>
DataFileKeyOffsetIndexReader::Create(
+ const std::shared_ptr<arrow::Field>& key_field, const
std::shared_ptr<DataFileMeta>& data_file,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::shared_ptr<FileSystem>& file_system, const
std::shared_ptr<MemoryPool>& memory_pool,
+ const std::map<std::string, std::string>& options) {
+ (void)options;
+ if (!key_field || !data_file || !path_factory || !file_system ||
!memory_pool) {
+ return Status::Invalid("key-offset reader is missing a required
dependency");
+ }
+ std::optional<std::string> file_name;
+ for (const std::optional<std::string>& extra : data_file->extra_files) {
+ if (extra && StringUtils::EndsWith(extra.value(), kFileSuffix)) {
+ if (file_name) {
+ return Status::Invalid("data file contains multiple .offset
sidecars");
+ }
+ file_name = extra.value();
+ }
+ }
+ if (!file_name) {
+ return Status::Invalid("data file does not contain the required
.offset sidecar: ",
+ data_file->file_name);
+ }
+
+ const std::string file_path =
path_factory->ToAlignedPath(file_name.value(), data_file);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input,
file_system->Open(file_path));
+ DataInputStream data_input(input);
+ PAIMON_ASSIGN_OR_RAISE(int64_t magic, data_input.ReadValue<int64_t>());
+ PAIMON_ASSIGN_OR_RAISE(int32_t version, data_input.ReadValue<int32_t>());
+ if (magic != kOffsetIndexMagic || version != kOffsetIndexVersion) {
+ return Status::Invalid("unsupported realtime .offset index format: ",
file_name.value());
+ }
+ PAIMON_ASSIGN_OR_RAISE(int64_t entry_count,
data_input.ReadValue<int64_t>());
+ if (entry_count < 0) {
+ return Status::Invalid("negative entry count in realtime .offset
index");
+ }
+ auto offsets = std::make_shared<std::map<std::string, int64_t>>();
+ for (int64_t entry = 0; entry < entry_count; ++entry) {
+ PAIMON_ASSIGN_OR_RAISE(int32_t key_size,
data_input.ReadValue<int32_t>());
+ if (key_size <= 0) {
Review Comment:
[P2] The reader rejects empty string keys produced by the writer
Empty string keys satisfy the non-null key requirement and serialize to zero
bytes. `AddBatch()` accepts them and `Finish()` writes `key_size=0`, but this
check rejects the same entry. A write, prepare and commit containing `key=""`
can therefore succeed, then `RefreshCommittedSnapshot()` or restart recovery
fails with `invalid key size in realtime .offset index` when the sidecar is
loaded.
##########
src/paimon/core/realtime/realtime_deduplicate_writer.cpp:
##########
@@ -0,0 +1,390 @@
+/*
+ * 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/realtime/realtime_deduplicate_writer.h"
+
+#include <algorithm>
+#include <optional>
+#include <set>
+#include <utility>
+
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
+#include "arrow/c/helpers.h"
+#include "paimon/common/reader/concat_batch_reader.h"
+#include "paimon/common/table/special_fields.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/scope_guard.h"
+#include "paimon/core/append/append_only_writer.h"
+#include "paimon/core/compact/compact_deletion_file.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/deletionvectors/bucketed_dv_maintainer.h"
+#include "paimon/core/index/index_file_meta.h"
+#include "paimon/core/io/compact_increment.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/core/io/data_increment.h"
+#include "paimon/core/realtime/arrow_deduplicate_realtime_store.h"
+#include "paimon/core/realtime/realtime_context_impl.h"
+#include "paimon/core/realtime/realtime_deduplicate_state.h"
+#include "paimon/core/realtime/realtime_key_offset_index.h"
+#include "paimon/core/realtime/realtime_offset_file_index_lookup.h"
+#include "paimon/core/realtime/realtime_offset_utils.h"
+#include "paimon/core/realtime/realtime_schema_layout.h"
+#include "paimon/core/realtime/realtime_utils.h"
+#include "paimon/core/utils/commit_increment.h"
+#include "paimon/macros.h"
+#include "paimon/realtime/realtime_context.h"
+
+namespace paimon {
+namespace {
+
+template <typename T>
+void AppendAll(const std::vector<std::shared_ptr<T>>& source,
+ std::vector<std::shared_ptr<T>>* target) {
+ target->insert(target->end(), source.begin(), source.end());
+}
+
+} // namespace
+
+Result<std::shared_ptr<RealtimeDeduplicateWriter>>
RealtimeDeduplicateWriter::Create(
+ const std::map<std::string, std::string>& partition, int32_t bucket,
+ const std::shared_ptr<RealtimeContext>& realtime_context,
+ const std::shared_ptr<AppendOnlyWriter>& file_writer,
+ const std::shared_ptr<RealtimeSchemaLayout>& schema_layout, int64_t
schema_id,
+ const std::vector<std::string>& deduplicate_key_fields,
+ const std::vector<std::shared_ptr<DataFileMeta>>& restored_data_files,
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const std::shared_ptr<BucketedDvMaintainer>& dv_maintainer, const
CoreOptions& options,
+ const std::string& temp_directory, const std::shared_ptr<MemoryPool>&
memory_pool) {
+ if (!realtime_context || !file_writer || !schema_layout ||
!data_file_path_factory ||
+ !dv_maintainer || !memory_pool) {
+ return Status::Invalid("deduplicate real-time writer is missing a
dependency");
+ }
+ PAIMON_ASSIGN_OR_RAISE(int32_t key_position,
+ RealtimeUtils::GetDeduplicateKeyPosition(
+ schema_layout->StoreWriteSchema(),
deduplicate_key_fields));
+
PAIMON_RETURN_NOT_OK(RealtimeUtils::ValidateOffsetField(schema_layout->StoreWriteSchema()));
+
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RealtimeContextImpl> context,
+ RealtimeContextImpl::Cast(realtime_context));
+ auto write_schema = std::make_unique<ArrowSchema>();
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(
+ arrow::ExportSchema(*schema_layout->StoreWriteSchema(),
write_schema.get()));
+ RealtimeStoreCreateRequest request{std::move(write_schema),
options.ToMap(), memory_pool,
+ RealtimeStoreMode::DEDUPLICATE,
+
options.GetRealtimeStoreStatisticsMode()};
+ request.deduplicate_key_fields = deduplicate_key_fields;
+ if (options.RealtimeSpillEnabled()) {
+ request.temp_directory = temp_directory;
+ request.file_system = options.GetFileSystem();
+ }
+ PAIMON_ASSIGN_OR_RAISE(RealtimeStoreState store_state,
+ context->GetOrCreateRealtimeStore(
+ std::move(request),
RealtimePartitionBucket(partition, bucket)));
+ std::shared_ptr<ArrowDeduplicateRealtimeStore> store =
+
std::dynamic_pointer_cast<ArrowDeduplicateRealtimeStore>(store_state.store);
+ if (!store || !store_state.deduplicate_state) {
+ return Status::Invalid("DEDUPLICATE mode requires the default Arrow
store and state");
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<RealtimeOffsetFileIndexLookup> created_lookup,
+ RealtimeOffsetFileIndexLookup::Create(
+ schema_layout->CommitSchema(), schema_id,
+ schema_layout->StoreWriteSchema()->field(key_position),
restored_data_files,
+ data_file_path_factory, options.GetFileSystem(), memory_pool,
options));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<RealtimeOffsetFileIndexLookup> committed_lookup,
+ store_state.deduplicate_state->AttachFileIndexLookup(created_lookup,
store));
+ (void)committed_lookup;
+ return std::shared_ptr<RealtimeDeduplicateWriter>(new
RealtimeDeduplicateWriter(
+ store, file_writer, schema_layout,
schema_layout->StoreWriteSchema()->field(key_position),
+ data_file_path_factory, options.GetFileSystem(), dv_maintainer,
store_state.initial_offset,
+ memory_pool, store_state.deduplicate_state, options.ToMap()));
+}
+
+RealtimeDeduplicateWriter::RealtimeDeduplicateWriter(
+ std::shared_ptr<ArrowDeduplicateRealtimeStore> realtime_store,
+ std::shared_ptr<AppendOnlyWriter> file_writer,
+ std::shared_ptr<RealtimeSchemaLayout> schema_layout,
+ std::shared_ptr<arrow::Field> deduplicate_key_field,
+ std::shared_ptr<DataFilePathFactory> data_file_path_factory,
+ std::shared_ptr<FileSystem> file_system,
std::shared_ptr<BucketedDvMaintainer> dv_maintainer,
+ int64_t next_offset, std::shared_ptr<MemoryPool> memory_pool,
+ std::shared_ptr<RealtimeDeduplicateState> deduplicate_state,
+ std::map<std::string, std::string> options)
+ : arrow_pool_(GetArrowPool(memory_pool)),
+ realtime_store_(std::move(realtime_store)),
+ file_writer_(std::move(file_writer)),
+ schema_layout_(std::move(schema_layout)),
+ deduplicate_key_field_(std::move(deduplicate_key_field)),
+ data_file_path_factory_(std::move(data_file_path_factory)),
+ file_system_(std::move(file_system)),
+ dv_maintainer_(std::move(dv_maintainer)),
+ memory_pool_(std::move(memory_pool)),
+ deduplicate_state_(std::move(deduplicate_state)),
+ options_(std::move(options)),
+ next_offset_(next_offset) {}
+
+Status RealtimeDeduplicateWriter::Write(std::unique_ptr<RecordBatch>&& batch) {
+ if (!batch || !batch->GetData()) {
+ return Status::Invalid("deduplicate real-time write batch is null");
+ }
+ if (batch->GetData()->length == 0) {
+ return Status::OK();
+ }
+ std::vector<RecordBatch::RowKind> row_kinds = batch->GetRowKind();
+ if (row_kinds.empty()) {
+ row_kinds.assign(batch->GetData()->length,
RecordBatch::RowKind::INSERT);
+ }
+ if (row_kinds.size() != static_cast<size_t>(batch->GetData()->length)) {
+ return Status::Invalid("deduplicate row-kind count does not match
batch row count");
+ }
+
+ std::lock_guard<std::mutex> lock(realtime_store_mutex_);
+ PAIMON_ASSIGN_OR_RAISE(RealtimeOffsetUtils::ValidatedBatch validated,
+ RealtimeOffsetUtils::ValidateBatch(
+ batch.get(), schema_layout_->InputSchema(),
next_offset_));
+ const int32_t key_position =
+
schema_layout_->StoreWriteSchema()->GetFieldIndex(deduplicate_key_field_->name());
+ if (key_position < 0) {
+ return Status::Invalid("deduplicate key field is missing from write
data");
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::StructArray> keys,
+ arrow::StructArray::Make({validated.data->field(key_position)},
{deduplicate_key_field_}));
+
+ // Pin the complete lookup under the short store lock, then perform
sidecar lookup without
+ // holding it. The writer mutex keeps application writes ordered while
Advance may proceed.
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<const KeyOffsetLookup> lookup,
+ realtime_store_->AcquireKeyOffsetLookup());
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 previous_offsets,
lookup->LookupOffsets(keys));
+ PAIMON_RETURN_NOT_OK(realtime_store_->WriteAndLookup(validated.data,
row_kinds,
+
validated.offset_range, previous_offsets));
Review Comment:
[P2] Each write keeps a building-index snapshot alive while mutating that
index
`lookup` remains alive during `WriteAndLookup()`, and
`AcquireKeyOffsetLookup()` includes a snapshot sharing the building index's
map. The first `MutableKeyOffsetIndex::Add()` or `Erase()` in each non-empty
write therefore sees `!offsets_.unique()` and clones the entire accumulated
building map, even without concurrent readers. Small batches over a growing
segment repeatedly copy all prior keys; one-row inserts can accumulate
quadratic copying and large transient allocations.
--
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]
