HaHaJeff commented on code in PR #224:
URL: https://github.com/apache/paimon-cpp/pull/224#discussion_r3855831588
##########
include/paimon/realtime/realtime_store.h:
##########
@@ -41,10 +43,42 @@ namespace paimon {
class MemoryPool;
class Predicate;
-/// A table record batch and its framework-assigned contiguous offset range.
+struct PAIMON_EXPORT AppendRealtimeStoreCreateConfig {
+ StatisticsMode statistics_mode;
+};
+
+struct PAIMON_EXPORT PrimaryKeyRealtimeStoreCreateConfig {
+ /// Primary-key fields after removing partition fields, in comparison
order.
+ std::vector<std::string> trimmed_primary_keys;
+};
+
+using RealtimeStoreCreateConfig =
+ std::variant<AppendRealtimeStoreCreateConfig,
PrimaryKeyRealtimeStoreCreateConfig>;
+
+/// Parameters used by a `RealtimeStoreFactory` to create a store.
+struct PAIMON_EXPORT RealtimeStoreCreateRequest {
+ /// Schema whose ownership is transferred to the factory. Append mode
receives the complete
+ /// table write schema. Primary-key mode receives the prepared transport
schema:
+ /// [_VALUE_KIND, _SEQUENCE_NUMBER, _REALTIME_OFFSET, table write fields].
+ std::unique_ptr<::ArrowSchema> write_schema;
+ /// Table options available to the store implementation.
+ std::map<std::string, std::string> options;
+ /// Memory pool for allocations retained by the store.
+ std::shared_ptr<MemoryPool> memory_pool;
+ /// Partition values identifying the store.
+ std::map<std::string, std::string> partition;
+ /// Bucket identifying the store within its partition.
+ int32_t bucket = -1;
+ /// Mode-specific store configuration.
+ RealtimeStoreCreateConfig mode_config;
Review Comment:
Fixed in `13cba67b9ef802f9acebe6b0b997d1e44dc528bf`: partition and bucket
were removed from `RealtimeStoreCreateRequest`, and a separate
`RealtimePartitionBucket` is now passed to `GetOrCreateRealtimeStore`.
##########
src/paimon/common/table/special_fields.h:
##########
Review Comment:
Fixed in `13cba67b9ef802f9acebe6b0b997d1e44dc528bf`: `_REALTIME_OFFSET` is
now included in `SpecialFields::IsSystemField`, with test coverage.
##########
src/paimon/core/mergetree/merge_tree_writer.cpp:
##########
@@ -154,6 +154,59 @@ Status
MergeTreeWriter::Write(std::unique_ptr<RecordBatch>&& moved_batch) {
return Status::OK();
}
+Status MergeTreeWriter::WriteSortedReaders(
+ std::vector<std::unique_ptr<KeyValueRecordReader>>&& readers) {
Review Comment:
Fixed in `13cba67b9ef802f9acebe6b0b997d1e44dc528bf`: the method was renamed
to `WriteSortedReadersToFiles`.
##########
src/paimon/core/operation/file_store_write.cpp:
##########
@@ -197,7 +198,26 @@ Result<std::unique_ptr<FileStoreWrite>>
FileStoreWrite::Create(std::unique_ptr<W
} else {
// pk table
if (ctx->GetRealtimeContext()) {
- return Status::Invalid("real-time write currently supports append
tables only");
+ PAIMON_RETURN_NOT_OK(ValidatePrimaryKeyRealtimeOptions(options,
*schema));
+ if (ignore_previous_files) {
+ return Status::NotImplemented(
+ "PK realtime v1 requires restore from the latest
snapshot");
+ }
+ if (!ctx->GetWriteSchema().empty()) {
+ return Status::NotImplemented(
+ "PK realtime v1 does not support a custom write schema");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> latest_snapshot,
+ snapshot_manager->LatestSnapshot());
+ if (latest_snapshot) {
+ PAIMON_ASSIGN_OR_RAISE(RealtimeOffsetMap
realtime_committed_offsets,
+ RealtimeCommitProperties::ReadOffsets(
+ latest_snapshot,
options.GetFileSystem()));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RealtimeContextImpl>
realtime_context_impl,
+
RealtimeContextImpl::Cast(ctx->GetRealtimeContext()));
+
PAIMON_RETURN_NOT_OK(realtime_context_impl->AdvanceCommittedProgress(
+ latest_snapshot->Id(), realtime_committed_offsets));
+ }
Review Comment:
Fixed in `13cba67b9ef802f9acebe6b0b997d1e44dc528bf`: append and primary-key
writer creation now share `RestoreRealtimeCommittedProgress`.
--
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]
