luoyuxia commented on code in PR #199:
URL: https://github.com/apache/paimon-cpp/pull/199#discussion_r3795419714
##########
src/paimon/core/table/source/realtime_split.h:
##########
@@ -59,25 +74,27 @@ class RealtimeSplit : public Split {
return disk_splits_;
}
- const std::shared_ptr<MemIndexer>& Indexer() const {
- return indexer_;
+ int64_t CommittedOffset() const {
+ return committed_offset_;
}
- const std::shared_ptr<MemReadView>& ReadView() const {
- return read_view_;
+ int64_t MemoryUpperOffset() const {
+ return memory_upper_offset_;
}
- int64_t CommittedOffset() const {
- return committed_offset_;
+ const std::string& OpaqueTicket() const {
+ return opaque_ticket_;
}
private:
+ int32_t version_;
+ std::optional<int64_t> snapshot_id_;
std::map<std::string, std::string> partition_;
int32_t bucket_;
std::vector<std::shared_ptr<Split>> disk_splits_;
- std::shared_ptr<MemIndexer> indexer_;
- std::shared_ptr<MemReadView> read_view_;
int64_t committed_offset_;
Review Comment:
Would it be more intuitive to represent committed progress as an exclusive
end offset? The committed snapshot covers offsets < committed_end_offset, and
the reader starts from committed_end_offset inclusively. This avoids the extra
+1 conversion, represents an empty committed range naturally with 0, and makes
adjacent ranges compose cleanly as half-open intervals.
Correspondingly, could memory_upper_offset be represented as an exclusive
memory_end_offset? At the RealtimeSplit level, the in-memory range could then
be expressed directly as [start_offset, end_offset): start_offset equals
committed_end_offset, and end_offset equals memory_end_offset. The reader
consumes exactly that range, which makes both boundaries explicit and avoids
mixing inclusive committed_offset and memory_upper_offset semantics.
##########
include/paimon/realtime/realtime_context.h:
##########
@@ -66,29 +65,12 @@ struct PAIMON_EXPORT RealtimePartitionBucket {
/// Largest committed offset for each partition-bucket.
using RealtimeOffsetMap = std::map<RealtimePartitionBucket, int64_t>;
-/// Memory indexer and its initial offset resolved from committed and retained
memory progress.
-struct PAIMON_EXPORT RealtimeMemIndexerState {
- /// Plugin instance associated with the requested partition-bucket.
- std::shared_ptr<MemIndexer> indexer;
- /// First offset after both committed rows and rows currently retained by
the indexer.
- int64_t initial_offset;
-};
-
-/// One partition-bucket and the immutable plugin view captured for a table
scan.
-struct PAIMON_EXPORT RealtimePartitionBucketView {
- /// Partition-bucket associated with this view.
- RealtimePartitionBucket partition_bucket;
- /// Plugin instance that creates readers from `read_view`.
- std::shared_ptr<MemIndexer> indexer;
- /// Immutable rows pinned for one query plan.
- std::shared_ptr<MemReadView> read_view;
-};
-
/// Shared context that owns the `MemIndexer` instances used by a real-time
writer.
///
-/// Applications share one context between `WriteContext` and `ScanContext`.
The context uses
-/// either the default Arrow implementation or an application-provided factory
and keeps each
-/// created indexer available across writes, prepare-commit operations, and
process-local reads.
+/// Applications share one context between `WriteContext`, `ScanContext`, and
`ReadContext`. The
+/// context uses either the default Arrow implementation or an
application-provided factory and
+/// keeps each created indexer available across writes, prepare-commit
operations, and
+/// process-local reads.
class PAIMON_EXPORT RealtimeContext {
Review Comment:
The current RealtimeContext hierarchy seems unclear.
RealtimeContext::Create() constructs a RealtimeContextImpl and upcasts it to
the public base type, while the write, scan, and read paths immediately
downcast it through RealtimeContextImpl::Cast() before performing any operation.
As a result, the effective dependency is still RealtimeContextImpl;
RealtimeContext only transports the concrete instance and does not define a
behavioral contract. If only one implementation is intended, a concrete
RealtimeContext with Pimpl would make the call hierarchy clearer. Otherwise,
the operations required by the write, scan, and read paths should be expressed
on the shared abstraction and invoked without downcasting.
The name RealtimeContext also seems misleading for its current role. Unlike
WriteContext, ScanContext, or ReadContext, this object is not an
operation-scoped collection of configuration and dependencies. It is a
long-lived, stateful object shared by the writer, scanner, and reader, and it
owns the mem-indexer registry, committed progress, pinned read views, and
their lifecycle. Would RealtimeStore better describe this ownership model and
make the relationship with writer, scan, and read more explicit?
This structure also makes it difficult to support another realtime-store
implementation. The current extension point is MemIndexerFactory, which can
replace the per-partition-bucket memory indexer, but it cannot replace the
writer, scan/planning, split, and reader behavior of the realtime store as a
whole. A different RealtimeContext implementation cannot be used through the
existing APIs because all write, scan, and read paths eventually require
RealtimeContextImpl::Cast(). Supporting another store would therefore require
modifying these core call sites and adding implementation-specific branches.
Would it be clearer to make RealtimeStore the behavioral boundary and
introduce a RealtimeStoreFactory that creates the complete store
implementation? The existing mem-indexer-based implementation could remain one
RealtimeStore, while another store could provide its own writer, planner,
split, and reader behavior without changing the core dispatch path.
##########
include/paimon/realtime/realtime_context.h:
##########
@@ -66,29 +65,12 @@ struct PAIMON_EXPORT RealtimePartitionBucket {
/// Largest committed offset for each partition-bucket.
using RealtimeOffsetMap = std::map<RealtimePartitionBucket, int64_t>;
-/// Memory indexer and its initial offset resolved from committed and retained
memory progress.
-struct PAIMON_EXPORT RealtimeMemIndexerState {
- /// Plugin instance associated with the requested partition-bucket.
- std::shared_ptr<MemIndexer> indexer;
- /// First offset after both committed rows and rows currently retained by
the indexer.
- int64_t initial_offset;
-};
-
-/// One partition-bucket and the immutable plugin view captured for a table
scan.
-struct PAIMON_EXPORT RealtimePartitionBucketView {
- /// Partition-bucket associated with this view.
- RealtimePartitionBucket partition_bucket;
- /// Plugin instance that creates readers from `read_view`.
- std::shared_ptr<MemIndexer> indexer;
- /// Immutable rows pinned for one query plan.
- std::shared_ptr<MemReadView> read_view;
-};
-
/// Shared context that owns the `MemIndexer` instances used by a real-time
writer.
///
-/// Applications share one context between `WriteContext` and `ScanContext`.
The context uses
-/// either the default Arrow implementation or an application-provided factory
and keeps each
-/// created indexer available across writes, prepare-commit operations, and
process-local reads.
+/// Applications share one context between `WriteContext`, `ScanContext`, and
`ReadContext`. The
+/// context uses either the default Arrow implementation or an
application-provided factory and
+/// keeps each created indexer available across writes, prepare-commit
operations, and
+/// process-local reads.
class PAIMON_EXPORT RealtimeContext {
Review Comment:
I am not fully sure whether the following API shape covers all of your
intended use cases, so please treat it as a design sketch for discussion.
The overall object model is inspired by Paimon’s Java CatalogFactory →
Catalog → Table creation model. This is an object-creation and ownership
hierarchy rather than inheritance between these three types: a factory creates
a backend-scoped store, the store loads a table-specific handle, and the table
handle creates or wraps the write, scan, and read operations.
Please note the follow api proposal is is roughly the API shape I have in
mind. I have not examined every related code path in detail, so the concrete
class names and method signatures may not be completely accurate. The main
point is the overall creation and composition hierarchy:
```
RealtimeStoreFactory
└── creates RealtimeStore
└── loads RealtimeTable
├── creates write operations
├── creates scan operations
└── creates read operations
```
Concrete backends implement the corresponding factory, store, and table
interfaces. The exact API may differ, but this is roughly the abstraction
boundary and dependency direction I have in mind.
## Propose API
```
/// Store-level configuration shared by all tables opened from one realtime
store.
struct PAIMON_EXPORT RealtimeStoreOptions {
std::map<std::string, std::string> options;
xxx
};
/// Identifies and configures one table loaded from a RealtimeStore.
struct PAIMON_EXPORT RealtimeTableDescriptor {
Identifier identifier;
std::string table_path;
std::string branch;
std::shared_ptr<Schema> schema;
std::map<std::string, std::string> options;
};
/// Factory for creating a backend-level RealtimeStore.
///
/// Implementations may be registered through the existing Paimon Factory
/// mechanism using a unique identifier.
class PAIMON_EXPORT RealtimeStoreFactory : public Factory {
public:
~RealtimeStoreFactory() override = default;
/// Creates a RealtimeStore using this factory.
virtual Result<std::shared_ptr<RealtimeStore>> Create(
const RealtimeStoreOptions& options) const = 0;
};
/// A long-lived realtime backend that may serve multiple tables.
///
/// Implementations may own backend-wide resources. Table-specific state
/// should be represented by RealtimeTable.
class PAIMON_EXPORT RealtimeStore {
public:
virtual ~RealtimeStore() = default;
/// Loads or opens a table from this realtime store.
///
/// The returned table may retain shared backend resources owned by
this
/// store. Implementations may cache table state internally.
virtual Result<std::shared_ptr<RealtimeTable>> LoadTable(
const RealtimeTableDescriptor& table) = 0;
};
/// A table-specific realtime handle.
///
/// The same table handle may be shared by write, scan, and read
operations.
/// An in-process implementation may keep table-level indexers, committed
/// progress, and query views here. A remote implementation may keep only
/// table metadata and backend clients.
class PAIMON_EXPORT RealtimeTable {
public:
virtual ~RealtimeTable() = default;
/// Creates the write path for this realtime table.
///
/// The returned object follows the existing FileStoreWrite contract,
/// including prepare-commit and committed-snapshot notification.
virtual Result<std::unique_ptr<FileStoreWrite>> CreateWrite(
std::unique_ptr<WriteContext> context) = 0;
/// Creates the scan/planning path for this realtime table.
///
/// CreatePlan() may combine the committed disk snapshot with the
/// realtime portion and return implementation-specific Split objects.
virtual Result<std::unique_ptr<TableScan>> CreateScan(
std::unique_ptr<ScanContext> context) = 0;
/// Creates the read path for splits planned by CreateScan().
///
/// The implementation is responsible for recognizing its own realtime
/// splits and composing disk and realtime readers as required.
virtual Result<std::unique_ptr<TableRead>> CreateRead(
std::unique_ptr<ReadContext> context) = 0;
};
```
## Example usage
```
RealtimeStoreOptions store_options;
store_options.options = options;
PAIMON_ASSIGN_OR_RAISE(
std::shared_ptr<RealtimeStore> realtime_store,
RealtimeStoreFactory::Create(realtime_store_type, store_options));
RealtimeTableDescriptor table_descriptor{
identifier,
table_path,
branch,
schema,
table_options};
PAIMON_ASSIGN_OR_RAISE(
std::shared_ptr<RealtimeTable> realtime_table,
realtime_store->LoadTable(table_descriptor));
Create a writer:
WriteContextBuilder write_builder(table_path, commit_user);
write_builder.SetOptions(write_options);
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<WriteContext> write_context,
write_builder.Finish());
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<FileStoreWrite> writer,
realtime_table->CreateWrite(std::move(write_context)));
Create a two-stage scan and read:
ScanContextBuilder scan_builder(table_path);
scan_builder.SetOptions(scan_options);
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<ScanContext> scan_context,
scan_builder.Finish());
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<TableScan> scan,
realtime_table->CreateScan(std::move(scan_context)));
PAIMON_ASSIGN_OR_RAISE(
std::shared_ptr<Plan> plan,
scan->CreatePlan());
ReadContextBuilder read_builder(table_path);
read_builder.SetOptions(read_options);
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<ReadContext> read_context,
read_builder.Finish());
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<TableRead> table_read,
realtime_table->CreateRead(std::move(read_context)));
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<BatchReader> reader,
table_read->CreateReader(plan->Splits()));
Possible implementation hierarchy:
RealtimeStoreFactory
├── MemoryRealtimeStoreFactory
└── OtherRealtimeStoreFactory
RealtimeStore
├── MemoryRealtimeStore
│ └── LoadTable() → MemoryRealtimeTable
└── OtherRealtimeStore
└── LoadTable() → OtherRealtimeTable
RealtimeTable
├── MemoryRealtimeTable
│ ├── CreateWrite() → mem-indexer-backed FileStoreWrite
│ ├── CreateScan() → memory union TableScan
│ └── CreateRead() → memory union TableRead
└── OtherRealtimeTable
├── CreateWrite() → backend-specific FileStoreWrite
├── CreateScan() → backend-specific TableScan
└── CreateRead() → backend-specific TableRead
```
With this structure, the current mem-indexer implementation could become
MemoryRealtimeStore and MemoryRealtimeTable. MemIndexerFactory would remain an
implementation detail of that backend.
The write, scan, and read paths would invoke the RealtimeTable interface
directly. They would no longer upcast a concrete implementation to
RealtimeContext and later recover it through RealtimeContextImpl::Cast(). This
proposal assumes that every realtime-table implementation exposed through
paimon-cpp can conform to the existing FileStoreWrite, TableScan, and TableRead
contracts. If that assumption does not match the intended writer or commit
lifecycle, the operation interfaces may need to be separated further.
--
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]
