This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 21228d92 feat(realtime): support spilling sealed segments (#327)
21228d92 is described below
commit 21228d92b30cc7e987239f0dbca4371120bdfb0b
Author: lxy <[email protected]>
AuthorDate: Tue Sep 15 10:29:27 2026 +0800
feat(realtime): support spilling sealed segments (#327)
---
include/paimon/defs.h | 4 +
include/paimon/file_store_write.h | 5 +-
include/paimon/realtime/realtime_store.h | 16 +-
include/paimon/write_context.h | 3 +-
src/paimon/CMakeLists.txt | 1 +
src/paimon/common/defs.cpp | 1 +
src/paimon/core/core_options.cpp | 7 +
src/paimon/core/core_options.h | 1 +
src/paimon/core/core_options_test.cpp | 3 +
src/paimon/core/io/arrow_ipc_file.cpp | 197 +++++++
src/paimon/core/io/arrow_ipc_file.h | 110 ++++
src/paimon/core/mergetree/spill_reader.cpp | 64 +--
src/paimon/core/mergetree/spill_reader.h | 24 +-
src/paimon/core/mergetree/spill_writer.cpp | 90 +--
src/paimon/core/mergetree/spill_writer.h | 33 +-
.../operation/append_only_file_store_write.cpp | 4 +-
.../core/operation/key_value_file_store_write.cpp | 13 +-
src/paimon/core/realtime/arrow_realtime_store.cpp | 611 +++++++++++++++++----
src/paimon/core/realtime/arrow_realtime_store.h | 26 +-
.../core/realtime/arrow_realtime_store_factory.cpp | 20 +-
.../core/realtime/arrow_realtime_store_test.cpp | 151 ++++-
.../realtime/primary_key_realtime_store_test.cpp | 69 ++-
.../core/realtime/realtime_append_only_writer.cpp | 6 +-
.../core/realtime/realtime_append_only_writer.h | 2 +-
src/paimon/core/realtime/realtime_context_impl.cpp | 10 +-
src/paimon/core/realtime/realtime_context_impl.h | 1 +
src/paimon/core/realtime/realtime_context_test.cpp | 26 +-
.../core/table/source/append_only_table_read.cpp | 3 +-
.../core/table/source/key_value_table_read.cpp | 3 +-
test/inte/realtime_write_inte_test.cpp | 218 ++++++++
30 files changed, 1421 insertions(+), 301 deletions(-)
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index 8ecda26a..cfa08fa0 100644
--- a/include/paimon/defs.h
+++ b/include/paimon/defs.h
@@ -612,6 +612,10 @@ struct PAIMON_EXPORT Options {
/// Default value is "false".
static const char REALTIME_ENABLED[];
+ /// "realtime.spill-enabled" - Whether the default real-time store spills
sealed segments to
+ /// the local temporary directory. Default value is "false".
+ static const char REALTIME_SPILL_ENABLED[];
+
/// "realtime.read-view-ttl" - Lifetime of a real-time memory view pinned
by scan planning
/// before reader creation. Default value is "5 min".
static const char REALTIME_READ_VIEW_TTL[];
diff --git a/include/paimon/file_store_write.h
b/include/paimon/file_store_write.h
index 5bc2fe73..71ca910b 100644
--- a/include/paimon/file_store_write.h
+++ b/include/paimon/file_store_write.h
@@ -61,8 +61,9 @@ class PAIMON_EXPORT FileStoreWrite {
virtual Status Write(std::unique_ptr<RecordBatch>&& batch) = 0;
/// Slices the current in-memory real-time data into a sealed segment so
it can be reclaimed
- /// independently. A future implementation will support spilling sealed
segments to a
- /// temporary directory; currently this method only creates the in-memory
segment boundary.
+ /// independently. When `realtime.spill-enabled` is true and the write
context has a temporary
+ /// directory, the default real-time store synchronously spills each
non-empty sealed segment
+ /// to a local Arrow IPC file.
/// Calling this method on a non-real-time writer returns an error.
/// If sealing fails, the caller must recreate both the `RealtimeContext`
and writer. The
/// upstream must then recover input from the durable recovery offset
persisted in the
diff --git a/include/paimon/realtime/realtime_store.h
b/include/paimon/realtime/realtime_store.h
index 6cc1b085..c15bc26a 100644
--- a/include/paimon/realtime/realtime_store.h
+++ b/include/paimon/realtime/realtime_store.h
@@ -39,6 +39,7 @@ struct ArrowSchema;
namespace paimon {
+class FileSystem;
class MemoryPool;
class Predicate;
@@ -62,6 +63,12 @@ struct PAIMON_EXPORT RealtimeStoreCreateRequest {
RealtimeStoreMode mode = RealtimeStoreMode::APPEND_ONLY;
/// Statistics collected by the store for query pruning.
StatisticsMode statistics_mode = StatisticsMode::NONE;
+ /// Directory used by the default store for immutable spill files when
+ /// `realtime.spill-enabled` is true. Custom stores may ignore this hint.
+ std::string temp_directory = "";
+ /// File system used to access `temp_directory`. The default store retains
shared ownership
+ /// for its lifetime. Custom stores may ignore this hint.
+ std::shared_ptr<FileSystem> file_system = nullptr;
};
/// A record batch and its application-assigned offset bounds.
@@ -131,6 +138,8 @@ struct PAIMON_EXPORT RealtimeQueryContext {
/// Optional predicate using field indexes from `read_schema`. A non-null
predicate allows the
/// plugin to prune candidate rows. Exact filtering is applied by the
Paimon read framework.
std::shared_ptr<Predicate> predicate;
+ /// Maximum number of rows in a returned query batch.
+ int32_t read_batch_size = 1024;
};
/// Customizable plugin interface for storing and querying real-time rows
before Paimon data-file
@@ -162,6 +171,8 @@ class PAIMON_EXPORT RealtimeStore {
/// preserve write order and contain `_REALTIME_OFFSET` followed by the
table write fields.
/// Primary-key readers contain the real-time primary-key store fields;
each reader's complete
/// stream is sorted by full primary key then sequence number.
+ /// Returned readers have independent mutable read state and may be
operated concurrently with
+ /// one another without external synchronization.
virtual Result<std::vector<std::unique_ptr<BatchReader>>>
CreateCommitReaders(
const std::shared_ptr<RealtimeSegmentHandle>& segment) = 0;
@@ -173,7 +184,10 @@ class PAIMON_EXPORT RealtimeStore {
/// Creates readers over rows in `view`. The readers collectively expose
every candidate row
/// exactly once. Primary-key reader streams are sorted by full primary
key then sequence
- /// number. Paimon retains `view` for the lifetime of the resulting
framework reader.
+ /// number. Every returned batch contains at most
`context.read_batch_size` rows. Paimon retains
+ /// `view` for the lifetime of the resulting framework reader.
+ /// Returned readers have independent mutable read state and may be
operated concurrently with
+ /// one another without external synchronization.
virtual Result<std::vector<std::unique_ptr<BatchReader>>>
CreateQueryReaders(
const std::shared_ptr<RealtimeReadView>& view, const
RealtimeQueryContext& context) = 0;
diff --git a/include/paimon/write_context.h b/include/paimon/write_context.h
index ff7245ee..1a6e5271 100644
--- a/include/paimon/write_context.h
+++ b/include/paimon/write_context.h
@@ -222,7 +222,8 @@ class PAIMON_EXPORT WriteContextBuilder {
/// @return Reference to this builder for method chaining.
WriteContextBuilder& WithExecutor(const std::shared_ptr<Executor>&
executor);
- /// Set the temporary directory path for IO operations (lookup and
external disk spill).
+ /// Set the temporary directory path for IO operations (lookup, external
disk spill, and
+ /// real-time spill).
/// @param temp_dir The temporary directory path.
/// @return Reference to this builder for method chaining.
WriteContextBuilder& WithTempDirectory(const std::string& temp_dir);
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index afea66bd..803bdb2b 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -212,6 +212,7 @@ endif()
set(PAIMON_CORE_SRCS
core/disk/file_io_channel.cpp
+ core/io/arrow_ipc_file.cpp
core/mergetree/spill_reader.cpp
core/mergetree/spill_writer.cpp
core/append/append_only_writer.cpp
diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp
index 619ce848..de1d4577 100644
--- a/src/paimon/common/defs.cpp
+++ b/src/paimon/common/defs.cpp
@@ -164,6 +164,7 @@ const char Options::TABLE_READ_SEQUENCE_NUMBER_ENABLED[] =
"table-read.sequence-
const char Options::KEY_VALUE_SEQUENCE_NUMBER_ENABLED[] =
"key-value.sequence_number.enabled";
const char Options::SCAN_TIMESTAMP_MILLIS[] = "scan.timestamp-millis";
const char Options::REALTIME_ENABLED[] = "realtime.enabled";
+const char Options::REALTIME_SPILL_ENABLED[] = "realtime.spill-enabled";
const char Options::REALTIME_READ_VIEW_TTL[] = "realtime.read-view-ttl";
const char Options::REALTIME_STORE_STATS_MODE[] = "realtime.store.stats-mode";
const char Options::SCAN_TIMESTAMP[] = "scan.timestamp";
diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp
index ffb4fffe..37fadc8e 100644
--- a/src/paimon/core/core_options.cpp
+++ b/src/paimon/core/core_options.cpp
@@ -435,6 +435,7 @@ struct CoreOptions::Impl {
std::optional<int32_t> global_index_thread_num;
bool realtime_enabled = false;
+ bool realtime_spill_enabled = false;
bool scan_manifest_entry_lazy_decode_enabled = true;
bool ignore_delete = false;
bool manifest_delete_file_drop_stats = false;
@@ -847,6 +848,8 @@ struct CoreOptions::Impl {
// Parse real-time write and read configurations.
Status ParseRealtimeOptions(const ConfigParser& parser) {
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::REALTIME_ENABLED,
&realtime_enabled));
+ PAIMON_RETURN_NOT_OK(
+ parser.Parse<bool>(Options::REALTIME_SPILL_ENABLED,
&realtime_spill_enabled));
PAIMON_RETURN_NOT_OK(parser.ParseTimeDuration(Options::REALTIME_READ_VIEW_TTL,
&realtime_read_view_ttl_millis));
if (realtime_read_view_ttl_millis <= 0) {
@@ -1186,6 +1189,10 @@ bool CoreOptions::RealtimeEnabled() const {
return impl_->realtime_enabled;
}
+bool CoreOptions::RealtimeSpillEnabled() const {
+ return impl_->realtime_spill_enabled;
+}
+
int64_t CoreOptions::GetRealtimeReadViewTtlMillis() const {
return impl_->realtime_read_view_ttl_millis;
}
diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h
index 5c2c5e45..86771e6a 100644
--- a/src/paimon/core/core_options.h
+++ b/src/paimon/core/core_options.h
@@ -108,6 +108,7 @@ class PAIMON_EXPORT CoreOptions {
std::optional<int64_t> GetScanSnapshotId() const;
std::optional<int64_t> GetScanTimestampMillis() const;
bool RealtimeEnabled() const;
+ bool RealtimeSpillEnabled() const;
int64_t GetRealtimeReadViewTtlMillis() const;
/// Returns the statistics mode used by the real-time store.
StatisticsMode GetRealtimeStoreStatisticsMode() const;
diff --git a/src/paimon/core/core_options_test.cpp
b/src/paimon/core/core_options_test.cpp
index 165c2661..3e489a4c 100644
--- a/src/paimon/core/core_options_test.cpp
+++ b/src/paimon/core/core_options_test.cpp
@@ -58,6 +58,7 @@ TEST(CoreOptionsTest, TestDefaultValue) {
ASSERT_EQ(std::nullopt, core_options.GetScanSnapshotId());
ASSERT_EQ(5 * 60 * 1000, core_options.GetRealtimeReadViewTtlMillis());
ASSERT_FALSE(core_options.RealtimeEnabled());
+ ASSERT_FALSE(core_options.RealtimeSpillEnabled());
ASSERT_EQ(StatisticsMode::NONE,
core_options.GetRealtimeStoreStatisticsMode());
ASSERT_EQ("zstd", core_options.GetFileCompression());
ASSERT_EQ(std::nullopt, core_options.GetChangelogFileCompression());
@@ -325,6 +326,7 @@ TEST(CoreOptionsTest, TestFromMap) {
{Options::TABLE_READ_SEQUENCE_NUMBER_ENABLED, "true"},
{Options::KEY_VALUE_SEQUENCE_NUMBER_ENABLED, "true"},
{Options::REALTIME_ENABLED, "true"},
+ {Options::REALTIME_SPILL_ENABLED, "true"},
{Options::BUCKET_FUNCTION_TYPE, "mod"},
{"fields.metrics.map.storage-layout", "shared-shredding"},
{"fields.metrics.map.shared-shredding.max-columns", "128"},
@@ -494,6 +496,7 @@ TEST(CoreOptionsTest, TestFromMap) {
ASSERT_TRUE(core_options.TableReadSequenceNumberEnabled());
ASSERT_TRUE(core_options.KeyValueSequenceNumberEnabled());
ASSERT_TRUE(core_options.RealtimeEnabled());
+ ASSERT_TRUE(core_options.RealtimeSpillEnabled());
ASSERT_TRUE(core_options.LookupRemoteFileEnabled());
ASSERT_EQ(core_options.GetLookupRemoteLevelThreshold(), 2);
ASSERT_EQ(BucketFunctionType::MOD, core_options.GetBucketFunctionType());
diff --git a/src/paimon/core/io/arrow_ipc_file.cpp
b/src/paimon/core/io/arrow_ipc_file.cpp
new file mode 100644
index 00000000..c780f040
--- /dev/null
+++ b/src/paimon/core/io/arrow_ipc_file.cpp
@@ -0,0 +1,197 @@
+/*
+ * 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/io/arrow_ipc_file.h"
+
+#include <utility>
+
+#include "arrow/ipc/api.h"
+#include "arrow/util/compression.h"
+#include "paimon/common/utils/arrow/arrow_input_stream_adapter.h"
+#include "paimon/common/utils/arrow/arrow_output_stream_adapter.h"
+#include "paimon/common/utils/arrow/arrow_utils.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/macros.h"
+
+namespace paimon {
+
+ArrowIpcFileWriter::ArrowIpcFileWriter(const std::shared_ptr<FileSystem>& fs,
std::string path,
+ const std::shared_ptr<arrow::Schema>&
schema,
+ std::string compression, int32_t
compression_level,
+ bool use_threads,
+ const
std::shared_ptr<arrow::MemoryPool>& arrow_pool)
+ : fs_(fs),
+ path_(std::move(path)),
+ schema_(schema),
+ compression_(std::move(compression)),
+ compression_level_(compression_level),
+ use_threads_(use_threads),
+ arrow_pool_(arrow_pool) {}
+
+ArrowIpcFileWriter::~ArrowIpcFileWriter() {
+ [[maybe_unused]] Status status = Close();
+}
+
+Result<std::unique_ptr<ArrowIpcFileWriter>> ArrowIpcFileWriter::Create(
+ const std::shared_ptr<FileSystem>& fs, const std::string& path,
+ const std::shared_ptr<arrow::Schema>& schema, const std::string&
compression,
+ int32_t compression_level, bool use_threads,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool) {
+ if (!fs || !schema || !arrow_pool || path.empty()) {
+ return Status::Invalid("invalid Arrow IPC file writer arguments");
+ }
+ std::unique_ptr<ArrowIpcFileWriter> writer(new ArrowIpcFileWriter(
+ fs, path, schema, compression, compression_level, use_threads,
arrow_pool));
+ PAIMON_RETURN_NOT_OK(writer->Open());
+ return writer;
+}
+
+Status ArrowIpcFileWriter::Open() {
+ auto cleanup_guard = ScopeGuard([&]() {
+ arrow_writer_.reset();
+ arrow_output_stream_adapter_.reset();
+ if (out_stream_) {
+ [[maybe_unused]] Status status = out_stream_->Close();
+ out_stream_.reset();
+ }
+ [[maybe_unused]] Status status = fs_->Delete(path_);
+ });
+ auto write_options = arrow::ipc::IpcWriteOptions::Defaults();
+ write_options.memory_pool = arrow_pool_.get();
+ write_options.use_threads = use_threads_;
+ PAIMON_ASSIGN_OR_RAISE(arrow::Compression::type arrow_compression,
+ ArrowUtils::GetCompressionType(compression_));
+ if (!arrow::util::Codec::SupportsCompressionLevel(arrow_compression)) {
+ compression_level_ = arrow::util::Codec::UseDefaultCompressionLevel();
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ write_options.codec, arrow::util::Codec::Create(arrow_compression,
compression_level_));
+ PAIMON_ASSIGN_OR_RAISE(out_stream_, fs_->Create(path_,
/*overwrite=*/false));
+ arrow_output_stream_adapter_ =
std::make_shared<ArrowOutputStreamAdapter>(out_stream_);
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ arrow_writer_,
+ arrow::ipc::MakeFileWriter(arrow_output_stream_adapter_, schema_,
write_options));
+ cleanup_guard.Release();
+ return Status::OK();
+}
+
+Status ArrowIpcFileWriter::WriteBatch(const
std::shared_ptr<arrow::RecordBatch>& batch) {
+ if (closed_ || !arrow_writer_) {
+ return Status::Invalid("Arrow IPC file writer is closed");
+ }
+ if (!batch) {
+ return Status::Invalid("Arrow IPC record batch is null");
+ }
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow_writer_->WriteRecordBatch(*batch));
+ return Status::OK();
+}
+
+Status ArrowIpcFileWriter::Close() {
+ if (closed_) {
+ return Status::OK();
+ }
+ if (arrow_writer_) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow_writer_->Close());
+ arrow_writer_.reset();
+ }
+ if (out_stream_) {
+ PAIMON_RETURN_NOT_OK(out_stream_->Close());
+ out_stream_.reset();
+ }
+ closed_ = true;
+ return Status::OK();
+}
+
+Result<int64_t> ArrowIpcFileWriter::GetFileSize() const {
+ if (!closed_ && arrow_output_stream_adapter_) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(int64_t size,
arrow_output_stream_adapter_->Tell());
+ return size;
+ }
+ PAIMON_ASSIGN_OR_RAISE(FileStatus status, fs_->GetFileStatus(path_));
+ return status.GetLen();
+}
+
+ArrowIpcFileReader::ArrowIpcFileReader(const std::shared_ptr<FileSystem>& fs,
std::string path,
+ bool use_threads,
+ const
std::shared_ptr<arrow::MemoryPool>& arrow_pool)
+ : fs_(fs), path_(std::move(path)), use_threads_(use_threads),
arrow_pool_(arrow_pool) {}
+
+ArrowIpcFileReader::~ArrowIpcFileReader() {
+ Close();
+}
+
+Result<std::unique_ptr<ArrowIpcFileReader>> ArrowIpcFileReader::Open(
+ const std::shared_ptr<FileSystem>& fs, const std::string& path, bool
use_threads,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool) {
+ if (!fs || !arrow_pool || path.empty()) {
+ return Status::Invalid("invalid Arrow IPC file reader arguments");
+ }
+ std::unique_ptr<ArrowIpcFileReader> reader(
+ new ArrowIpcFileReader(fs, path, use_threads, arrow_pool));
+ PAIMON_RETURN_NOT_OK(reader->DoOpen());
+ return reader;
+}
+
+Status ArrowIpcFileReader::DoOpen() {
+ PAIMON_ASSIGN_OR_RAISE(in_stream_, fs_->Open(path_));
+ PAIMON_ASSIGN_OR_RAISE(FileStatus status, fs_->GetFileStatus(path_));
+ arrow_input_stream_adapter_ =
+ std::make_shared<ArrowInputStreamAdapter>(in_stream_, status.GetLen(),
arrow_pool_);
+ auto read_options = arrow::ipc::IpcReadOptions::Defaults();
+ read_options.memory_pool = arrow_pool_.get();
+ read_options.use_threads = use_threads_;
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ arrow_reader_,
+ arrow::ipc::RecordBatchFileReader::Open(arrow_input_stream_adapter_,
read_options));
+ record_batch_count_ = arrow_reader_->num_record_batches();
+ return Status::OK();
+}
+
+int32_t ArrowIpcFileReader::GetRecordBatchCount() const {
+ return record_batch_count_;
+}
+
+Result<std::shared_ptr<arrow::RecordBatch>>
ArrowIpcFileReader::ReadRecordBatch(
+ int32_t batch_index) {
+ if (closed_ || !arrow_reader_) {
+ return Status::Invalid("Arrow IPC file reader is closed");
+ }
+ if (batch_index < 0 || batch_index >= record_batch_count_) {
+ return Status::Invalid("Arrow IPC record batch index out of range: ",
batch_index);
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::RecordBatch>
batch,
+
arrow_reader_->ReadRecordBatch(batch_index));
+ return batch;
+}
+
+void ArrowIpcFileReader::Close() {
+ if (closed_) {
+ return;
+ }
+ arrow_reader_.reset();
+ arrow_input_stream_adapter_.reset();
+ if (in_stream_) {
+ [[maybe_unused]] Status status = in_stream_->Close();
+ in_stream_.reset();
+ }
+ closed_ = true;
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/arrow_ipc_file.h
b/src/paimon/core/io/arrow_ipc_file.h
new file mode 100644
index 00000000..965dec7e
--- /dev/null
+++ b/src/paimon/core/io/arrow_ipc_file.h
@@ -0,0 +1,110 @@
+/*
+ * 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 <cstdint>
+#include <memory>
+#include <string>
+
+#include "arrow/ipc/type_fwd.h"
+#include "arrow/type_fwd.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+class ArrowInputStreamAdapter;
+class ArrowOutputStreamAdapter;
+class FileSystem;
+class InputStream;
+class OutputStream;
+
+/// Low-level Arrow IPC file writer shared by temporary spill implementations.
+class ArrowIpcFileWriter {
+ public:
+ static Result<std::unique_ptr<ArrowIpcFileWriter>> Create(
+ const std::shared_ptr<FileSystem>& fs, const std::string& path,
+ const std::shared_ptr<arrow::Schema>& schema, const std::string&
compression,
+ int32_t compression_level, bool use_threads,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
+
+ ~ArrowIpcFileWriter();
+
+ ArrowIpcFileWriter(const ArrowIpcFileWriter&) = delete;
+ ArrowIpcFileWriter& operator=(const ArrowIpcFileWriter&) = delete;
+
+ Status WriteBatch(const std::shared_ptr<arrow::RecordBatch>& batch);
+ Status Close();
+ Result<int64_t> GetFileSize() const;
+
+ private:
+ ArrowIpcFileWriter(const std::shared_ptr<FileSystem>& fs, std::string path,
+ const std::shared_ptr<arrow::Schema>& schema,
std::string compression,
+ int32_t compression_level, bool use_threads,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
+
+ Status Open();
+
+ std::shared_ptr<FileSystem> fs_;
+ std::string path_;
+ std::shared_ptr<arrow::Schema> schema_;
+ std::string compression_;
+ int32_t compression_level_;
+ bool use_threads_;
+ std::shared_ptr<OutputStream> out_stream_;
+ std::shared_ptr<ArrowOutputStreamAdapter> arrow_output_stream_adapter_;
+ std::shared_ptr<arrow::MemoryPool> arrow_pool_;
+ std::shared_ptr<arrow::ipc::RecordBatchWriter> arrow_writer_;
+ bool closed_ = false;
+};
+
+/// One independently opened Arrow IPC file reader.
+class ArrowIpcFileReader {
+ public:
+ static Result<std::unique_ptr<ArrowIpcFileReader>> Open(
+ const std::shared_ptr<FileSystem>& fs, const std::string& path, bool
use_threads,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
+
+ ~ArrowIpcFileReader();
+
+ ArrowIpcFileReader(const ArrowIpcFileReader&) = delete;
+ ArrowIpcFileReader& operator=(const ArrowIpcFileReader&) = delete;
+
+ int32_t GetRecordBatchCount() const;
+ Result<std::shared_ptr<arrow::RecordBatch>> ReadRecordBatch(int32_t
batch_index);
+ void Close();
+
+ private:
+ ArrowIpcFileReader(const std::shared_ptr<FileSystem>& fs, std::string
path, bool use_threads,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
+
+ Status DoOpen();
+
+ std::shared_ptr<FileSystem> fs_;
+ std::string path_;
+ bool use_threads_;
+ std::shared_ptr<arrow::MemoryPool> arrow_pool_;
+ std::shared_ptr<InputStream> in_stream_;
+ std::shared_ptr<ArrowInputStreamAdapter> arrow_input_stream_adapter_;
+ std::shared_ptr<arrow::ipc::RecordBatchFileReader> arrow_reader_;
+ int32_t record_batch_count_ = 0;
+ bool closed_ = false;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/mergetree/spill_reader.cpp
b/src/paimon/core/mergetree/spill_reader.cpp
index c7916b22..a1a27537 100644
--- a/src/paimon/core/mergetree/spill_reader.cpp
+++ b/src/paimon/core/mergetree/spill_reader.cpp
@@ -18,55 +18,43 @@
#include "paimon/core/mergetree/spill_reader.h"
+#include "arrow/api.h"
#include "paimon/common/data/columnar/columnar_row_ref.h"
#include "paimon/common/metrics/metrics_impl.h"
#include "paimon/common/table/special_fields.h"
#include "paimon/common/types/row_kind.h"
-#include "paimon/common/utils/arrow/arrow_input_stream_adapter.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/core/io/arrow_ipc_file.h"
+#include "paimon/macros.h"
namespace paimon {
-SpillReader::SpillReader(const std::shared_ptr<FileSystem>& fs,
- const std::shared_ptr<arrow::Schema>& key_schema,
- const std::shared_ptr<arrow::Schema>& value_schema,
bool use_threads,
+SpillReader::~SpillReader() = default;
+
+SpillReader::SpillReader(const std::shared_ptr<arrow::Schema>& key_schema,
+ const std::shared_ptr<arrow::Schema>& value_schema,
+ std::unique_ptr<ArrowIpcFileReader>&& ipc_reader,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool,
const std::shared_ptr<MemoryPool>& pool)
- : fs_(fs),
- key_schema_(key_schema),
+ : key_schema_(key_schema),
value_schema_(value_schema),
pool_(pool),
- arrow_pool_(GetArrowPool(pool)),
- use_threads_(use_threads),
- metrics_(std::make_shared<MetricsImpl>()) {}
+ arrow_pool_(arrow_pool),
+ metrics_(std::make_shared<MetricsImpl>()),
+ ipc_reader_(std::move(ipc_reader)),
+ num_record_batches_(ipc_reader_->GetRecordBatchCount()) {}
Result<std::unique_ptr<SpillReader>> SpillReader::Create(
const std::shared_ptr<FileSystem>& fs, const
std::shared_ptr<arrow::Schema>& key_schema,
const std::shared_ptr<arrow::Schema>& value_schema, bool use_threads,
const FileIOChannel::ID& channel_id, const std::shared_ptr<MemoryPool>&
pool) {
- std::unique_ptr<SpillReader> reader(
- new SpillReader(fs, key_schema, value_schema, use_threads, pool));
- PAIMON_RETURN_NOT_OK(reader->Open(channel_id));
- return reader;
-}
-
-Status SpillReader::Open(const FileIOChannel::ID& channel_id) {
- const std::string& file_path = channel_id.GetPath();
- PAIMON_ASSIGN_OR_RAISE(in_stream_, fs_->Open(file_path));
- PAIMON_ASSIGN_OR_RAISE(FileStatus file_status,
fs_->GetFileStatus(file_path));
- int64_t file_len = file_status.GetLen();
- arrow_input_stream_adapter_ =
- std::make_shared<ArrowInputStreamAdapter>(in_stream_, file_len,
arrow_pool_);
- auto ipc_read_options = arrow::ipc::IpcReadOptions::Defaults();
- ipc_read_options.memory_pool = arrow_pool_.get();
- ipc_read_options.use_threads = use_threads_;
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
- arrow_reader_,
- arrow::ipc::RecordBatchFileReader::Open(arrow_input_stream_adapter_,
ipc_read_options));
- num_record_batches_ = arrow_reader_->num_record_batches();
- current_batch_index_ = 0;
- return Status::OK();
+ std::shared_ptr<arrow::MemoryPool> arrow_pool = GetArrowPool(pool);
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<ArrowIpcFileReader> ipc_reader,
+ ArrowIpcFileReader::Open(fs, channel_id.GetPath(), use_threads,
arrow_pool));
+ return std::unique_ptr<SpillReader>(
+ new SpillReader(key_schema, value_schema, std::move(ipc_reader),
arrow_pool, pool));
}
SpillReader::Iterator::Iterator(SpillReader* reader) : reader_(reader) {}
@@ -91,8 +79,8 @@ Result<std::unique_ptr<KeyValueRecordReader::Iterator>>
SpillReader::NextBatch()
if (current_batch_index_ >= num_record_batches_) {
return std::unique_ptr<KeyValueRecordReader::Iterator>();
}
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::RecordBatch>
record_batch,
-
arrow_reader_->ReadRecordBatch(current_batch_index_));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::RecordBatch> record_batch,
+ ipc_reader_->ReadRecordBatch(current_batch_index_));
current_batch_index_++;
batch_length_ = record_batch->num_rows();
@@ -150,11 +138,9 @@ std::shared_ptr<Metrics> SpillReader::GetReaderMetrics()
const {
void SpillReader::Close() {
Reset();
- arrow_reader_.reset();
- arrow_input_stream_adapter_.reset();
- if (in_stream_) {
- [[maybe_unused]] auto status = in_stream_->Close();
- in_stream_.reset();
+ if (ipc_reader_) {
+ ipc_reader_->Close();
+ ipc_reader_.reset();
}
}
diff --git a/src/paimon/core/mergetree/spill_reader.h
b/src/paimon/core/mergetree/spill_reader.h
index af05d271..4bc2853d 100644
--- a/src/paimon/core/mergetree/spill_reader.h
+++ b/src/paimon/core/mergetree/spill_reader.h
@@ -21,7 +21,6 @@
#include <memory>
#include "arrow/array/array_primitive.h"
-#include "arrow/ipc/api.h"
#include "paimon/common/data/columnar/columnar_batch_context.h"
#include "paimon/core/disk/file_io_channel.h"
#include "paimon/core/io/key_value_record_reader.h"
@@ -31,11 +30,12 @@
namespace arrow {
class MemoryPool;
+class Schema;
} // namespace arrow
namespace paimon {
-class ArrowInputStreamAdapter;
+class ArrowIpcFileReader;
class Metrics;
class SpillReader : public KeyValueRecordReader {
@@ -48,6 +48,8 @@ class SpillReader : public KeyValueRecordReader {
SpillReader(const SpillReader&) = delete;
SpillReader& operator=(const SpillReader&) = delete;
+ ~SpillReader() override;
+
class Iterator : public KeyValueRecordReader::Iterator {
public:
explicit Iterator(SpillReader* reader);
@@ -66,25 +68,23 @@ class SpillReader : public KeyValueRecordReader {
void Close() override;
private:
- SpillReader(const std::shared_ptr<FileSystem>& fs,
- const std::shared_ptr<arrow::Schema>& key_schema,
- const std::shared_ptr<arrow::Schema>& value_schema, bool
use_threads,
+ SpillReader(const std::shared_ptr<arrow::Schema>& key_schema,
+ const std::shared_ptr<arrow::Schema>& value_schema,
+ std::unique_ptr<ArrowIpcFileReader>&& ipc_reader,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool,
const std::shared_ptr<MemoryPool>& pool);
-
- Status Open(const FileIOChannel::ID& channel_id);
void Reset();
- std::shared_ptr<FileSystem> fs_;
std::shared_ptr<arrow::Schema> key_schema_;
std::shared_ptr<arrow::Schema> value_schema_;
std::shared_ptr<MemoryPool> pool_;
+ // Record batches returned by Arrow keep only a raw pointer to their
allocation pool. Keep the
+ // adapter alive after closing the IPC reader while rows from the last
batch may still be held
+ // by the merge reader.
std::shared_ptr<arrow::MemoryPool> arrow_pool_;
- bool use_threads_;
std::shared_ptr<Metrics> metrics_;
- std::shared_ptr<InputStream> in_stream_;
- std::shared_ptr<ArrowInputStreamAdapter> arrow_input_stream_adapter_;
- std::shared_ptr<arrow::ipc::RecordBatchFileReader> arrow_reader_;
+ std::unique_ptr<ArrowIpcFileReader> ipc_reader_;
int32_t current_batch_index_ = 0;
int32_t num_record_batches_ = 0;
diff --git a/src/paimon/core/mergetree/spill_writer.cpp
b/src/paimon/core/mergetree/spill_writer.cpp
index 1e0e1a33..f5ca4868 100644
--- a/src/paimon/core/mergetree/spill_writer.cpp
+++ b/src/paimon/core/mergetree/spill_writer.cpp
@@ -18,29 +18,19 @@
#include "paimon/core/mergetree/spill_writer.h"
-#include "paimon/common/utils/arrow/arrow_output_stream_adapter.h"
-#include "paimon/common/utils/arrow/arrow_utils.h"
#include "paimon/common/utils/arrow/mem_utils.h"
-#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/io/arrow_ipc_file.h"
#include "paimon/core/mergetree/spill_channel_manager.h"
+#include "paimon/macros.h"
namespace paimon {
-SpillWriter::SpillWriter(const std::shared_ptr<FileSystem>& fs,
- const std::shared_ptr<arrow::Schema>& schema,
- const std::shared_ptr<FileIOChannel::Enumerator>&
channel_enumerator,
- const std::shared_ptr<SpillChannelManager>&
spill_channel_manager,
- const std::string& compression, int32_t
compression_level,
- bool use_threads, const std::shared_ptr<MemoryPool>&
pool)
- : fs_(fs),
- schema_(schema),
- channel_enumerator_(channel_enumerator),
- spill_channel_manager_(spill_channel_manager),
- compression_(compression),
- compression_level_(compression_level),
- use_threads_(use_threads),
- arrow_pool_(GetArrowPool(pool)) {}
+SpillWriter::~SpillWriter() = default;
+
+SpillWriter::SpillWriter(std::unique_ptr<ArrowIpcFileWriter>&& ipc_writer,
+ const FileIOChannel::ID& channel_id)
+ : ipc_writer_(std::move(ipc_writer)), channel_id_(channel_id) {}
Result<std::unique_ptr<SpillWriter>> SpillWriter::Create(
const std::shared_ptr<FileSystem>& fs, const
std::shared_ptr<arrow::Schema>& schema,
@@ -48,75 +38,35 @@ Result<std::unique_ptr<SpillWriter>> SpillWriter::Create(
const std::shared_ptr<SpillChannelManager>& spill_channel_manager,
const std::string& compression, int32_t compression_level, bool
use_threads,
const std::shared_ptr<MemoryPool>& pool) {
- std::unique_ptr<SpillWriter> writer(new SpillWriter(fs, schema,
channel_enumerator,
- spill_channel_manager,
compression,
- compression_level,
use_threads, pool));
- PAIMON_RETURN_NOT_OK(writer->Open());
- return writer;
-}
-
-Status SpillWriter::Open() {
- channel_id_ = channel_enumerator_->Next();
- auto ipc_write_options = arrow::ipc::IpcWriteOptions::Defaults();
- ipc_write_options.memory_pool = arrow_pool_.get();
- ipc_write_options.use_threads = use_threads_;
+ FileIOChannel::ID channel_id = channel_enumerator->Next();
auto cleanup_guard = ScopeGuard([&]() {
- arrow_writer_.reset();
- arrow_output_stream_adapter_.reset();
- if (out_stream_) {
- [[maybe_unused]] auto status = out_stream_->Close();
- out_stream_.reset();
- }
- if (!channel_id_.GetPath().empty()) {
- [[maybe_unused]] auto status = fs_->Delete(channel_id_.GetPath());
+ if (!channel_id.GetPath().empty()) {
+ [[maybe_unused]] auto status = fs->Delete(channel_id.GetPath());
}
});
- PAIMON_ASSIGN_OR_RAISE(arrow::Compression::type arrow_compression,
- ArrowUtils::GetCompressionType(compression_));
- if (!arrow::util::Codec::SupportsCompressionLevel(arrow_compression)) {
- compression_level_ = arrow::util::Codec::UseDefaultCompressionLevel();
- }
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
- ipc_write_options.codec, arrow::util::Codec::Create(arrow_compression,
compression_level_));
- PAIMON_ASSIGN_OR_RAISE(out_stream_, fs_->Create(channel_id_.GetPath(),
/*overwrite=*/false));
- arrow_output_stream_adapter_ =
std::make_shared<ArrowOutputStreamAdapter>(out_stream_);
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
- arrow_writer_,
- arrow::ipc::MakeFileWriter(arrow_output_stream_adapter_, schema_,
ipc_write_options));
- spill_channel_manager_->AddChannel(channel_id_);
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<ArrowIpcFileWriter> ipc_writer,
+ ArrowIpcFileWriter::Create(fs, channel_id.GetPath(), schema,
compression, compression_level,
+ use_threads, GetArrowPool(pool)));
+ std::unique_ptr<SpillWriter> writer(new SpillWriter(std::move(ipc_writer),
channel_id));
+ spill_channel_manager->AddChannel(channel_id);
cleanup_guard.Release();
- return Status::OK();
+ return writer;
}
Status SpillWriter::WriteBatch(const std::shared_ptr<arrow::RecordBatch>&
batch) {
- PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow_writer_->WriteRecordBatch(*batch));
- return Status::OK();
+ return ipc_writer_->WriteBatch(batch);
}
Status SpillWriter::Close() {
- if (closed_) {
- return Status::OK();
- }
- if (arrow_writer_) {
- PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow_writer_->Close());
- }
- if (out_stream_) {
- PAIMON_RETURN_NOT_OK(out_stream_->Close());
- }
- closed_ = true;
- return Status::OK();
+ return ipc_writer_->Close();
}
Result<int64_t> SpillWriter::GetFileSize() const {
if (channel_id_.GetPath().empty()) {
return Status::Invalid("spill writer has no channel id");
}
- if (!closed_ && arrow_output_stream_adapter_ != nullptr) {
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(int64_t file_size,
arrow_output_stream_adapter_->Tell());
- return file_size;
- }
- PAIMON_ASSIGN_OR_RAISE(FileStatus file_status,
fs_->GetFileStatus(channel_id_.GetPath()));
- return file_status.GetLen();
+ return ipc_writer_->GetFileSize();
}
const FileIOChannel::ID& SpillWriter::GetChannelId() const {
diff --git a/src/paimon/core/mergetree/spill_writer.h
b/src/paimon/core/mergetree/spill_writer.h
index 3aaf1b04..0f02cf58 100644
--- a/src/paimon/core/mergetree/spill_writer.h
+++ b/src/paimon/core/mergetree/spill_writer.h
@@ -21,20 +21,15 @@
#include <memory>
#include <string>
-#include "arrow/ipc/api.h"
+#include "arrow/type_fwd.h"
#include "paimon/core/disk/file_io_channel.h"
#include "paimon/fs/file_system.h"
#include "paimon/result.h"
#include "paimon/status.h"
-namespace arrow {
-class RecordBatch;
-class Schema;
-} // namespace arrow
-
namespace paimon {
-class ArrowOutputStreamAdapter;
+class ArrowIpcFileWriter;
class MemoryPool;
class SpillChannelManager;
@@ -47,6 +42,8 @@ class SpillWriter {
const std::string& compression, int32_t compression_level, bool
use_threads,
const std::shared_ptr<MemoryPool>& pool);
+ ~SpillWriter();
+
SpillWriter(const SpillWriter&) = delete;
SpillWriter& operator=(const SpillWriter&) = delete;
@@ -56,27 +53,11 @@ class SpillWriter {
const FileIOChannel::ID& GetChannelId() const;
private:
- SpillWriter(const std::shared_ptr<FileSystem>& fs, const
std::shared_ptr<arrow::Schema>& schema,
- const std::shared_ptr<FileIOChannel::Enumerator>&
channel_enumerator,
- const std::shared_ptr<SpillChannelManager>&
spill_channel_manager,
- const std::string& compression, int32_t compression_level,
bool use_threads,
- const std::shared_ptr<MemoryPool>& pool);
-
- Status Open();
+ SpillWriter(std::unique_ptr<ArrowIpcFileWriter>&& ipc_writer,
+ const FileIOChannel::ID& channel_id);
- std::shared_ptr<FileSystem> fs_;
- std::shared_ptr<arrow::Schema> schema_;
- std::shared_ptr<FileIOChannel::Enumerator> channel_enumerator_;
- std::shared_ptr<SpillChannelManager> spill_channel_manager_;
- std::string compression_;
- int32_t compression_level_;
- bool use_threads_;
- std::shared_ptr<OutputStream> out_stream_;
- std::shared_ptr<ArrowOutputStreamAdapter> arrow_output_stream_adapter_;
- std::shared_ptr<arrow::MemoryPool> arrow_pool_;
- std::shared_ptr<arrow::ipc::RecordBatchWriter> arrow_writer_;
+ std::unique_ptr<ArrowIpcFileWriter> ipc_writer_;
FileIOChannel::ID channel_id_;
- bool closed_ = false;
};
} // namespace paimon
diff --git a/src/paimon/core/operation/append_only_file_store_write.cpp
b/src/paimon/core/operation/append_only_file_store_write.cpp
index 874fdb3e..f0001f64 100644
--- a/src/paimon/core/operation/append_only_file_store_write.cpp
+++ b/src/paimon/core/operation/append_only_file_store_write.cpp
@@ -37,6 +37,7 @@
#include "paimon/core/append/bucketed_append_compact_manager.h"
#include "paimon/core/compact/noop_compact_manager.h"
#include "paimon/core/core_options.h"
+#include "paimon/core/disk/io_manager.h"
#include "paimon/core/io/append_data_file_writer_factory.h"
#include "paimon/core/io/data_file_meta.h"
#include "paimon/core/io/data_file_path_factory.h"
@@ -300,7 +301,8 @@ Result<std::shared_ptr<BatchWriter>>
AppendOnlyFileStoreWrite::CreateWriter(
std::map<std::string, std::string> partition_map(partition_values.begin(),
partition_values.end());
return RealtimeAppendOnlyWriter::Create(partition_map, bucket,
realtime_context_, writer,
- realtime_schema_layout_, options_,
pool_);
+ realtime_schema_layout_, options_,
+ io_manager_ ?
io_manager_->GetTempDir() : "", pool_);
}
Result<AppendOnlyFileStoreWrite::WriterFactory>
AppendOnlyFileStoreWrite::GetDataFileWriterFactory(
diff --git a/src/paimon/core/operation/key_value_file_store_write.cpp
b/src/paimon/core/operation/key_value_file_store_write.cpp
index 5bb34a66..a8cf0ea7 100644
--- a/src/paimon/core/operation/key_value_file_store_write.cpp
+++ b/src/paimon/core/operation/key_value_file_store_write.cpp
@@ -26,6 +26,7 @@
#include "paimon/common/table/special_fields.h"
#include "paimon/core/compact/noop_compact_manager.h"
#include "paimon/core/core_options.h"
+#include "paimon/core/disk/io_manager.h"
#include "paimon/core/io/data_file_meta.h"
#include "paimon/core/manifest/manifest_file.h"
#include "paimon/core/manifest/manifest_list.h"
@@ -139,13 +140,17 @@ Result<std::shared_ptr<BatchWriter>>
KeyValueFileStoreWrite::CreateWriter(
auto c_write_schema = std::make_unique<ArrowSchema>();
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(
*realtime_schema_layout_->StoreWriteSchema(),
c_write_schema.get()));
+ RealtimeStoreCreateRequest request{std::move(c_write_schema),
options_.ToMap(), pool_,
+ RealtimeStoreMode::PRIMARY_KEY,
+
options_.GetRealtimeStoreStatisticsMode()};
+ if (options_.RealtimeSpillEnabled()) {
+ request.temp_directory = io_manager_ ? io_manager_->GetTempDir() :
"";
+ request.file_system = options_.GetFileSystem();
+ }
PAIMON_ASSIGN_OR_RAISE(
RealtimeStoreState store_state,
realtime_context_impl->GetOrCreateRealtimeStore(
- RealtimeStoreCreateRequest{std::move(c_write_schema),
options_.ToMap(), pool_,
- RealtimeStoreMode::PRIMARY_KEY,
-
options_.GetRealtimeStoreStatisticsMode()},
- RealtimePartitionBucket(partition_map, bucket)));
+ std::move(request), RealtimePartitionBucket(partition_map,
bucket)));
realtime_store_state = std::move(store_state);
compact_manager = std::make_shared<NoopCompactManager>();
} else {
diff --git a/src/paimon/core/realtime/arrow_realtime_store.cpp
b/src/paimon/core/realtime/arrow_realtime_store.cpp
index 4950ecfb..45be410a 100644
--- a/src/paimon/core/realtime/arrow_realtime_store.cpp
+++ b/src/paimon/core/realtime/arrow_realtime_store.cpp
@@ -20,6 +20,7 @@
#include "paimon/core/realtime/arrow_realtime_store.h"
#include <algorithm>
+#include <numeric>
#include <optional>
#include <utility>
@@ -30,14 +31,17 @@
#include "paimon/common/data/columnar/columnar_row.h"
#include "paimon/common/metrics/metrics_impl.h"
#include "paimon/common/predicate/predicate_filter.h"
-#include "paimon/common/reader/reader_utils.h"
#include "paimon/common/utils/arrow/arrow_utils.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/projected_array.h"
#include "paimon/common/utils/projected_row.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/disk/io_manager.h"
+#include "paimon/core/io/arrow_ipc_file.h"
#include "paimon/core/utils/nested_projection_utils.h"
+#include "paimon/fs/file_system.h"
#include "paimon/macros.h"
namespace paimon {
@@ -75,18 +79,95 @@ Result<std::shared_ptr<arrow::StructArray>> ProjectBatch(
return checked_pointer_cast<arrow::StructArray>(projected);
}
-} // namespace
+Result<std::shared_ptr<arrow::StructArray>> ToStructArray(
+ const std::shared_ptr<arrow::RecordBatch>& batch) {
+ if (!batch) {
+ return Status::Invalid("Arrow IPC reader returned a null record
batch");
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::StructArray> result,
+ arrow::StructArray::Make(batch->columns(), batch->schema()->fields()));
+ return result;
+}
-class ArrowRealtimeStore::Segment : public RealtimeSegmentHandle {
+Result<BatchReader::ReadBatch> ExportStructBatch(
+ const std::shared_ptr<arrow::StructArray>& data,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> normalized,
+ ArrowUtils::NormalizeArrayOffsets(data,
arrow_pool.get()));
+ auto c_array = std::make_unique<ArrowArray>();
+ auto c_schema = std::make_unique<ArrowSchema>();
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*normalized,
c_array.get(), c_schema.get()));
+ PAIMON_RETURN_NOT_OK(AddArrowArrayLifetime(c_array.get(), c_schema.get(),
arrow_pool));
+ return BatchReader::ReadBatch(std::move(c_array), std::move(c_schema));
+}
+
+class SlicedQueryBatchReader final : public BatchReader {
public:
- Segment(const OffsetRange& offset_range, std::vector<StoredBatch>&&
batches)
- : offset_range_(offset_range), batches_(std::move(batches)) {
- for (const StoredBatch& batch : batches_) {
- row_count_ += batch.data->length();
- memory_usage_ += batch.memory_usage;
+ SlicedQueryBatchReader(std::unique_ptr<BatchReader>&& reader, int32_t
read_batch_size,
+ const std::shared_ptr<arrow::MemoryPool>&
arrow_pool)
+ : reader_(std::move(reader)), read_batch_size_(read_batch_size),
arrow_pool_(arrow_pool) {}
+
+ Result<ReadBatch> NextBatch() override {
+ while (!current_batch_ || next_row_ >= current_batch_->length()) {
+ current_batch_.reset();
+ next_row_ = 0;
+ PAIMON_ASSIGN_OR_RAISE(ReadBatch batch, reader_->NextBatch());
+ if (IsEofBatch(batch)) {
+ return batch;
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::Array> array,
+ arrow::ImportArray(batch.first.get(), batch.second.get()));
+ if (!array || array->type_id() != arrow::Type::STRUCT) {
+ return Status::Invalid("real-time query batch must be a
StructArray");
+ }
+ current_batch_ = checked_pointer_cast<arrow::StructArray>(array);
}
+
+ const int64_t length =
+ std::min<int64_t>(read_batch_size_, current_batch_->length() -
next_row_);
+ std::shared_ptr<arrow::StructArray> slice =
+
checked_pointer_cast<arrow::StructArray>(current_batch_->Slice(next_row_,
length));
+ next_row_ += length;
+ return ExportStructBatch(slice, arrow_pool_);
}
+ std::shared_ptr<Metrics> GetReaderMetrics() const override {
+ return reader_->GetReaderMetrics();
+ }
+
+ void Close() override {
+ current_batch_.reset();
+ arrow_pool_.reset();
+ reader_->Close();
+ }
+
+ private:
+ std::unique_ptr<BatchReader> reader_;
+ int32_t read_batch_size_;
+ std::shared_ptr<arrow::MemoryPool> arrow_pool_;
+ std::shared_ptr<arrow::StructArray> current_batch_;
+ int64_t next_row_ = 0;
+};
+
+std::vector<std::unique_ptr<BatchReader>> SliceQueryReaders(
+ std::vector<std::unique_ptr<BatchReader>>&& readers, int32_t
read_batch_size,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool) {
+ for (std::unique_ptr<BatchReader>& reader : readers) {
+ reader = std::make_unique<SlicedQueryBatchReader>(std::move(reader),
read_batch_size,
+ arrow_pool);
+ }
+ return std::move(readers);
+}
+
+} // namespace
+
+class ArrowRealtimeStore::Segment : public RealtimeSegmentHandle {
+ public:
+ Segment(const OffsetRange& offset_range, int64_t row_count)
+ : offset_range_(offset_range), row_count_(row_count) {}
+
OffsetRange GetOffsetRange() const override {
return offset_range_;
}
@@ -95,21 +176,132 @@ class ArrowRealtimeStore::Segment : public
RealtimeSegmentHandle {
return row_count_;
}
+ virtual uint64_t GetMemoryUsage() const = 0;
+
+ private:
+ OffsetRange offset_range_;
+ int64_t row_count_;
+};
+
+class ArrowRealtimeStore::MemorySegment final : public
ArrowRealtimeStore::Segment {
+ public:
+ MemorySegment(const OffsetRange& offset_range, std::vector<StoredBatch>&&
batches)
+ : Segment(offset_range, CountRows(batches)),
batches_(std::move(batches)) {
+ for (const StoredBatch& batch : batches_) {
+ memory_usage_ += batch.memory_usage;
+ }
+ }
+
const std::vector<StoredBatch>& GetBatches() const {
return batches_;
}
- uint64_t GetMemoryUsage() const {
+ uint64_t GetMemoryUsage() const override {
return memory_usage_;
}
private:
- OffsetRange offset_range_;
+ static int64_t CountRows(const std::vector<StoredBatch>& batches) {
+ int64_t result = 0;
+ for (const StoredBatch& batch : batches) {
+ result += batch.data->length();
+ }
+ return result;
+ }
+
std::vector<StoredBatch> batches_;
- int64_t row_count_ = 0;
uint64_t memory_usage_ = 0;
};
+class ArrowRealtimeStore::SpillFile {
+ public:
+ SpillFile(const std::shared_ptr<IOManager>& io_manager,
+ const std::shared_ptr<FileSystem>& file_system,
FileIOChannel::ID channel_id)
+ : io_manager_(io_manager), file_system_(file_system),
channel_id_(std::move(channel_id)) {}
+
+ ~SpillFile() {
+ if (file_system_) {
+ [[maybe_unused]] Status status =
file_system_->Delete(channel_id_.GetPath());
+ }
+ }
+
+ const std::shared_ptr<FileSystem>& GetFileSystem() const {
+ return file_system_;
+ }
+
+ const std::string& GetPath() const {
+ return channel_id_.GetPath();
+ }
+
+ private:
+ // Keep the manager and its root directory alive as long as this file is
referenced.
+ std::shared_ptr<IOManager> io_manager_;
+ std::shared_ptr<FileSystem> file_system_;
+ FileIOChannel::ID channel_id_;
+};
+
+class ArrowRealtimeStore::SpilledSegment final : public
ArrowRealtimeStore::Segment {
+ public:
+ SpilledSegment(const OffsetRange& offset_range, int64_t row_count,
+ std::vector<SpilledBatch>&& batches,
std::shared_ptr<SpillFile> file)
+ : Segment(offset_range, row_count), batches_(std::move(batches)),
file_(std::move(file)) {
+ for (const SpilledBatch& batch : batches_) {
+ memory_usage_ += batch.memory_usage;
+ }
+ }
+
+ const std::vector<SpilledBatch>& GetBatches() const {
+ return batches_;
+ }
+
+ const std::shared_ptr<SpillFile>& GetFile() const {
+ return file_;
+ }
+
+ uint64_t GetMemoryUsage() const override {
+ return memory_usage_;
+ }
+
+ private:
+ std::vector<SpilledBatch> batches_;
+ std::shared_ptr<SpillFile> file_;
+ uint64_t memory_usage_ = 0;
+};
+
+class ArrowRealtimeStore::SharedSpillFileReader {
+ public:
+ static Result<std::shared_ptr<SharedSpillFileReader>> Create(
+ const std::shared_ptr<SpillFile>& spill_file, int32_t
expected_batch_count,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<ArrowIpcFileReader> file_reader,
+ ArrowIpcFileReader::Open(spill_file->GetFileSystem(),
spill_file->GetPath(),
+ /*use_threads=*/false, arrow_pool));
+ if (file_reader->GetRecordBatchCount() != expected_batch_count) {
+ return Status::Invalid(
+ "real-time spill file batch count does not match segment
metadata");
+ }
+ return std::shared_ptr<SharedSpillFileReader>(
+ new SharedSpillFileReader(spill_file, std::move(file_reader)));
+ }
+
+ Result<std::shared_ptr<arrow::RecordBatch>> ReadRecordBatch(int32_t
batch_index) {
+ std::lock_guard<std::mutex> lock(mutex_);
+ return file_reader_->ReadRecordBatch(batch_index);
+ }
+
+ private:
+ SharedSpillFileReader(const std::shared_ptr<SpillFile>& spill_file,
+ std::unique_ptr<ArrowIpcFileReader>&& file_reader)
+ : spill_file_(spill_file), file_reader_(std::move(file_reader)) {}
+
+ // Keep the spill file alive until the shared IPC reader closes. Members
are destroyed in
+ // reverse declaration order, so the reader closes before the file can be
deleted.
+ std::shared_ptr<SpillFile> spill_file_;
+ std::unique_ptr<ArrowIpcFileReader> file_reader_;
+ std::mutex mutex_;
+};
+
class ArrowRealtimeStore::ReadView : public RealtimeReadView {
public:
explicit ReadView(std::vector<std::shared_ptr<Segment>>&& segments)
@@ -135,7 +327,7 @@ class ArrowRealtimeStore::ReadView : public
RealtimeReadView {
class ArrowRealtimeStore::AppendCommitBatchReader : public BatchReader {
public:
- explicit AppendCommitBatchReader(const std::shared_ptr<Segment>& segment)
+ explicit AppendCommitBatchReader(const std::shared_ptr<MemorySegment>&
segment)
: segment_(segment), metrics_(std::make_shared<MetricsImpl>()) {}
Result<ReadBatch> NextBatch() override {
@@ -159,7 +351,7 @@ class ArrowRealtimeStore::AppendCommitBatchReader : public
BatchReader {
}
private:
- std::shared_ptr<Segment> segment_;
+ std::shared_ptr<MemorySegment> segment_;
std::shared_ptr<Metrics> metrics_;
size_t next_batch_ = 0;
};
@@ -174,16 +366,10 @@ class ArrowRealtimeStore::StoredBatchReader : public
BatchReader {
if (!data_) {
return MakeEofBatch();
}
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> normalized_array,
- ArrowUtils::NormalizeArrayOffsets(data_,
arrow_pool_.get()));
- auto c_array = std::make_unique<ArrowArray>();
- auto c_schema = std::make_unique<ArrowSchema>();
- PAIMON_RETURN_NOT_OK_FROM_ARROW(
- arrow::ExportArray(*normalized_array, c_array.get(),
c_schema.get()));
- PAIMON_RETURN_NOT_OK(AddArrowArrayLifetime(c_array.get(),
c_schema.get(), arrow_pool_));
+ PAIMON_ASSIGN_OR_RAISE(ReadBatch batch, ExportStructBatch(data_,
arrow_pool_));
data_.reset();
arrow_pool_.reset();
- return ReadBatch(std::move(c_array), std::move(c_schema));
+ return batch;
}
std::shared_ptr<Metrics> GetReaderMetrics() const override {
@@ -201,6 +387,53 @@ class ArrowRealtimeStore::StoredBatchReader : public
BatchReader {
std::shared_ptr<Metrics> metrics_;
};
+class ArrowRealtimeStore::SpillBatchReader : public BatchReader {
+ public:
+ SpillBatchReader(const std::shared_ptr<SharedSpillFileReader>& file_reader,
+ std::vector<int32_t>&& batch_indexes,
+ const std::shared_ptr<arrow::Schema>& read_schema,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool)
+ : file_reader_(file_reader),
+ batch_indexes_(std::move(batch_indexes)),
+ read_schema_(read_schema),
+ arrow_pool_(arrow_pool),
+ metrics_(std::make_shared<MetricsImpl>()) {}
+
+ Result<ReadBatch> NextBatch() override {
+ if (!file_reader_ || next_batch_ >= batch_indexes_.size()) {
+ return MakeEofBatch();
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::RecordBatch>
record_batch,
+
file_reader_->ReadRecordBatch(batch_indexes_[next_batch_]));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::StructArray> data,
+ ToStructArray(record_batch));
+ if (read_schema_) {
+ PAIMON_ASSIGN_OR_RAISE(data, ProjectBatch(data, read_schema_,
arrow_pool_.get()));
+ }
+ PAIMON_ASSIGN_OR_RAISE(ReadBatch batch, ExportStructBatch(data,
arrow_pool_));
+ ++next_batch_;
+ return batch;
+ }
+
+ std::shared_ptr<Metrics> GetReaderMetrics() const override {
+ return metrics_;
+ }
+
+ void Close() override {
+ file_reader_.reset();
+ read_schema_.reset();
+ arrow_pool_.reset();
+ }
+
+ private:
+ std::shared_ptr<SharedSpillFileReader> file_reader_;
+ std::vector<int32_t> batch_indexes_;
+ std::shared_ptr<arrow::Schema> read_schema_;
+ std::shared_ptr<arrow::MemoryPool> arrow_pool_;
+ std::shared_ptr<Metrics> metrics_;
+ size_t next_batch_ = 0;
+};
+
class ArrowRealtimeStore::AppendQueryBatchReader : public BatchReader {
public:
AppendQueryBatchReader(const std::shared_ptr<ReadView>& view,
@@ -218,40 +451,70 @@ class ArrowRealtimeStore::AppendQueryBatchReader : public
BatchReader {
metrics_(std::make_shared<MetricsImpl>()) {}
Result<ReadBatch> NextBatch() override {
- return Status::Invalid(
- "paimon inner reader ArrowRealtimeStore::AppendQueryBatchReader
should use "
- "NextBatchWithBitmap");
- }
-
- Result<ReadBatchWithBitmap> NextBatchWithBitmap() override {
- // TODO(xinyu.lxy): Memory query reads return complete stored write
batches and
- // intentionally ignore the configured read batch size.
while (view_ && next_segment_ < view_->GetSegments().size()) {
- const std::vector<StoredBatch>& batches =
- view_->GetSegments()[next_segment_]->GetBatches();
- if (next_batch_ >= batches.size()) {
+ const std::shared_ptr<Segment>& segment =
view_->GetSegments()[next_segment_];
+ std::shared_ptr<MemorySegment> memory_segment =
+ std::dynamic_pointer_cast<MemorySegment>(segment);
+ std::shared_ptr<SpilledSegment> spilled_segment =
+ std::dynamic_pointer_cast<SpilledSegment>(segment);
+ if (!memory_segment && !spilled_segment) {
+ return Status::Invalid("unknown Arrow real-time segment type");
+ }
+ size_t batch_count = memory_segment ?
memory_segment->GetBatches().size()
+ :
spilled_segment->GetBatches().size();
+ if (next_batch_ >= batch_count) {
++next_segment_;
next_batch_ = 0;
+ spill_reader_.reset();
continue;
}
- const StoredBatch& stored = batches[next_batch_++];
- PAIMON_ASSIGN_OR_RAISE(bool may_match,
ArrowRealtimeStore::MayMatchStatistics(
- stored, read_schema_,
predicate_filter_,
- statistics_mapping_,
memory_pool_));
+ int64_t row_count;
+ const std::optional<BatchStatistics>* statistics;
+ std::shared_ptr<arrow::StructArray> source;
+ if (memory_segment) {
+ const StoredBatch& stored =
memory_segment->GetBatches()[next_batch_];
+ row_count = stored.data->length();
+ statistics = &stored.statistics;
+ source = stored.data;
+ } else if (spilled_segment) {
+ const SpilledBatch& stored =
spilled_segment->GetBatches()[next_batch_];
+ row_count = stored.row_count;
+ statistics = &stored.statistics;
+ } else {
+ return Status::Invalid("unknown Arrow real-time segment type");
+ }
+ PAIMON_ASSIGN_OR_RAISE(bool may_match,
+ ArrowRealtimeStore::MayMatchStatistics(
+ row_count, *statistics, read_schema_,
predicate_filter_,
+ statistics_mapping_, memory_pool_));
if (!may_match) {
+ ++next_batch_;
continue;
}
+ if (spilled_segment) {
+ if (!spill_reader_) {
+ PAIMON_ASSIGN_OR_RAISE(
+ spill_reader_,
+
ArrowIpcFileReader::Open(spilled_segment->GetFile()->GetFileSystem(),
+
spilled_segment->GetFile()->GetPath(),
+ /*use_threads=*/false,
arrow_pool_));
+ if (spill_reader_->GetRecordBatchCount() !=
+
static_cast<int32_t>(spilled_segment->GetBatches().size())) {
+ return Status::Invalid(
+ "real-time spill file batch count does not match
segment metadata");
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::RecordBatch>
record_batch,
+
spill_reader_->ReadRecordBatch(next_batch_));
+ PAIMON_ASSIGN_OR_RAISE(source, ToStructArray(record_batch));
+ }
+ ++next_batch_;
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::StructArray> output,
- ProjectBatch(stored.data, read_schema_,
arrow_pool_.get()));
- auto c_array = std::make_unique<ArrowArray>();
- auto c_schema = std::make_unique<ArrowSchema>();
- PAIMON_RETURN_NOT_OK_FROM_ARROW(
- arrow::ExportArray(*output, c_array.get(), c_schema.get()));
- PAIMON_RETURN_NOT_OK(AddArrowArrayLifetime(c_array.get(),
c_schema.get(), arrow_pool_));
- return ReaderUtils::AddAllValidBitmap(
- ReadBatch(std::move(c_array), std::move(c_schema)));
+ ProjectBatch(source, read_schema_,
arrow_pool_.get()));
+ PAIMON_ASSIGN_OR_RAISE(ReadBatch batch, ExportStructBatch(output,
arrow_pool_));
+ return batch;
}
- return MakeEofBatchWithBitmap();
+ return MakeEofBatch();
}
std::shared_ptr<Metrics> GetReaderMetrics() const override {
@@ -260,6 +523,7 @@ class ArrowRealtimeStore::AppendQueryBatchReader : public
BatchReader {
void Close() override {
view_ = nullptr;
+ spill_reader_.reset();
}
private:
@@ -270,19 +534,31 @@ class ArrowRealtimeStore::AppendQueryBatchReader : public
BatchReader {
std::shared_ptr<PredicateFilter> predicate_filter_;
std::vector<int32_t> statistics_mapping_;
std::shared_ptr<Metrics> metrics_;
+ std::unique_ptr<ArrowIpcFileReader> spill_reader_;
size_t next_segment_ = 0;
size_t next_batch_ = 0;
};
ArrowRealtimeStore::ArrowRealtimeStore(const std::shared_ptr<arrow::Schema>&
write_schema,
RealtimeStoreMode mode, StatisticsMode
statistics_mode,
+ const std::string& temp_directory,
+ const std::shared_ptr<FileSystem>&
spill_file_system,
+ const std::string& spill_compression,
+ int32_t spill_compression_level,
const std::shared_ptr<MemoryPool>&
memory_pool,
const
std::shared_ptr<arrow::MemoryPool>& arrow_pool)
: write_schema_(write_schema),
memory_pool_(memory_pool),
arrow_pool_(arrow_pool),
mode_(mode),
- statistics_mode_(statistics_mode) {}
+ statistics_mode_(statistics_mode),
+ spill_file_system_(spill_file_system),
+ spill_compression_(spill_compression),
+ spill_compression_level_(spill_compression_level) {
+ if (!temp_directory.empty()) {
+ io_manager_ = std::make_shared<IOManager>(temp_directory,
spill_file_system_);
+ }
+}
Result<std::optional<ArrowRealtimeStore::BatchStatistics>>
ArrowRealtimeStore::CollectStatistics(
const std::shared_ptr<arrow::StructArray>& data) const {
@@ -346,24 +622,28 @@
Result<std::optional<ArrowRealtimeStore::BatchStatistics>> ArrowRealtimeStore::C
}
Result<bool> ArrowRealtimeStore::MayMatchStatistics(
- const StoredBatch& stored, const std::shared_ptr<arrow::Schema>&
read_schema,
+ int64_t row_count, const std::optional<BatchStatistics>& statistics,
+ const std::shared_ptr<arrow::Schema>& read_schema,
const std::shared_ptr<PredicateFilter>& predicate_filter,
const std::vector<int32_t>& statistics_mapping,
const std::shared_ptr<MemoryPool>& memory_pool) {
- if (!predicate_filter || !stored.statistics) {
+ if (!predicate_filter || !statistics) {
return true;
}
- const BatchStatistics& statistics = stored.statistics.value();
+ const BatchStatistics& batch_statistics = statistics.value();
std::shared_ptr<InternalRow> min_row = std::make_shared<ColumnarRow>(
- statistics.min_values, statistics.min_values->fields(), memory_pool,
/*row_id=*/0);
+ batch_statistics.min_values, batch_statistics.min_values->fields(),
memory_pool,
+ /*row_id=*/0);
std::shared_ptr<InternalRow> max_row = std::make_shared<ColumnarRow>(
- statistics.max_values, statistics.max_values->fields(), memory_pool,
/*row_id=*/0);
+ batch_statistics.max_values, batch_statistics.max_values->fields(),
memory_pool,
+ /*row_id=*/0);
ProjectedRow projected_min(min_row, statistics_mapping);
ProjectedRow projected_max(max_row, statistics_mapping);
- std::shared_ptr<InternalArray> null_counts =
std::make_shared<ColumnarArray>(
- statistics.null_counts.get(), memory_pool, /*offset=*/0,
statistics.null_counts->length());
+ std::shared_ptr<InternalArray> null_counts =
+ std::make_shared<ColumnarArray>(batch_statistics.null_counts.get(),
memory_pool,
+ /*offset=*/0,
batch_statistics.null_counts->length());
ProjectedArray projected_null_counts(null_counts, statistics_mapping);
- return predicate_filter->Test(read_schema, stored.data->length(),
projected_min, projected_max,
+ return predicate_filter->Test(read_schema, row_count, projected_min,
projected_max,
projected_null_counts);
}
@@ -409,19 +689,72 @@ Status ArrowRealtimeStore::Write(RealtimeWriteBatch&&
write_batch) {
}
Result<std::optional<std::shared_ptr<RealtimeSegmentHandle>>>
ArrowRealtimeStore::SealForCommit() {
- std::lock_guard<std::mutex> lock(mutex_);
- if (building_batches_.empty()) {
- return std::optional<std::shared_ptr<RealtimeSegmentHandle>>();
- }
- auto segment = std::make_shared<Segment>(building_range_.value(),
std::move(building_batches_));
- sealed_segments_.push_back(segment);
- sealed_memory_usage_ += building_memory_usage_;
- sealed_row_count_ += building_row_count_;
- building_batches_.clear();
- building_range_.reset();
- building_memory_usage_ = 0;
- building_row_count_ = 0;
- return
std::optional<std::shared_ptr<RealtimeSegmentHandle>>(std::move(segment));
+ std::shared_ptr<MemorySegment> memory_segment;
+ {
+ std::lock_guard<std::mutex> lock(mutex_);
+ if (building_batches_.empty()) {
+ return std::optional<std::shared_ptr<RealtimeSegmentHandle>>();
+ }
+ memory_segment =
+ std::make_shared<MemorySegment>(building_range_.value(),
std::move(building_batches_));
+ sealed_segments_.push_back(memory_segment);
+ sealed_memory_usage_ += building_memory_usage_;
+ sealed_row_count_ += building_row_count_;
+ building_batches_.clear();
+ building_range_.reset();
+ building_memory_usage_ = 0;
+ building_row_count_ = 0;
+ }
+
+ if (!io_manager_) {
+ return
std::optional<std::shared_ptr<RealtimeSegmentHandle>>(memory_segment);
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(FileIOChannel::ID channel_id,
io_manager_->CreateChannel("realtime"));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<ArrowIpcFileWriter> writer,
+ ArrowIpcFileWriter::Create(spill_file_system_, channel_id.GetPath(),
write_schema_,
+ spill_compression_,
spill_compression_level_,
+ /*use_threads=*/false, arrow_pool_));
+ auto cleanup_guard = ScopeGuard([&]() {
+ writer.reset();
+ [[maybe_unused]] Status status =
spill_file_system_->Delete(channel_id.GetPath());
+ });
+ std::vector<SpilledBatch> spilled_batches;
+ spilled_batches.reserve(memory_segment->GetBatches().size());
+ for (const StoredBatch& batch : memory_segment->GetBatches()) {
+ std::shared_ptr<arrow::RecordBatch> record_batch =
+ arrow::RecordBatch::Make(write_schema_, batch.data->length(),
batch.data->fields());
+ PAIMON_RETURN_NOT_OK(writer->WriteBatch(record_batch));
+ uint64_t statistics_memory_usage = 0;
+ if (batch.statistics) {
+ statistics_memory_usage =
+
ArrowUtils::GetArrayMemoryUsage(batch.statistics->min_values->data()) +
+
ArrowUtils::GetArrayMemoryUsage(batch.statistics->max_values->data()) +
+
ArrowUtils::GetArrayMemoryUsage(batch.statistics->null_counts->data());
+ }
+ spilled_batches.push_back(SpilledBatch{batch.offset_range,
batch.statistics,
+ batch.data->length(),
statistics_memory_usage});
+ }
+ PAIMON_RETURN_NOT_OK(writer->Close());
+ writer.reset();
+ auto spill_file =
+ std::make_shared<SpillFile>(io_manager_, spill_file_system_,
std::move(channel_id));
+ auto spilled_segment = std::make_shared<SpilledSegment>(
+ memory_segment->GetOffsetRange(), memory_segment->GetRowCount(),
std::move(spilled_batches),
+ std::move(spill_file));
+ cleanup_guard.Release();
+
+ {
+ std::lock_guard<std::mutex> lock(mutex_);
+ auto iter = std::find(sealed_segments_.begin(),
sealed_segments_.end(), memory_segment);
+ if (iter != sealed_segments_.end()) {
+ *iter = spilled_segment;
+ sealed_memory_usage_ -= memory_segment->GetMemoryUsage();
+ sealed_memory_usage_ += spilled_segment->GetMemoryUsage();
+ }
+ }
+ return
std::optional<std::shared_ptr<RealtimeSegmentHandle>>(std::move(spilled_segment));
}
Result<std::vector<std::unique_ptr<BatchReader>>>
ArrowRealtimeStore::CreateCommitReaders(
@@ -431,13 +764,40 @@ Result<std::vector<std::unique_ptr<BatchReader>>>
ArrowRealtimeStore::CreateComm
return Status::Invalid("segment was not created by the Arrow real-time
store");
}
std::vector<std::unique_ptr<BatchReader>> readers;
+ std::shared_ptr<MemorySegment> memory_segment =
+ std::dynamic_pointer_cast<MemorySegment>(arrow_segment);
+ if (memory_segment) {
+ if (mode_ == RealtimeStoreMode::APPEND_ONLY) {
+
readers.push_back(std::make_unique<AppendCommitBatchReader>(memory_segment));
+ return readers;
+ }
+ readers.reserve(memory_segment->GetBatches().size());
+ for (const StoredBatch& batch : memory_segment->GetBatches()) {
+ readers.push_back(std::make_unique<StoredBatchReader>(batch.data,
arrow_pool_));
+ }
+ return readers;
+ }
+
+ std::shared_ptr<SpilledSegment> spilled_segment =
+ std::dynamic_pointer_cast<SpilledSegment>(arrow_segment);
+ if (!spilled_segment) {
+ return Status::Invalid("unknown Arrow real-time segment type");
+ }
+ const auto batch_count =
static_cast<int32_t>(spilled_segment->GetBatches().size());
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<SharedSpillFileReader> file_reader,
+ SharedSpillFileReader::Create(spilled_segment->GetFile(), batch_count,
arrow_pool_));
if (mode_ == RealtimeStoreMode::APPEND_ONLY) {
-
readers.push_back(std::make_unique<AppendCommitBatchReader>(arrow_segment));
+ std::vector<int32_t> indexes(batch_count);
+ std::iota(indexes.begin(), indexes.end(), 0);
+ readers.push_back(std::make_unique<SpillBatchReader>(file_reader,
std::move(indexes),
+
/*read_schema=*/nullptr, arrow_pool_));
return readers;
}
- readers.reserve(arrow_segment->GetBatches().size());
- for (const StoredBatch& batch : arrow_segment->GetBatches()) {
- readers.push_back(std::make_unique<StoredBatchReader>(batch.data,
arrow_pool_));
+ readers.reserve(spilled_segment->GetBatches().size());
+ for (int32_t i = 0; i < batch_count; ++i) {
+ readers.push_back(std::make_unique<SpillBatchReader>(file_reader,
std::vector<int32_t>{i},
+
/*read_schema=*/nullptr, arrow_pool_));
}
return readers;
}
@@ -446,8 +806,8 @@ Result<std::shared_ptr<RealtimeReadView>>
ArrowRealtimeStore::AcquireReadView()
std::lock_guard<std::mutex> lock(mutex_);
std::vector<std::shared_ptr<Segment>> segments = sealed_segments_;
if (!building_batches_.empty()) {
- segments.push_back(std::make_shared<Segment>(building_range_.value(),
-
std::vector<StoredBatch>(building_batches_)));
+ segments.push_back(std::make_shared<MemorySegment>(
+ building_range_.value(),
std::vector<StoredBatch>(building_batches_)));
}
std::shared_ptr<ReadView> view =
std::make_shared<ReadView>(std::move(segments));
return std::shared_ptr<RealtimeReadView>(std::move(view));
@@ -462,11 +822,14 @@ Result<std::vector<std::unique_ptr<BatchReader>>>
ArrowRealtimeStore::CreateQuer
if (context.read_schema == nullptr || context.read_schema->release ==
nullptr) {
return Status::Invalid("mem query read schema is null");
}
+ if (context.read_batch_size <= 0) {
+ return Status::Invalid("real-time query read batch size must be
positive");
+ }
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema>
read_schema,
arrow::ImportSchema(context.read_schema));
std::vector<std::unique_ptr<BatchReader>> readers;
if (!arrow_view->GetOffsetRange()) {
- return readers;
+ return SliceQueryReaders(std::move(readers), context.read_batch_size,
arrow_pool_);
}
std::shared_ptr<PredicateFilter> predicate_filter;
if (context.predicate) {
@@ -479,49 +842,91 @@ Result<std::vector<std::unique_ptr<BatchReader>>>
ArrowRealtimeStore::CreateQuer
}
if (mode_ == RealtimeStoreMode::PRIMARY_KEY) {
for (const std::shared_ptr<Segment>& segment :
arrow_view->GetSegments()) {
- for (const StoredBatch& batch : segment->GetBatches()) {
- PAIMON_ASSIGN_OR_RAISE(bool may_match,
- MayMatchStatistics(batch, read_schema,
predicate_filter,
- statistics_mapping,
memory_pool_));
- if (!may_match) {
- continue;
+ std::shared_ptr<MemorySegment> memory_segment =
+ std::dynamic_pointer_cast<MemorySegment>(segment);
+ if (memory_segment) {
+ for (const StoredBatch& batch : memory_segment->GetBatches()) {
+ PAIMON_ASSIGN_OR_RAISE(
+ bool may_match,
+ MayMatchStatistics(batch.data->length(),
batch.statistics, read_schema,
+ predicate_filter,
statistics_mapping, memory_pool_));
+ if (!may_match) {
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<arrow::StructArray> projected,
+ ProjectBatch(batch.data, read_schema,
arrow_pool_.get()));
+
readers.push_back(std::make_unique<StoredBatchReader>(projected, arrow_pool_));
+ }
+ continue;
+ }
+
+ std::shared_ptr<SpilledSegment> spilled_segment =
+ std::dynamic_pointer_cast<SpilledSegment>(segment);
+ if (!spilled_segment) {
+ return Status::Invalid("unknown Arrow real-time segment type");
+ }
+ std::vector<int32_t> matching_indexes;
+ for (int32_t i = 0; i <
static_cast<int32_t>(spilled_segment->GetBatches().size());
+ ++i) {
+ const SpilledBatch& batch = spilled_segment->GetBatches()[i];
+ PAIMON_ASSIGN_OR_RAISE(
+ bool may_match,
+ MayMatchStatistics(batch.row_count, batch.statistics,
read_schema,
+ predicate_filter, statistics_mapping,
memory_pool_));
+ if (may_match) {
+ matching_indexes.push_back(i);
}
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::StructArray>
projected,
- ProjectBatch(batch.data, read_schema,
arrow_pool_.get()));
-
readers.push_back(std::make_unique<StoredBatchReader>(projected, arrow_pool_));
+ }
+ if (matching_indexes.empty()) {
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<SharedSpillFileReader> file_reader,
+ SharedSpillFileReader::Create(
+ spilled_segment->GetFile(),
+
static_cast<int32_t>(spilled_segment->GetBatches().size()), arrow_pool_));
+ for (int32_t batch_index : matching_indexes) {
+ readers.push_back(std::make_unique<SpillBatchReader>(
+ file_reader, std::vector<int32_t>{batch_index},
read_schema, arrow_pool_));
}
}
- return readers;
+ return SliceQueryReaders(std::move(readers), context.read_batch_size,
arrow_pool_);
}
std::unique_ptr<BatchReader> reader =
std::make_unique<AppendQueryBatchReader>(
arrow_view, read_schema, predicate_filter,
std::move(statistics_mapping), arrow_pool_,
memory_pool_);
readers.push_back(std::move(reader));
- return readers;
+ return SliceQueryReaders(std::move(readers), context.read_batch_size,
arrow_pool_);
}
Status ArrowRealtimeStore::AdvanceCommittedOffset(int64_t
committed_end_offset) {
- std::lock_guard<std::mutex> lock(mutex_);
- // TODO(xinyu.lxy): Consider deferring segment destruction to a
reclamation queue. Existing
- // read views may pin reclaimed batches, so the last query releasing a
view can otherwise pay
- // the full buffer destruction cost and observe higher tail latency.
- uint64_t reclaimed_memory_usage = 0;
- uint64_t reclaimed_row_count = 0;
- for (const std::shared_ptr<Segment>& segment : sealed_segments_) {
- if (segment->GetOffsetRange().end <= committed_end_offset) {
- reclaimed_memory_usage += segment->GetMemoryUsage();
- reclaimed_row_count +=
static_cast<uint64_t>(segment->GetRowCount());
+ // Drop the store's references only after releasing the write mutex.
Segment destruction may
+ // release large Arrow buffers or delete a spill file.
+ std::vector<std::shared_ptr<Segment>> reclaimed_segments;
+ {
+ std::lock_guard<std::mutex> lock(mutex_);
+ reclaimed_segments.reserve(sealed_segments_.size());
+ uint64_t reclaimed_memory_usage = 0;
+ uint64_t reclaimed_row_count = 0;
+ auto retained_end = sealed_segments_.begin();
+ for (auto iter = sealed_segments_.begin(); iter !=
sealed_segments_.end(); ++iter) {
+ if ((*iter)->GetOffsetRange().end <= committed_end_offset) {
+ reclaimed_memory_usage += (*iter)->GetMemoryUsage();
+ reclaimed_row_count +=
static_cast<uint64_t>((*iter)->GetRowCount());
+ reclaimed_segments.push_back(std::move(*iter));
+ continue;
+ }
+ if (retained_end != iter) {
+ *retained_end = std::move(*iter);
+ }
+ ++retained_end;
}
+ sealed_segments_.erase(retained_end, sealed_segments_.end());
+ sealed_memory_usage_ -= reclaimed_memory_usage;
+ sealed_row_count_ -= reclaimed_row_count;
}
- sealed_segments_.erase(
- std::remove_if(sealed_segments_.begin(), sealed_segments_.end(),
- [committed_end_offset](const std::shared_ptr<Segment>&
segment) {
- return segment->GetOffsetRange().end <=
committed_end_offset;
- }),
- sealed_segments_.end());
- sealed_memory_usage_ -= reclaimed_memory_usage;
- sealed_row_count_ -= reclaimed_row_count;
return Status::OK();
}
diff --git a/src/paimon/core/realtime/arrow_realtime_store.h
b/src/paimon/core/realtime/arrow_realtime_store.h
index 630744f5..e92a7199 100644
--- a/src/paimon/core/realtime/arrow_realtime_store.h
+++ b/src/paimon/core/realtime/arrow_realtime_store.h
@@ -23,6 +23,7 @@
#include <memory>
#include <mutex>
#include <optional>
+#include <string>
#include <vector>
#include "paimon/realtime/realtime_store.h"
@@ -35,6 +36,8 @@ class StructArray;
} // namespace arrow
namespace paimon {
+class FileSystem;
+class IOManager;
class MemoryPool;
class PredicateFilter;
@@ -42,7 +45,9 @@ class PredicateFilter;
class ArrowRealtimeStore final : public RealtimeStore {
public:
ArrowRealtimeStore(const std::shared_ptr<arrow::Schema>& write_schema,
RealtimeStoreMode mode,
- StatisticsMode statistics_mode,
+ StatisticsMode statistics_mode, const std::string&
temp_directory,
+ const std::shared_ptr<FileSystem>& spill_file_system,
+ const std::string& spill_compression, int32_t
spill_compression_level,
const std::shared_ptr<MemoryPool>& memory_pool,
const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
@@ -82,16 +87,29 @@ class ArrowRealtimeStore final : public RealtimeStore {
uint64_t memory_usage;
};
+ struct SpilledBatch {
+ OffsetRange offset_range;
+ std::optional<BatchStatistics> statistics;
+ int64_t row_count;
+ uint64_t memory_usage;
+ };
+
class Segment;
+ class MemorySegment;
+ class SpilledSegment;
+ class SpillFile;
class ReadView;
class AppendCommitBatchReader;
class StoredBatchReader;
+ class SharedSpillFileReader;
+ class SpillBatchReader;
class AppendQueryBatchReader;
Result<std::optional<BatchStatistics>> CollectStatistics(
const std::shared_ptr<arrow::StructArray>& data) const;
- static Result<bool> MayMatchStatistics(const StoredBatch& stored,
+ static Result<bool> MayMatchStatistics(int64_t row_count,
+ const
std::optional<BatchStatistics>& statistics,
const
std::shared_ptr<arrow::Schema>& read_schema,
const
std::shared_ptr<PredicateFilter>& predicate_filter,
const std::vector<int32_t>&
statistics_mapping,
@@ -102,6 +120,10 @@ class ArrowRealtimeStore final : public RealtimeStore {
std::shared_ptr<arrow::MemoryPool> arrow_pool_;
RealtimeStoreMode mode_;
StatisticsMode statistics_mode_;
+ std::shared_ptr<IOManager> io_manager_;
+ std::shared_ptr<FileSystem> spill_file_system_;
+ std::string spill_compression_;
+ int32_t spill_compression_level_;
mutable std::mutex mutex_;
std::vector<StoredBatch> building_batches_;
std::vector<std::shared_ptr<Segment>> sealed_segments_;
diff --git a/src/paimon/core/realtime/arrow_realtime_store_factory.cpp
b/src/paimon/core/realtime/arrow_realtime_store_factory.cpp
index d42d3791..4777ee35 100644
--- a/src/paimon/core/realtime/arrow_realtime_store_factory.cpp
+++ b/src/paimon/core/realtime/arrow_realtime_store_factory.cpp
@@ -19,11 +19,14 @@
#include "paimon/realtime/arrow_realtime_store_factory.h"
+#include <string>
+
#include "arrow/c/bridge.h"
#include "arrow/c/helpers.h"
#include "paimon/common/utils/arrow/mem_utils.h"
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/core_options.h"
#include "paimon/core/realtime/arrow_realtime_store.h"
#include "paimon/macros.h"
@@ -47,8 +50,23 @@ Result<std::shared_ptr<RealtimeStore>>
ArrowRealtimeStoreFactory::Create(
static_cast<int32_t>(request.mode));
}
std::shared_ptr<arrow::MemoryPool> arrow_pool =
GetArrowPool(request.memory_pool);
+ PAIMON_ASSIGN_OR_RAISE(CoreOptions options,
CoreOptions::FromMap(request.options));
+ const CompressOptions& spill_compression =
options.GetSpillCompressOptions();
+ std::string spill_directory;
+ if (options.RealtimeSpillEnabled()) {
+ if (request.temp_directory.empty()) {
+ return Status::Invalid(
+ "realtime.spill-enabled requires a non-empty temporary
directory");
+ }
+ if (!request.file_system) {
+ return Status::Invalid("realtime.spill-enabled requires a file
system");
+ }
+ spill_directory = request.temp_directory;
+ }
return std::make_shared<ArrowRealtimeStore>(
- imported_schema, request.mode, request.statistics_mode,
request.memory_pool, arrow_pool);
+ imported_schema, request.mode, request.statistics_mode,
spill_directory,
+ request.file_system, spill_compression.compress,
spill_compression.zstd_level,
+ request.memory_pool, arrow_pool);
}
} // namespace paimon
diff --git a/src/paimon/core/realtime/arrow_realtime_store_test.cpp
b/src/paimon/core/realtime/arrow_realtime_store_test.cpp
index 430f8b27..c8d988e3 100644
--- a/src/paimon/core/realtime/arrow_realtime_store_test.cpp
+++ b/src/paimon/core/realtime/arrow_realtime_store_test.cpp
@@ -32,11 +32,13 @@
#include "paimon/common/utils/arrow/mem_utils.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/core/realtime/realtime_offset_batch_reader.h"
+#include "paimon/defs.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/predicate/literal.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/realtime/arrow_realtime_store_factory.h"
#include "paimon/record_batch.h"
+#include "paimon/testing/utils/read_result_collector.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
@@ -74,8 +76,11 @@ class ArrowRealtimeStoreTest : public testing::Test {
}
std::shared_ptr<ArrowRealtimeStore> CreateStore(StatisticsMode
statistics_mode) const {
- return std::make_shared<ArrowRealtimeStore>(schema_,
RealtimeStoreMode::APPEND_ONLY,
- statistics_mode, pool_,
arrow_pool_);
+ return std::make_shared<ArrowRealtimeStore>(
+ schema_, RealtimeStoreMode::APPEND_ONLY, statistics_mode,
/*temp_directory=*/"",
+ /*spill_file_system=*/nullptr,
+ /*spill_compression=*/"zstd",
+ /*spill_compression_level=*/1, pool_, arrow_pool_);
}
std::unique_ptr<RecordBatch> MakeBatch(const std::string& json) const {
@@ -105,17 +110,16 @@ class ArrowRealtimeStoreTest : public testing::Test {
return c_schema;
}
- std::vector<int64_t> ReadIds(const BatchReader::ReadBatchWithBitmap&
batch) const {
- std::shared_ptr<arrow::Array> array =
- arrow::ImportArray(batch.first.first.get(),
batch.first.second.get()).ValueOrDie();
- std::shared_ptr<arrow::StructArray> struct_array =
- checked_pointer_cast<arrow::StructArray>(array);
- std::shared_ptr<arrow::Int64Array> ids =
-
checked_pointer_cast<arrow::Int64Array>(struct_array->GetFieldByName("id"));
+ std::vector<int64_t> ReadIds(const std::shared_ptr<arrow::ChunkedArray>&
chunked_array) const {
std::vector<int64_t> result;
- for (RoaringBitmap32::Iterator iter = batch.second.Begin(); iter !=
batch.second.End();
- ++iter) {
- result.push_back(ids->Value(*iter));
+ for (const std::shared_ptr<arrow::Array>& chunk :
chunked_array->chunks()) {
+ std::shared_ptr<arrow::StructArray> struct_array =
+ checked_pointer_cast<arrow::StructArray>(chunk);
+ std::shared_ptr<arrow::Int64Array> ids =
+
checked_pointer_cast<arrow::Int64Array>(struct_array->GetFieldByName("id"));
+ for (int64_t i = 0; i < ids->length(); ++i) {
+ result.push_back(ids->Value(i));
+ }
}
return result;
}
@@ -221,10 +225,9 @@ TEST_F(ArrowRealtimeStoreTest,
TestQueryReaderClipsCommittedOffsetWithBitmap) {
RealtimeQueryContext context{c_schema.get(), /*predicate=*/nullptr};
ASSERT_OK_AND_ASSIGN(std::vector<std::unique_ptr<BatchReader>> readers,
store_->CreateQueryReaders(view, context));
- ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap batch,
- readers[0]->NextBatchWithBitmap());
+ ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch batch,
readers[0]->NextBatch());
arrow::Result<std::shared_ptr<arrow::Array>> import_result =
- arrow::ImportArray(batch.first.first.get(),
batch.first.second.get());
+ arrow::ImportArray(batch.first.get(), batch.second.get());
ASSERT_TRUE(import_result.ok());
std::shared_ptr<arrow::Array> array =
std::move(import_result).ValueOrDie();
ASSERT_TRUE(array->type()->Equals(arrow::struct_(read_schema->fields())));
@@ -300,6 +303,104 @@ TEST_F(ArrowRealtimeStoreTest,
TestCommitReaderPreservesSlicedBatch) {
<< "expected: " << expected_array->ToString() << ", actual: " <<
actual_array->ToString();
}
+TEST_F(ArrowRealtimeStoreTest, TestSealSpillsAndKeepsPinnedMemoryViewReadable)
{
+ std::unique_ptr<UniqueTestDirectory> temp_directory =
UniqueTestDirectory::Create();
+ ASSERT_NE(nullptr, temp_directory);
+ ArrowRealtimeStoreFactory factory;
+ std::unique_ptr<ArrowSchema> write_schema = MakeReadSchema(schema_);
+ RealtimeStoreCreateRequest request{std::move(write_schema),
+ {{Options::REALTIME_SPILL_ENABLED,
"true"}},
+ pool_,
+ RealtimeStoreMode::APPEND_ONLY,
+ StatisticsMode::NONE};
+ request.temp_directory = temp_directory->Str();
+ request.file_system = temp_directory->GetFileSystem();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeStore> realtime_store,
+ factory.Create(std::move(request)));
+ std::shared_ptr<ArrowRealtimeStore> store =
+ std::dynamic_pointer_cast<ArrowRealtimeStore>(realtime_store);
+ ASSERT_NE(nullptr, store);
+ ASSERT_OK(store->Write(
+ RealtimeWriteBatch{MakeBatch(R"([[0, 1, "a"], [1, 2, "b"]])"),
OffsetRange(0, 2)}));
+ ASSERT_OK(store->Write(RealtimeWriteBatch{MakeBatch(R"([[2, 3, "c"]])"),
OffsetRange(2, 3)}));
+ ASSERT_GT(store->GetMemoryUsage(), 0);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeReadView> pinned_view,
store->AcquireReadView());
+
+ ASSERT_OK_AND_ASSIGN(std::optional<std::shared_ptr<RealtimeSegmentHandle>>
segment,
+ store->SealForCommit());
+ ASSERT_TRUE(segment.has_value());
+ ASSERT_EQ(0, store->GetMemoryUsage());
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::unique_ptr<BatchReader>>
commit_readers,
+ store->CreateCommitReaders(segment.value()));
+ ASSERT_EQ(1, commit_readers.size());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> committed,
+
ReadResultCollector::CollectResult(std::move(commit_readers[0])));
+ ASSERT_EQ(std::vector<int64_t>({1, 2, 3}), ReadIds(committed));
+
+ std::unique_ptr<ArrowSchema> read_schema = MakeReadSchema(schema_);
+ RealtimeQueryContext context{read_schema.get(), /*predicate=*/nullptr,
+ /*read_batch_size=*/1};
+ ASSERT_OK_AND_ASSIGN(std::vector<std::unique_ptr<BatchReader>>
query_readers,
+ store->CreateQueryReaders(pinned_view, context));
+ ASSERT_EQ(1, query_readers.size());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> pinned,
+
ReadResultCollector::CollectResult(std::move(query_readers[0])));
+ ASSERT_EQ(std::vector<int64_t>({1, 2, 3}), ReadIds(pinned));
+ ASSERT_EQ(3, pinned->num_chunks());
+ for (const std::shared_ptr<arrow::Array>& chunk : pinned->chunks()) {
+ ASSERT_EQ(1, chunk->length());
+ }
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeReadView> spilled_view,
store->AcquireReadView());
+ std::unique_ptr<ArrowSchema> spilled_read_schema = MakeReadSchema(schema_);
+ RealtimeQueryContext spilled_context{spilled_read_schema.get(),
/*predicate=*/nullptr,
+ /*read_batch_size=*/1};
+ ASSERT_OK_AND_ASSIGN(std::vector<std::unique_ptr<BatchReader>>
spilled_readers,
+ store->CreateQueryReaders(spilled_view,
spilled_context));
+ ASSERT_EQ(1, spilled_readers.size());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> spilled,
+
ReadResultCollector::CollectResult(std::move(spilled_readers[0])));
+ ASSERT_EQ(std::vector<int64_t>({1, 2, 3}), ReadIds(spilled));
+ ASSERT_EQ(3, spilled->num_chunks());
+ for (const std::shared_ptr<arrow::Array>& chunk : spilled->chunks()) {
+ ASSERT_EQ(1, chunk->length());
+ }
+}
+
+TEST_F(ArrowRealtimeStoreTest, TestFactoryDoesNotSpillWithoutOption) {
+ std::unique_ptr<UniqueTestDirectory> temp_directory =
UniqueTestDirectory::Create();
+ ASSERT_NE(nullptr, temp_directory);
+ ArrowRealtimeStoreFactory factory;
+ std::unique_ptr<ArrowSchema> write_schema = MakeReadSchema(schema_);
+ RealtimeStoreCreateRequest request{std::move(write_schema),
/*options=*/{}, pool_,
+ RealtimeStoreMode::APPEND_ONLY,
StatisticsMode::NONE};
+ request.temp_directory = temp_directory->Str();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeStore> realtime_store,
+ factory.Create(std::move(request)));
+ std::shared_ptr<ArrowRealtimeStore> store =
+ std::dynamic_pointer_cast<ArrowRealtimeStore>(realtime_store);
+ ASSERT_NE(nullptr, store);
+ ASSERT_OK(store->Write(RealtimeWriteBatch{MakeBatch(R"([[0, 1, "a"]])"),
OffsetRange(0, 1)}));
+
+ ASSERT_OK_AND_ASSIGN(std::optional<std::shared_ptr<RealtimeSegmentHandle>>
segment,
+ store->SealForCommit());
+ ASSERT_TRUE(segment.has_value());
+ ASSERT_GT(store->GetMemoryUsage(), 0);
+}
+
+TEST_F(ArrowRealtimeStoreTest, TestFactoryRejectsSpillWithoutTempDirectory) {
+ ArrowRealtimeStoreFactory factory;
+ std::unique_ptr<ArrowSchema> write_schema = MakeReadSchema(schema_);
+ RealtimeStoreCreateRequest request{std::move(write_schema),
+ {{Options::REALTIME_SPILL_ENABLED,
"true"}},
+ pool_,
+ RealtimeStoreMode::APPEND_ONLY,
+ StatisticsMode::NONE};
+ ASSERT_NOK_WITH_MSG(factory.Create(std::move(request)),
+ "realtime.spill-enabled requires a non-empty temporary
directory");
+}
+
TEST_F(ArrowRealtimeStoreTest, TestFullStatisticsPrunesNonMatchingBatch) {
ArrowRealtimeStoreFactory factory;
std::unique_ptr<ArrowSchema> write_schema = MakeReadSchema(schema_);
@@ -327,20 +428,18 @@ TEST_F(ArrowRealtimeStoreTest,
TestFullStatisticsPrunesNonMatchingBatch) {
readers[0] =
std::make_unique<RealtimeOffsetBatchReader>(std::move(readers[0]),
OffsetRange(3, 4));
- ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap batch,
readers[0]->NextBatchWithBitmap());
- ASSERT_FALSE(BatchReader::IsEofBatch(batch));
- ASSERT_EQ(std::vector<int64_t>({11}), ReadIds(batch));
- ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap eof,
readers[0]->NextBatchWithBitmap());
- ASSERT_TRUE(BatchReader::IsEofBatch(eof));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> filtered,
+
ReadResultCollector::CollectResult(std::move(readers[0])));
+ ASSERT_EQ(std::vector<int64_t>({11}), ReadIds(filtered));
std::unique_ptr<ArrowSchema> unfiltered_read_schema =
MakeReadSchema(schema_);
RealtimeQueryContext unfiltered_context{unfiltered_read_schema.get(),
/*predicate=*/nullptr};
ASSERT_OK_AND_ASSIGN(std::vector<std::unique_ptr<BatchReader>>
unfiltered_readers,
store->CreateQueryReaders(view, unfiltered_context));
ASSERT_EQ(1, unfiltered_readers.size());
- ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap unfiltered_batch,
- unfiltered_readers[0]->NextBatchWithBitmap());
- ASSERT_EQ(std::vector<int64_t>({0, 1}), ReadIds(unfiltered_batch));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> unfiltered,
+
ReadResultCollector::CollectResult(std::move(unfiltered_readers[0])));
+ ASSERT_EQ(std::vector<int64_t>({0, 1, 10, 11}), ReadIds(unfiltered));
}
TEST_F(ArrowRealtimeStoreTest, TestMissingStatisticsRetainsNonMatchingBatch) {
@@ -358,9 +457,9 @@ TEST_F(ArrowRealtimeStoreTest,
TestMissingStatisticsRetainsNonMatchingBatch) {
store_->CreateQueryReaders(view, context));
ASSERT_EQ(1, readers.size());
- ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap batch,
readers[0]->NextBatchWithBitmap());
- ASSERT_FALSE(BatchReader::IsEofBatch(batch));
- ASSERT_EQ(std::vector<int64_t>({0, 1}), ReadIds(batch));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> retained,
+
ReadResultCollector::CollectResult(std::move(readers[0])));
+ ASSERT_EQ(std::vector<int64_t>({0, 1, 10, 11}), ReadIds(retained));
}
TEST_F(ArrowRealtimeStoreTest,
TestRejectsHandlesFromAnotherStoreImplementation) {
diff --git a/src/paimon/core/realtime/primary_key_realtime_store_test.cpp
b/src/paimon/core/realtime/primary_key_realtime_store_test.cpp
index e4e85ede..d393f210 100644
--- a/src/paimon/core/realtime/primary_key_realtime_store_test.cpp
+++ b/src/paimon/core/realtime/primary_key_realtime_store_test.cpp
@@ -74,9 +74,12 @@ std::shared_ptr<arrow::Schema> NestedStoreWriteSchema() {
std::shared_ptr<ArrowRealtimeStore> CreateStore(
const std::shared_ptr<arrow::Schema>& schema, const
std::shared_ptr<MemoryPool>& pool,
- StatisticsMode statistics_mode = StatisticsMode::NONE) {
- return std::make_shared<ArrowRealtimeStore>(schema,
RealtimeStoreMode::PRIMARY_KEY,
- statistics_mode, pool,
GetArrowPool(pool));
+ StatisticsMode statistics_mode = StatisticsMode::NONE, const std::string&
temp_directory = "",
+ const std::shared_ptr<FileSystem>& spill_file_system = nullptr) {
+ return std::make_shared<ArrowRealtimeStore>(
+ schema, RealtimeStoreMode::PRIMARY_KEY, statistics_mode,
temp_directory, spill_file_system,
+ /*spill_compression=*/"zstd",
+ /*spill_compression_level=*/1, pool, GetArrowPool(pool));
}
std::unique_ptr<RecordBatch> MakeBatch(const std::string& json) {
@@ -182,6 +185,66 @@ TEST(PrimaryKeyRealtimeStoreTest,
TestCommitReaderPerStoredBatch) {
ASSERT_TRUE(expected->Equals(actual));
}
+TEST(PrimaryKeyRealtimeStoreTest,
TestSpilledFileReturnsOneIndependentReaderPerStoredBatch) {
+ std::unique_ptr<UniqueTestDirectory> temp_directory =
UniqueTestDirectory::Create();
+ ASSERT_NE(nullptr, temp_directory);
+ std::shared_ptr<ArrowRealtimeStore> store =
+ CreateStore(StoreWriteSchema(), GetDefaultPool(), StatisticsMode::NONE,
+ temp_directory->Str(), temp_directory->GetFileSystem());
+ ASSERT_OK(store->Write(RealtimeWriteBatch{
+ MakeBatch(R"([[6, 1, 1, 1, "before"], [5, 0, 0, 3, "three"]])"),
OffsetRange(0, 2)}));
+ ASSERT_OK(store->Write(
+ RealtimeWriteBatch{MakeBatch(R"([[7, 2, 2, 2, "after"]])"),
OffsetRange(2, 3)}));
+
+ ASSERT_OK_AND_ASSIGN(std::optional<std::shared_ptr<RealtimeSegmentHandle>>
segment,
+ store->SealForCommit());
+ ASSERT_TRUE(segment.has_value());
+ ASSERT_EQ(0, store->GetMemoryUsage());
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::unique_ptr<BatchReader>>
commit_readers,
+ store->CreateCommitReaders(segment.value()));
+ ASSERT_EQ(2, commit_readers.size());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> committed,
+ ReadArray(std::move(commit_readers)));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<arrow::Array> expected,
+ ArrayFromJson(committed->type(),
+ R"([[6, 1, 1, 1, "before"], [5, 0, 0, 3, "three"], [7,
2, 2, 2, "after"]])"));
+ ASSERT_TRUE(expected->Equals(committed));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeReadView> view,
store->AcquireReadView());
+ auto c_schema = std::make_unique<ArrowSchema>();
+ ASSERT_TRUE(arrow::ExportSchema(*StoreWriteSchema(), c_schema.get()).ok());
+ RealtimeQueryContext context{c_schema.get(), /*predicate=*/nullptr,
+ /*read_batch_size=*/1};
+ ASSERT_OK_AND_ASSIGN(std::vector<std::unique_ptr<BatchReader>>
query_readers,
+ store->CreateQueryReaders(view, context));
+ ASSERT_EQ(2, query_readers.size());
+
+ ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch first,
query_readers[0]->NextBatch());
+ ASSERT_FALSE(BatchReader::IsEofBatch(first));
+ AssertOffsetsZero(first.first.get());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> first_array,
+ ReadResultCollector::GetArray(std::move(first)));
+ ASSERT_EQ(1, first_array->length());
+ ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch independent,
query_readers[1]->NextBatch());
+ ASSERT_FALSE(BatchReader::IsEofBatch(independent));
+ AssertOffsetsZero(independent.first.get());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> independent_array,
+
ReadResultCollector::GetArray(std::move(independent)));
+ ASSERT_EQ(1, independent_array->length());
+ ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch second,
query_readers[0]->NextBatch());
+ ASSERT_FALSE(BatchReader::IsEofBatch(second));
+ AssertOffsetsZero(second.first.get());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> second_array,
+ ReadResultCollector::GetArray(std::move(second)));
+ ASSERT_EQ(1, second_array->length());
+ ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch first_eof,
query_readers[0]->NextBatch());
+ ASSERT_TRUE(BatchReader::IsEofBatch(first_eof));
+ ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch independent_eof,
query_readers[1]->NextBatch());
+ ASSERT_TRUE(BatchReader::IsEofBatch(independent_eof));
+}
+
void AssertSlicedBatch(BatchReader* reader) {
ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch batch, reader->NextBatch());
ASSERT_FALSE(BatchReader::IsEofBatch(batch));
diff --git a/src/paimon/core/realtime/realtime_append_only_writer.cpp
b/src/paimon/core/realtime/realtime_append_only_writer.cpp
index f205d773..6b89fd51 100644
--- a/src/paimon/core/realtime/realtime_append_only_writer.cpp
+++ b/src/paimon/core/realtime/realtime_append_only_writer.cpp
@@ -52,7 +52,7 @@ Result<std::shared_ptr<RealtimeAppendOnlyWriter>>
RealtimeAppendOnlyWriter::Crea
const std::shared_ptr<RealtimeContext>& realtime_context,
const std::shared_ptr<AppendOnlyWriter>& file_writer,
const std::shared_ptr<RealtimeSchemaLayout>& schema_layout, const
CoreOptions& options,
- const std::shared_ptr<MemoryPool>& memory_pool) {
+ const std::string& temp_directory, const std::shared_ptr<MemoryPool>&
memory_pool) {
if (!realtime_context) {
return Status::Invalid("real-time context is null");
}
@@ -67,6 +67,10 @@ Result<std::shared_ptr<RealtimeAppendOnlyWriter>>
RealtimeAppendOnlyWriter::Crea
RealtimeStoreCreateRequest request{std::move(write_schema),
options.ToMap(), memory_pool,
RealtimeStoreMode::APPEND_ONLY,
options.GetRealtimeStoreStatisticsMode()};
+ if (options.RealtimeSpillEnabled()) {
+ request.temp_directory = temp_directory;
+ request.file_system = options.GetFileSystem();
+ }
PAIMON_ASSIGN_OR_RAISE(RealtimeStoreState store_state,
realtime_context_impl->GetOrCreateRealtimeStore(
std::move(request),
RealtimePartitionBucket(partition, bucket)));
diff --git a/src/paimon/core/realtime/realtime_append_only_writer.h
b/src/paimon/core/realtime/realtime_append_only_writer.h
index cf27245f..c007e337 100644
--- a/src/paimon/core/realtime/realtime_append_only_writer.h
+++ b/src/paimon/core/realtime/realtime_append_only_writer.h
@@ -49,7 +49,7 @@ class RealtimeAppendOnlyWriter : public BatchWriter {
const std::shared_ptr<RealtimeContext>& realtime_context,
const std::shared_ptr<AppendOnlyWriter>& file_writer,
const std::shared_ptr<RealtimeSchemaLayout>& schema_layout, const
CoreOptions& options,
- const std::shared_ptr<MemoryPool>& memory_pool);
+ const std::string& temp_directory, const std::shared_ptr<MemoryPool>&
memory_pool);
Status Write(std::unique_ptr<RecordBatch>&& batch) override;
diff --git a/src/paimon/core/realtime/realtime_context_impl.cpp
b/src/paimon/core/realtime/realtime_context_impl.cpp
index 72a5dd13..6b476617 100644
--- a/src/paimon/core/realtime/realtime_context_impl.cpp
+++ b/src/paimon/core/realtime/realtime_context_impl.cpp
@@ -178,6 +178,12 @@ Result<RealtimeStoreState>
RealtimeContextImpl::GetOrCreateRealtimeStore(
"the RealtimeContext",
PartitionToString(partition_bucket.partition),
partition_bucket.bucket));
}
+ if (iter->second.temp_directory != request.temp_directory) {
+ return Status::Invalid(fmt::format(
+ "real-time store temporary directory mismatch for partition
{}, bucket {}; "
+ "recreate the RealtimeContext",
+ PartitionToString(partition_bucket.partition),
partition_bucket.bucket));
+ }
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RealtimeReadView> read_view,
iter->second.store->AcquireReadView());
if (!read_view) {
@@ -202,12 +208,14 @@ Result<RealtimeStoreState>
RealtimeContextImpl::GetOrCreateRealtimeStore(
PAIMON_RETURN_NOT_OK_FROM_ARROW(
arrow::ExportSchema(*requested_schema, request.write_schema.get()));
RealtimeStoreMode mode = request.mode;
+ std::string temp_directory = request.temp_directory;
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RealtimeStore> store,
factory_->Create(std::move(request)));
if (!store) {
return Status::Invalid("real-time store factory returned a null
store");
}
- stores_.emplace(partition_bucket, StoreEntry{store, requested_schema,
mode});
+ stores_.emplace(partition_bucket,
+ StoreEntry{store, requested_schema, mode,
std::move(temp_directory)});
if (offset_iter != committed_offsets_.end()) {
reclaimed_offsets_.emplace(partition_bucket, offset_iter->second);
}
diff --git a/src/paimon/core/realtime/realtime_context_impl.h
b/src/paimon/core/realtime/realtime_context_impl.h
index 6c7c8d0b..2286925f 100644
--- a/src/paimon/core/realtime/realtime_context_impl.h
+++ b/src/paimon/core/realtime/realtime_context_impl.h
@@ -114,6 +114,7 @@ class PAIMON_EXPORT RealtimeContextImpl final : public
RealtimeContext {
std::shared_ptr<RealtimeStore> store;
std::shared_ptr<arrow::Schema> write_schema;
RealtimeStoreMode mode;
+ std::string temp_directory;
int64_t materialized_max_sequence_number = -1;
};
diff --git a/src/paimon/core/realtime/realtime_context_test.cpp
b/src/paimon/core/realtime/realtime_context_test.cpp
index af2fe717..c691fd51 100644
--- a/src/paimon/core/realtime/realtime_context_test.cpp
+++ b/src/paimon/core/realtime/realtime_context_test.cpp
@@ -130,11 +130,12 @@ Result<RealtimeStoreState> GetOrCreateAppendStore(
const std::map<std::string, std::string>& partition, int32_t bucket,
std::unique_ptr<ArrowSchema> write_schema, const std::map<std::string,
std::string>& options,
const std::shared_ptr<MemoryPool>& memory_pool,
- StatisticsMode statistics_mode = StatisticsMode::NONE) {
- return context->GetOrCreateRealtimeStore(
- RealtimeStoreCreateRequest{std::move(write_schema), options,
memory_pool,
- RealtimeStoreMode::APPEND_ONLY,
statistics_mode},
- RealtimePartitionBucket(partition, bucket));
+ StatisticsMode statistics_mode = StatisticsMode::NONE, const std::string&
temp_directory = "") {
+ RealtimeStoreCreateRequest request{std::move(write_schema), options,
memory_pool,
+ RealtimeStoreMode::APPEND_ONLY,
statistics_mode};
+ request.temp_directory = temp_directory;
+ return context->GetOrCreateRealtimeStore(std::move(request),
+
RealtimePartitionBucket(partition, bucket));
}
TEST(RealtimeContextTest, TestReusesStoreAndCapturesRegisteredViews) {
@@ -226,6 +227,21 @@ TEST(RealtimeContextTest,
TestRejectsMismatchedModeOnStoreReuse) {
ASSERT_EQ(1, factory->stores.size());
}
+TEST(RealtimeContextTest, TestRejectsMismatchedTempDirectoryOnStoreReuse) {
+ auto factory = std::make_shared<TestingRealtimeStoreFactory>();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context,
CreateContext(factory));
+ const std::map<std::string, std::string> partition = {{"dt",
"2026-08-02"}};
+ ASSERT_OK(GetOrCreateAppendStore(context, partition, 0, MakeWriteSchema(),
{}, GetDefaultPool(),
+ StatisticsMode::NONE, "first"));
+
+ ASSERT_NOK_WITH_MSG(
+ GetOrCreateAppendStore(context, partition, 0, MakeWriteSchema(), {},
GetDefaultPool(),
+ StatisticsMode::NONE, "second"),
+ "real-time store temporary directory mismatch for partition
{dt=2026-08-02}, bucket 0; "
+ "recreate the RealtimeContext");
+ ASSERT_EQ(1, factory->stores.size());
+}
+
TEST(RealtimeContextTest, TestRejectsMismatchedSchemaOnStoreReuse) {
auto factory = std::make_shared<TestingRealtimeStoreFactory>();
ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context,
CreateContext(factory));
diff --git a/src/paimon/core/table/source/append_only_table_read.cpp
b/src/paimon/core/table/source/append_only_table_read.cpp
index d5151ed4..b6596e7b 100644
--- a/src/paimon/core/table/source/append_only_table_read.cpp
+++ b/src/paimon/core/table/source/append_only_table_read.cpp
@@ -180,7 +180,8 @@ Result<std::unique_ptr<BatchReader>>
AppendOnlyTableRead::CreateRealtimeReader(
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Predicate> realtime_predicate,
PredicateUtils::CreatePickedFieldFilter(context_->GetPredicate(),
realtime_field_name_to_index));
- RealtimeQueryContext query_context{c_read_schema.get(),
std::move(realtime_predicate)};
+ RealtimeQueryContext query_context{c_read_schema.get(),
std::move(realtime_predicate),
+
context_->GetCoreOptions().GetReadBatchSize()};
PAIMON_ASSIGN_OR_RAISE(std::vector<std::unique_ptr<BatchReader>>
memory_readers,
memory.store->CreateQueryReaders(memory.read_view,
query_context));
const size_t first_memory_reader = readers.size();
diff --git a/src/paimon/core/table/source/key_value_table_read.cpp
b/src/paimon/core/table/source/key_value_table_read.cpp
index 9feae79e..166d0b6d 100644
--- a/src/paimon/core/table/source/key_value_table_read.cpp
+++ b/src/paimon/core/table/source/key_value_table_read.cpp
@@ -120,7 +120,8 @@ Result<std::vector<std::unique_ptr<KeyValueRecordReader>>>
CreateMemoryReaders(
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Predicate> primary_key_predicate,
PredicateUtils::CreatePickedFieldFilter(context->GetPredicate(),
primary_key_name_to_index));
- RealtimeQueryContext query_context{c_schema.get(),
std::move(primary_key_predicate)};
+ RealtimeQueryContext query_context{c_schema.get(),
std::move(primary_key_predicate),
+
context->GetCoreOptions().GetReadBatchSize()};
PAIMON_ASSIGN_OR_RAISE(std::vector<std::unique_ptr<BatchReader>>
batch_readers,
memory.store->CreateQueryReaders(memory.read_view,
query_context));
PAIMON_ASSIGN_OR_RAISE(
diff --git a/test/inte/realtime_write_inte_test.cpp
b/test/inte/realtime_write_inte_test.cpp
index ba05af0f..569a9d03 100644
--- a/test/inte/realtime_write_inte_test.cpp
+++ b/test/inte/realtime_write_inte_test.cpp
@@ -505,8 +505,17 @@ class RealtimeWriteInteTest : public ::testing::Test {
Result<std::unique_ptr<FileStoreWrite>> CreateRealtimeWriter(
const std::shared_ptr<RealtimeContext>& realtime_context) const {
+ return CreateRealtimeWriter(realtime_context, /*temp_directory=*/"");
+ }
+
+ Result<std::unique_ptr<FileStoreWrite>> CreateRealtimeWriter(
+ const std::shared_ptr<RealtimeContext>& realtime_context,
+ const std::string& temp_directory) const {
WriteContextBuilder builder(table_path_, commit_user_);
builder.SetOptions(options_).WithStreamingMode(true).WithRealtimeContext(realtime_context);
+ if (!temp_directory.empty()) {
+ builder.WithTempDirectory(temp_directory);
+ }
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriteContext> context,
builder.Finish());
return FileStoreWrite::Create(std::move(context));
}
@@ -517,6 +526,19 @@ class RealtimeWriteInteTest : public ::testing::Test {
return CreateRealtimeWriter(realtime_context);
}
+ bool WaitForChannelFileCount(const std::string& temp_directory, int64_t
expected_count) const {
+ const auto deadline = std::chrono::steady_clock::now() +
std::chrono::seconds(2);
+ while (std::chrono::steady_clock::now() < deadline) {
+ if (TestHelper::CountChannelFiles(dir_->GetFileSystem(),
temp_directory) ==
+ expected_count) {
+ return true;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
+ return TestHelper::CountChannelFiles(dir_->GetFileSystem(),
temp_directory) ==
+ expected_count;
+ }
+
void ResetExternalOffset(const std::map<std::string, std::string>&
partition, int32_t bucket,
int64_t next_offset) const {
std::lock_guard<std::mutex> lock(external_offsets_mutex_);
@@ -1227,6 +1249,42 @@ TEST_F(RealtimeWriteInteTest, TestAppendCommitAndRead) {
FinalizeCommitAndCheck(writer.get(), /*realtime_commits=*/{},
/*prepare_identifier=*/0, rows);
}
+TEST_F(RealtimeWriteInteTest, TestRealtimeReadBatchSizeSplitsLargeAppendBatch)
{
+ CreateTable(/*partition_keys=*/{});
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
+ RealtimeContext::Create());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> writer,
+ CreateRealtimeWriter(realtime_context));
+
+ constexpr int64_t kRowCount = 1000;
+ const std::vector<Row> expected_rows = MakeRows(/*first_id=*/0, kRowCount,
/*partition=*/"p0");
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch,
+ MakeBatch(expected_rows, /*partitioned=*/false));
+ ASSERT_OK(writer->Write(std::move(batch)));
+
+ for (int32_t read_batch_size : {1, 17, 128, 2048}) {
+ options_[Options::READ_BATCH_SIZE] = std::to_string(read_batch_size);
+ ASSERT_OK_AND_ASSIGN(std::vector<Row> actual_rows,
ReadRows(realtime_context));
+ ASSERT_EQ(expected_rows, actual_rows);
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan,
+ CreatePlan(realtime_context,
/*predicate=*/nullptr));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<arrow::ChunkedArray> result,
+ ReadPlan(plan, realtime_context, {"id", "payload", "pt"},
/*predicate=*/nullptr,
+ /*enable_predicate_filter=*/false));
+ ASSERT_NE(nullptr, result);
+ ASSERT_EQ(kRowCount, result->length());
+ ASSERT_EQ((kRowCount + read_batch_size - 1) / read_batch_size,
result->num_chunks());
+ for (const std::shared_ptr<arrow::Array>& chunk : result->chunks()) {
+ ASSERT_GT(chunk->length(), 0);
+ ASSERT_LE(chunk->length(), read_batch_size);
+ }
+ }
+ FinalizeCommitAndCheck(writer.get(), /*realtime_commits=*/{},
/*prepare_identifier=*/0,
+ expected_rows);
+}
+
TEST_F(RealtimeWriteInteTest,
TestRealtimeMetricsTrackBuildingSealedAndCommittedData) {
CreateTable(/*partition_keys=*/{});
ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
@@ -4237,6 +4295,166 @@ TEST_F(RealtimeWriteInteTest,
TestReaderPinsMemoryAcrossRefresh) {
ASSERT_OK(writer->Close());
}
+TEST_F(RealtimeWriteInteTest, TestAppendRealtimeSpillFileLifecycle) {
+ options_[Options::REALTIME_SPILL_ENABLED] = "true";
+ CreateTable(/*partition_keys=*/{});
+ const std::string temp_directory = PathUtil::JoinPath(dir_->Str(),
"append-realtime-spill");
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
+ RealtimeContext::Create());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> writer,
+ CreateRealtimeWriter(realtime_context,
temp_directory));
+
+ std::vector<Row> first_rows = MakeRows(/*first_id=*/0, /*count=*/4,
/*partition=*/"p0");
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> first_batch,
+ MakeBatch(std::vector<Row>(first_rows.begin(),
first_rows.begin() + 2),
+ /*partitioned=*/false));
+ ASSERT_OK(writer->Write(std::move(first_batch)));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> second_batch,
+ MakeBatch(std::vector<Row>(first_rows.begin() + 2,
first_rows.end()),
+ /*partitioned=*/false));
+ ASSERT_OK(writer->Write(std::move(second_batch)));
+ ASSERT_OK_AND_ASSIGN(RealtimeMetricValues memory_metrics_before_spill,
+ ReadRealtimeMetrics(writer->GetMetrics()));
+ ASSERT_GT(memory_metrics_before_spill.building_memory, 0);
+ ASSERT_EQ(0, memory_metrics_before_spill.sealed_memory);
+ ASSERT_EQ(memory_metrics_before_spill.building_memory,
+ memory_metrics_before_spill.total_memory);
+
+ ASSERT_OK(writer->Seal());
+ ASSERT_EQ(1, TestHelper::CountChannelFiles(dir_->GetFileSystem(),
temp_directory));
+ ASSERT_OK_AND_ASSIGN(RealtimeMetricValues memory_metrics_after_spill,
+ ReadRealtimeMetrics(writer->GetMetrics()));
+ ASSERT_EQ(0, memory_metrics_after_spill.building_memory);
+ ASSERT_EQ(0, memory_metrics_after_spill.sealed_memory);
+ ASSERT_EQ(0, memory_metrics_after_spill.total_memory);
+ ASSERT_EQ(0, memory_metrics_after_spill.building_rows);
+ ASSERT_EQ(4, memory_metrics_after_spill.sealed_rows);
+ ASSERT_EQ(4, memory_metrics_after_spill.total_rows);
+
+ ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress> first_progress,
+
writer->PrepareCommitWithProgress(/*commit_identifier=*/0));
+ ASSERT_EQ(1, first_progress.size());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> pinned_plan,
+ CreatePlan(realtime_context, /*predicate=*/nullptr));
+ ASSERT_EQ(1, pinned_plan->Splits().size());
+ ASSERT_NE(nullptr,
std::dynamic_pointer_cast<RealtimeSplit>(pinned_plan->Splits()[0]));
+ ASSERT_OK_AND_ASSIGN(int64_t first_snapshot_id,
+ Commit(first_progress, /*commit_identifier=*/0));
+
+ ASSERT_OK(writer->RefreshCommittedSnapshot(first_snapshot_id));
+ ASSERT_EQ(1, TestHelper::CountChannelFiles(dir_->GetFileSystem(),
temp_directory));
+ ASSERT_OK_AND_ASSIGN(RealtimeMetricValues memory_metrics_after_reclaim,
+ ReadRealtimeMetrics(writer->GetMetrics()));
+ ASSERT_EQ(0, memory_metrics_after_reclaim.total_memory);
+ ASSERT_EQ(0, memory_metrics_after_reclaim.total_rows);
+ ASSERT_OK_AND_ASSIGN(std::vector<Row> pinned_rows, ReadRows(pinned_plan,
realtime_context));
+ ASSERT_EQ(first_rows, pinned_rows);
+ ASSERT_TRUE(WaitForChannelFileCount(temp_directory, /*expected_count=*/0));
+
+ std::vector<Row> second_rows = MakeRows(/*first_id=*/4, /*count=*/4,
/*partition=*/"p0");
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> third_batch,
+ MakeBatch(std::vector<Row>(second_rows.begin(),
second_rows.begin() + 2),
+ /*partitioned=*/false));
+ ASSERT_OK(writer->Write(std::move(third_batch)));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> fourth_batch,
+ MakeBatch(std::vector<Row>(second_rows.begin() + 2,
second_rows.end()),
+ /*partitioned=*/false));
+ ASSERT_OK(writer->Write(std::move(fourth_batch)));
+ ASSERT_OK(writer->Seal());
+ ASSERT_EQ(1, TestHelper::CountChannelFiles(dir_->GetFileSystem(),
temp_directory));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress> second_progress,
+
writer->PrepareCommitWithProgress(/*commit_identifier=*/1));
+ ASSERT_EQ(1, second_progress.size());
+ ASSERT_OK_AND_ASSIGN(int64_t second_snapshot_id,
+ Commit(second_progress, /*commit_identifier=*/1));
+ ASSERT_OK(writer->RefreshCommittedSnapshot(second_snapshot_id));
+ ASSERT_EQ(0, TestHelper::CountChannelFiles(dir_->GetFileSystem(),
temp_directory));
+
+ std::vector<Row> expected_rows = first_rows;
+ expected_rows.insert(expected_rows.end(), second_rows.begin(),
second_rows.end());
+ ASSERT_OK_AND_ASSIGN(std::vector<Row> actual_rows,
ReadRows(realtime_context));
+ ASSERT_EQ(expected_rows, actual_rows);
+ ASSERT_OK(writer->Close());
+}
+
+TEST_F(RealtimeWriteInteTest, TestPkDvRealtimeSpillFileLifecycle) {
+ options_[Options::FILE_FORMAT] = "parquet";
+ options_[Options::DELETION_VECTORS_ENABLED] = "true";
+ options_[Options::REALTIME_SPILL_ENABLED] = "true";
+ CreatePkTable();
+ const std::string temp_directory = PathUtil::JoinPath(dir_->Str(),
"pk-dv-realtime-spill");
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
+ RealtimeContext::Create());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> writer,
+ CreateRealtimeWriter(realtime_context,
temp_directory));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> first_batch,
+ MakeBatch({Row{1, "old-one", "p0"}, Row{2, "two",
"p0"}},
+ /*partitioned=*/false, /*bucket=*/0,
+ {RecordBatch::RowKind::INSERT,
RecordBatch::RowKind::INSERT}));
+ ASSERT_OK(writer->Write(std::move(first_batch)));
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> second_batch,
+ MakeBatch({Row{1, "new-one", "p0"}, Row{3, "three", "p0"}},
+ /*partitioned=*/false, /*bucket=*/0,
+ {RecordBatch::RowKind::UPDATE_AFTER,
RecordBatch::RowKind::INSERT}));
+ ASSERT_OK(writer->Write(std::move(second_batch)));
+ ASSERT_OK(writer->Seal());
+ ASSERT_EQ(1, TestHelper::CountChannelFiles(dir_->GetFileSystem(),
temp_directory));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress> first_progress,
+
writer->PrepareCommitWithProgress(/*commit_identifier=*/0));
+ ASSERT_EQ(1, first_progress.size());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> pinned_plan,
+ CreatePlan(realtime_context, /*predicate=*/nullptr));
+ ASSERT_EQ(1, pinned_plan->Splits().size());
+ ASSERT_NE(nullptr,
std::dynamic_pointer_cast<RealtimeSplit>(pinned_plan->Splits()[0]));
+ ASSERT_OK_AND_ASSIGN(int64_t first_snapshot_id,
+ Commit(first_progress, /*commit_identifier=*/0));
+
+ ASSERT_OK(writer->RefreshCommittedSnapshot(first_snapshot_id));
+ ASSERT_EQ(1, TestHelper::CountChannelFiles(dir_->GetFileSystem(),
temp_directory));
+ ASSERT_OK_AND_ASSIGN(std::vector<Row> pinned_rows, ReadRows(pinned_plan,
realtime_context));
+ ASSERT_EQ((std::vector<Row>{{1, "new-one", "p0"}, {2, "two", "p0"}, {3,
"three", "p0"}}),
+ pinned_rows);
+ ASSERT_TRUE(WaitForChannelFileCount(temp_directory, /*expected_count=*/0));
+
+ ASSERT_OK_AND_ASSIGN(Snapshot compact_snapshot,
CompactAndCommit(/*partition=*/{}, /*bucket=*/0,
+
/*commit_identifier=*/1));
+ ASSERT_OK(writer->RefreshCommittedSnapshot(compact_snapshot.Id()));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> third_batch,
+ MakeBatch({Row{2, "latest-two", "p0"}},
/*partitioned=*/false,
+ /*bucket=*/0,
{RecordBatch::RowKind::UPDATE_AFTER}));
+ ASSERT_OK(writer->Write(std::move(third_batch)));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> fourth_batch,
+ MakeBatch({Row{3, "deleted-three", "p0"}, Row{4,
"four", "p0"}},
+ /*partitioned=*/false, /*bucket=*/0,
+ {RecordBatch::RowKind::DELETE,
RecordBatch::RowKind::INSERT}));
+ ASSERT_OK(writer->Write(std::move(fourth_batch)));
+ ASSERT_OK(writer->Seal());
+ ASSERT_EQ(1, TestHelper::CountChannelFiles(dir_->GetFileSystem(),
temp_directory));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress> second_progress,
+
writer->PrepareCommitWithProgress(/*commit_identifier=*/2));
+ ASSERT_EQ(1, second_progress.size());
+ ASSERT_OK_AND_ASSIGN(int64_t second_snapshot_id,
+ Commit(second_progress, /*commit_identifier=*/2));
+
+ const std::vector<Row> expected_rows = {
+ {1, "new-one", "p0"}, {2, "latest-two", "p0"}, {4, "four", "p0"}};
+ ASSERT_OK_AND_ASSIGN(std::vector<Row> rows_before_reclaim,
ReadRows(realtime_context));
+ ASSERT_EQ(expected_rows, rows_before_reclaim);
+
+ ASSERT_OK(writer->RefreshCommittedSnapshot(second_snapshot_id));
+ ASSERT_EQ(0, TestHelper::CountChannelFiles(dir_->GetFileSystem(),
temp_directory));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<Row> actual_rows,
ReadRows(realtime_context));
+ ASSERT_EQ(expected_rows, actual_rows);
+ ASSERT_OK(writer->Close());
+}
+
TEST_F(RealtimeWriteInteTest, TestRepeatedCommitReadAndRefresh) {
CreateTable(/*partition_keys=*/{});
ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
