You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "zjw1111 (via GitHub)" <gi...@apache.org> on 2026/10/19 16:27:02 UTC
[PR] feat(file-index): support V2 containers and streaming writes [paimon-cpp]
zjw1111 opened a new pull request, #446:
URL: https://github.com/apache/paimon-cpp/pull/446
### Purpose
Closes #445.
V1 uses 32-bit payload offsets and lengths, and the data-file index write path buffers the complete container before deciding whether to embed it or create an external file. Add Java-compatible V2 containers with 64-bit payload ranges and streamed payload writes, while keeping V1 as the default.
Split the format implementations into V1/V2 writers, add `FileIndexWriter::WriteTo()` with a `SerializedBytes()` fallback, and implement direct streaming for Lumina. The data-file index writer now spills to an external file when the container crosses the inline threshold.
### Tests
- Final submission checks: `pre-commit run --files` for all 40 changed files, `git diff --cached --check`, and `git diff --check` passed.
- Earlier focused validation passed for format reading/writing, Java V2 golden bytes, large ranges, plugin limits, data-file index writing, spill thresholds and cleanup, and Lumina round trips. The latest Lumina validation passed all four tests.
- The golden bytes were independently generated by running the unmodified Java Paimon format sources at commit `5a6739c6c10e96052bac678eb245742ad92bc556`; all 86 bytes match the C++ fixture.
- No local compilation or test rerun was performed for this final submission, as requested by the author.
### API and Format
- Add `file-index.format-version=1|2` (default: `1`), with automatic reader dispatch based on the stored version.
- V2 stores payloads before a footer and trailer, with int64 offsets and lengths. V1 layout remains compatible.
- Widen `FileIndexer::CreateReader()` offsets and lengths to int64. Individual plugins retain their own payload limits; bitmap and range bitmap explicitly reject lengths above `INT32_MAX`.
- Add `Result<bool> FileIndexWriter::WriteTo(const std::shared_ptr<OutputStream>&) const`. False means no payload; true means a non-empty payload was written. Null or zero-length `SerializedBytes()` results produce no index.
- Replace container `WriteColumnIndexes()` with `WriteIndex()` and `Finish()`, and add a version argument to `CreateWriter()` with a V1 default. Downstream API implementations and callers must adapt to these interface changes.
### Documentation
Document version selection and streaming behavior in the public format API comments. Describe each serialized layout in `file_index_format_v1.h` and `file_index_format_v2.h`.
### Generative AI tooling
Generated-by: OpenAI Codex (GPT-6)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(file-index): support V2 containers and streaming writes [paimon-cpp]
Posted by "lxy-9602 (via GitHub)" <gi...@apache.org>.
lxy-9602 commented on code in PR #446:
URL: https://github.com/apache/paimon-cpp/pull/446#discussion_r4332697029
##########
src/paimon/common/file_index/file_index_format_test.cpp:
##########
@@ -17,31 +17,60 @@
*/
#include "paimon/file_index/file_index_format.h"
+#include <algorithm>
+#include <array>
+#include <cstring>
+#include <limits>
+#include <map>
#include <utility>
+#include "arrow/api.h"
#include "gtest/gtest.h"
#include "paimon/common/file_index/bitmap/bitmap_file_index.h"
#include "paimon/common/file_index/bloomfilter/bloom_filter_file_index.h"
#include "paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h"
#include "paimon/common/file_index/empty/empty_file_index_reader.h"
+#include "paimon/common/file_index/file_index_format_utils.h"
+#include "paimon/common/file_index/rangebitmap/range_bitmap_file_index.h"
#include "paimon/common/io/byte_array_output_stream.h"
+#include "paimon/common/io/data_output_stream.h"
#include "paimon/data/timestamp.h"
#include "paimon/defs.h"
#include "paimon/file_index/file_index_result.h"
#include "paimon/file_index/scored_file_index_result.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/io/byte_array_input_stream.h"
+#include "paimon/io/data_input_stream.h"
#include "paimon/memory/bytes.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/predicate/literal.h"
#include "paimon/predicate/vector_search.h"
#include "paimon/status.h"
+#include "paimon/testing/mock/mock_file_system.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
Review Comment:
add inte tests
##########
src/paimon/common/file_index/file_index_format_test.cpp:
##########
@@ -17,31 +17,60 @@
*/
#include "paimon/file_index/file_index_format.h"
+#include <algorithm>
+#include <array>
+#include <cstring>
+#include <limits>
+#include <map>
#include <utility>
+#include "arrow/api.h"
#include "gtest/gtest.h"
#include "paimon/common/file_index/bitmap/bitmap_file_index.h"
#include "paimon/common/file_index/bloomfilter/bloom_filter_file_index.h"
#include "paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h"
#include "paimon/common/file_index/empty/empty_file_index_reader.h"
+#include "paimon/common/file_index/file_index_format_utils.h"
+#include "paimon/common/file_index/rangebitmap/range_bitmap_file_index.h"
#include "paimon/common/io/byte_array_output_stream.h"
+#include "paimon/common/io/data_output_stream.h"
#include "paimon/data/timestamp.h"
#include "paimon/defs.h"
#include "paimon/file_index/file_index_result.h"
#include "paimon/file_index/scored_file_index_result.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/io/byte_array_input_stream.h"
+#include "paimon/io/data_input_stream.h"
#include "paimon/memory/bytes.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/predicate/literal.h"
#include "paimon/predicate/vector_search.h"
#include "paimon/status.h"
+#include "paimon/testing/mock/mock_file_system.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
class FileIndexFormatTest : public ::testing::Test {
public:
+ Status WriteIndexes(FileIndexFormat::Writer* writer,
+ const FileIndexFormat::ColumnIndexes& indexes) const {
+ for (const auto& [column, entries] : indexes) {
Review Comment:
Should the output parameters be placed at the end?
##########
src/paimon/common/file_index/file_index_format_test.cpp:
##########
@@ -865,4 +894,302 @@ TEST_F(FileIndexFormatTest, TestBitmapIndexWithTimestamp) {
check_nano("ts_tz_nano");
}
+namespace {
+
+std::shared_ptr<ByteArrayOutputStream> NewIndexOutput(const std::shared_ptr<MemoryPool>& pool) {
+ return std::make_shared<ByteArrayOutputStream>(std::make_unique<MemorySegmentOutputStream>(
+ MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool));
+}
+
+// Generated with FileIndexFormatV2.Writer from Java Paimon.
+// Contains an empty index, a two-byte unknown index, and a present zero-byte index.
+const std::vector<uint8_t> kJavaV2Golden = {
+ 0, 5, 78, 78, 208, 26, 53, 174, 0, 0, 0, 2, 42, 43, 0, 0, 0, 1,
+ 0, 2, 99, 49, 0, 0, 0, 2, 0, 5, 101, 109, 112, 116, 121, 255, 255, 255,
+ 255, 255, 255, 255, 255, 0, 0, 0, 0, 0, 0, 0, 0, 0, 7, 117, 110, 107,
+ 110, 111, 119, 110, 0, 0, 0, 0, 0, 0, 0, 12, 0, 0, 0, 0, 0, 0,
+ 0, 2, 0, 0, 0, 60, 80, 65, 70, 73, 68, 88, 48, 50,
+};
+
+struct SparseIndexState {
+ std::map<int64_t, std::string> segments;
+ int64_t length = 0;
+ bool discard = false;
+};
+
+class SparseIndexOutput : public MockOutputStream {
+ public:
+ explicit SparseIndexOutput(const std::shared_ptr<SparseIndexState>& state) : state_(state) {}
+ Result<int64_t> Write(const char* buffer, int64_t size) override {
+ if (!state_->discard) {
+ state_->segments[state_->length] = std::string(buffer, size);
+ }
+ state_->length += size;
+ return size;
+ }
+ Result<int64_t> GetPos() const override {
+ return state_->length;
+ }
+
+ private:
+ std::shared_ptr<SparseIndexState> state_;
+};
+
+class SparseIndexInput : public MockInputStream {
+ public:
+ explicit SparseIndexInput(const std::shared_ptr<SparseIndexState>& state) : state_(state) {}
+ Status Seek(int64_t offset, SeekOrigin origin) override {
+ if (origin == FS_SEEK_CUR) {
+ offset += position_;
+ }
+ if (origin == FS_SEEK_END) {
+ offset += state_->length;
+ }
+ if (offset < 0 || offset > state_->length) {
+ return Status::Invalid("Invalid seek");
+ }
+ position_ = offset;
+ return Status::OK();
+ }
+ Result<int64_t> GetPos() const override {
+ return position_;
+ }
+ Result<int64_t> Length() const override {
+ return state_->length;
+ }
+ Result<int64_t> Read(char* buffer, int64_t size) override {
+ PAIMON_ASSIGN_OR_RAISE(int64_t read, Read(buffer, size, position_));
+ position_ += read;
+ return read;
+ }
+ Result<int64_t> Read(char* buffer, int64_t size, int64_t offset) override {
+ if (offset < 0 || size < 0 || offset > state_->length) {
+ return Status::Invalid("Invalid read");
+ }
+ size = std::min(size, state_->length - offset);
+ std::memset(buffer, 0, size);
+ for (const auto& [start, bytes] : state_->segments) {
+ int64_t end = start + bytes.size();
+ int64_t from = std::max(start, offset);
+ int64_t to = std::min(end, offset + size);
+ if (from < to) {
+ std::memcpy(buffer + from - offset, bytes.data() + from - start, to - from);
+ }
+ }
+ return size;
+ }
+
+ private:
+ std::shared_ptr<SparseIndexState> state_;
+ int64_t position_ = 0;
+};
+} // namespace
+
+TEST_F(FileIndexFormatTest, TestV2JavaGoldenBytes) {
+ FileIndexFormat::ColumnIndexes indexes;
+ indexes["c1"]["empty"] = nullptr;
+ indexes["c1"]["unknown"] = std::make_shared<Bytes>(std::string("*+"), pool_.get());
+ auto output = NewIndexOutput(pool_);
+ ASSERT_OK_AND_ASSIGN(auto writer,
+ FileIndexFormat::CreateWriter(output, pool_, FileIndexFormat::V_2));
+ ASSERT_OK(WriteIndexes(writer.get(), indexes));
+ ASSERT_OK(writer->Close());
+ ASSERT_OK_AND_ASSIGN(auto bytes, output->Finish(pool_.get()));
+ ASSERT_EQ(kJavaV2Golden, std::vector<uint8_t>(bytes->data(), bytes->data() + bytes->size()));
+
+ // Read the Java golden bytes, including empty-index handling and
+ // skipping unknown backend types without opening their payloads.
+ auto input = std::make_shared<ByteArrayInputStream>(
+ reinterpret_cast<const char*>(kJavaV2Golden.data()), kJavaV2Golden.size());
+ ASSERT_OK_AND_ASSIGN(auto reader, FileIndexFormat::CreateReader(input, pool_, /*options=*/{}));
+ auto schema = arrow::schema({arrow::field("c1", arrow::int32())});
+ ASSERT_OK_AND_ASSIGN(auto readers,
+ reader->ReadColumnIndex("c1", CreateArrowSchema(schema).get()));
+ ASSERT_EQ(1, readers.size());
+ ASSERT_TRUE(dynamic_cast<EmptyFileIndexReader*>(readers[0].get()));
+}
+
+TEST_F(FileIndexFormatTest, TestV2InvalidContainers) {
+ auto check = [this](const std::vector<uint8_t>& bytes) {
+ auto input = std::make_shared<ByteArrayInputStream>(
+ bytes.empty() ? "" : reinterpret_cast<const char*>(bytes.data()), bytes.size());
+ ASSERT_NOK(FileIndexFormat::CreateReader(input, pool_, /*options=*/{}));
+ };
+ for (size_t size : {static_cast<size_t>(0), static_cast<size_t>(11), static_cast<size_t>(24),
+ kJavaV2Golden.size() - 1}) {
+ check(std::vector<uint8_t>(kJavaV2Golden.begin(), kJavaV2Golden.begin() + size));
+ }
+ for (const auto& [offset, value] : std::vector<std::pair<size_t, uint8_t>>{
+ {11, 3},
+ {kJavaV2Golden.size() - 1, 0},
+ {kJavaV2Golden.size() - 12, 0x7f}, // Impossible footer length.
+ {kJavaV2Golden.size() - 12, 0xff}}) { // Negative footer length.
+ auto bytes = kJavaV2Golden;
+ bytes[offset] = value;
+ check(bytes);
+ }
+}
+
+TEST_F(FileIndexFormatTest, TestV2StreamingLargeContainer) {
+ auto state = std::make_shared<SparseIndexState>();
+ auto output = std::make_shared<SparseIndexOutput>(state);
+ ASSERT_OK_AND_ASSIGN(auto writer,
+ FileIndexFormat::CreateWriter(output, pool_, FileIndexFormat::V_2));
+ // Stream a virtual >2 GiB payload in bounded chunks; the mock discards only this gap.
+ const int64_t gap_length = int64_t{1} << 31;
+ ASSERT_OK(writer->WriteIndex(
+ "gap", "unknown", [state](const std::shared_ptr<OutputStream>& stream) -> Result<bool> {
+ state->discard = true;
+ std::array<char, 65536> zeros{};
+ for (int64_t i = 0; i < gap_length; i += zeros.size()) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t written, stream->Write(zeros.data(), zeros.size()));
+ if (written != static_cast<int64_t>(zeros.size())) {
+ return Status::IOError("short write");
+ }
+ }
+ state->discard = false;
+ return true;
+ }));
+ auto schema = arrow::schema({arrow::field("f0", arrow::int32())});
+ std::vector<std::pair<std::string, std::shared_ptr<FileIndexer>>> indexers = {
+ {"bitmap", std::make_shared<BitmapFileIndex>(std::map<std::string, std::string>{})},
+ {"bsi",
+ std::make_shared<BitSliceIndexBitmapFileIndex>(std::map<std::string, std::string>{})},
+ {"bloom-filter",
+ std::make_shared<BloomFilterFileIndex>(std::map<std::string, std::string>{})},
+ {"range-bitmap",
+ std::make_shared<RangeBitmapFileIndex>(std::map<std::string, std::string>{})}};
+ arrow::Int32Builder builder;
+ ASSERT_TRUE(builder.AppendValues({1, 2, 1}).ok());
+ auto values_result = builder.Finish();
+ ASSERT_TRUE(values_result.ok());
+ std::shared_ptr<arrow::Array> values = values_result.ValueOrDie();
+ auto batch_result = arrow::StructArray::Make({values}, schema->fields());
+ ASSERT_TRUE(batch_result.ok());
+ auto batch = batch_result.ValueOrDie();
Review Comment:
use json?
##########
src/paimon/common/file_index/file_index_format_test.cpp:
##########
@@ -865,4 +894,302 @@ TEST_F(FileIndexFormatTest, TestBitmapIndexWithTimestamp) {
check_nano("ts_tz_nano");
}
+namespace {
+
+std::shared_ptr<ByteArrayOutputStream> NewIndexOutput(const std::shared_ptr<MemoryPool>& pool) {
+ return std::make_shared<ByteArrayOutputStream>(std::make_unique<MemorySegmentOutputStream>(
+ MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool));
+}
+
+// Generated with FileIndexFormatV2.Writer from Java Paimon.
+// Contains an empty index, a two-byte unknown index, and a present zero-byte index.
+const std::vector<uint8_t> kJavaV2Golden = {
+ 0, 5, 78, 78, 208, 26, 53, 174, 0, 0, 0, 2, 42, 43, 0, 0, 0, 1,
+ 0, 2, 99, 49, 0, 0, 0, 2, 0, 5, 101, 109, 112, 116, 121, 255, 255, 255,
+ 255, 255, 255, 255, 255, 0, 0, 0, 0, 0, 0, 0, 0, 0, 7, 117, 110, 107,
+ 110, 111, 119, 110, 0, 0, 0, 0, 0, 0, 0, 12, 0, 0, 0, 0, 0, 0,
+ 0, 2, 0, 0, 0, 60, 80, 65, 70, 73, 68, 88, 48, 50,
+};
+
+struct SparseIndexState {
+ std::map<int64_t, std::string> segments;
+ int64_t length = 0;
+ bool discard = false;
+};
+
+class SparseIndexOutput : public MockOutputStream {
+ public:
+ explicit SparseIndexOutput(const std::shared_ptr<SparseIndexState>& state) : state_(state) {}
+ Result<int64_t> Write(const char* buffer, int64_t size) override {
+ if (!state_->discard) {
+ state_->segments[state_->length] = std::string(buffer, size);
+ }
+ state_->length += size;
+ return size;
+ }
+ Result<int64_t> GetPos() const override {
+ return state_->length;
+ }
+
+ private:
+ std::shared_ptr<SparseIndexState> state_;
+};
+
+class SparseIndexInput : public MockInputStream {
+ public:
+ explicit SparseIndexInput(const std::shared_ptr<SparseIndexState>& state) : state_(state) {}
+ Status Seek(int64_t offset, SeekOrigin origin) override {
+ if (origin == FS_SEEK_CUR) {
+ offset += position_;
+ }
+ if (origin == FS_SEEK_END) {
+ offset += state_->length;
+ }
+ if (offset < 0 || offset > state_->length) {
+ return Status::Invalid("Invalid seek");
+ }
+ position_ = offset;
+ return Status::OK();
+ }
+ Result<int64_t> GetPos() const override {
+ return position_;
+ }
+ Result<int64_t> Length() const override {
+ return state_->length;
+ }
+ Result<int64_t> Read(char* buffer, int64_t size) override {
+ PAIMON_ASSIGN_OR_RAISE(int64_t read, Read(buffer, size, position_));
+ position_ += read;
+ return read;
+ }
+ Result<int64_t> Read(char* buffer, int64_t size, int64_t offset) override {
+ if (offset < 0 || size < 0 || offset > state_->length) {
+ return Status::Invalid("Invalid read");
+ }
+ size = std::min(size, state_->length - offset);
+ std::memset(buffer, 0, size);
+ for (const auto& [start, bytes] : state_->segments) {
+ int64_t end = start + bytes.size();
+ int64_t from = std::max(start, offset);
+ int64_t to = std::min(end, offset + size);
+ if (from < to) {
+ std::memcpy(buffer + from - offset, bytes.data() + from - start, to - from);
+ }
+ }
+ return size;
+ }
+
+ private:
+ std::shared_ptr<SparseIndexState> state_;
+ int64_t position_ = 0;
+};
+} // namespace
+
+TEST_F(FileIndexFormatTest, TestV2JavaGoldenBytes) {
+ FileIndexFormat::ColumnIndexes indexes;
+ indexes["c1"]["empty"] = nullptr;
+ indexes["c1"]["unknown"] = std::make_shared<Bytes>(std::string("*+"), pool_.get());
+ auto output = NewIndexOutput(pool_);
+ ASSERT_OK_AND_ASSIGN(auto writer,
+ FileIndexFormat::CreateWriter(output, pool_, FileIndexFormat::V_2));
+ ASSERT_OK(WriteIndexes(writer.get(), indexes));
+ ASSERT_OK(writer->Close());
+ ASSERT_OK_AND_ASSIGN(auto bytes, output->Finish(pool_.get()));
+ ASSERT_EQ(kJavaV2Golden, std::vector<uint8_t>(bytes->data(), bytes->data() + bytes->size()));
+
+ // Read the Java golden bytes, including empty-index handling and
+ // skipping unknown backend types without opening their payloads.
+ auto input = std::make_shared<ByteArrayInputStream>(
+ reinterpret_cast<const char*>(kJavaV2Golden.data()), kJavaV2Golden.size());
+ ASSERT_OK_AND_ASSIGN(auto reader, FileIndexFormat::CreateReader(input, pool_, /*options=*/{}));
+ auto schema = arrow::schema({arrow::field("c1", arrow::int32())});
+ ASSERT_OK_AND_ASSIGN(auto readers,
+ reader->ReadColumnIndex("c1", CreateArrowSchema(schema).get()));
+ ASSERT_EQ(1, readers.size());
+ ASSERT_TRUE(dynamic_cast<EmptyFileIndexReader*>(readers[0].get()));
+}
+
+TEST_F(FileIndexFormatTest, TestV2InvalidContainers) {
+ auto check = [this](const std::vector<uint8_t>& bytes) {
+ auto input = std::make_shared<ByteArrayInputStream>(
+ bytes.empty() ? "" : reinterpret_cast<const char*>(bytes.data()), bytes.size());
+ ASSERT_NOK(FileIndexFormat::CreateReader(input, pool_, /*options=*/{}));
+ };
+ for (size_t size : {static_cast<size_t>(0), static_cast<size_t>(11), static_cast<size_t>(24),
+ kJavaV2Golden.size() - 1}) {
+ check(std::vector<uint8_t>(kJavaV2Golden.begin(), kJavaV2Golden.begin() + size));
+ }
+ for (const auto& [offset, value] : std::vector<std::pair<size_t, uint8_t>>{
+ {11, 3},
+ {kJavaV2Golden.size() - 1, 0},
+ {kJavaV2Golden.size() - 12, 0x7f}, // Impossible footer length.
+ {kJavaV2Golden.size() - 12, 0xff}}) { // Negative footer length.
+ auto bytes = kJavaV2Golden;
+ bytes[offset] = value;
+ check(bytes);
+ }
+}
+
+TEST_F(FileIndexFormatTest, TestV2StreamingLargeContainer) {
+ auto state = std::make_shared<SparseIndexState>();
+ auto output = std::make_shared<SparseIndexOutput>(state);
+ ASSERT_OK_AND_ASSIGN(auto writer,
+ FileIndexFormat::CreateWriter(output, pool_, FileIndexFormat::V_2));
+ // Stream a virtual >2 GiB payload in bounded chunks; the mock discards only this gap.
+ const int64_t gap_length = int64_t{1} << 31;
+ ASSERT_OK(writer->WriteIndex(
+ "gap", "unknown", [state](const std::shared_ptr<OutputStream>& stream) -> Result<bool> {
+ state->discard = true;
+ std::array<char, 65536> zeros{};
+ for (int64_t i = 0; i < gap_length; i += zeros.size()) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t written, stream->Write(zeros.data(), zeros.size()));
+ if (written != static_cast<int64_t>(zeros.size())) {
+ return Status::IOError("short write");
+ }
+ }
+ state->discard = false;
+ return true;
+ }));
+ auto schema = arrow::schema({arrow::field("f0", arrow::int32())});
+ std::vector<std::pair<std::string, std::shared_ptr<FileIndexer>>> indexers = {
+ {"bitmap", std::make_shared<BitmapFileIndex>(std::map<std::string, std::string>{})},
+ {"bsi",
+ std::make_shared<BitSliceIndexBitmapFileIndex>(std::map<std::string, std::string>{})},
+ {"bloom-filter",
+ std::make_shared<BloomFilterFileIndex>(std::map<std::string, std::string>{})},
+ {"range-bitmap",
+ std::make_shared<RangeBitmapFileIndex>(std::map<std::string, std::string>{})}};
+ arrow::Int32Builder builder;
+ ASSERT_TRUE(builder.AppendValues({1, 2, 1}).ok());
+ auto values_result = builder.Finish();
+ ASSERT_TRUE(values_result.ok());
+ std::shared_ptr<arrow::Array> values = values_result.ValueOrDie();
+ auto batch_result = arrow::StructArray::Make({values}, schema->fields());
+ ASSERT_TRUE(batch_result.ok());
+ auto batch = batch_result.ValueOrDie();
+ for (const auto& [type, indexer] : indexers) {
+ ASSERT_OK_AND_ASSIGN(auto index_writer,
+ indexer->CreateWriter(CreateArrowSchema(schema).get(), pool_));
+ ArrowArray c_batch;
+ ASSERT_TRUE(arrow::ExportArray(*batch, &c_batch).ok());
+ ASSERT_OK(index_writer->AddBatch(&c_batch));
+ ASSERT_OK_AND_ASSIGN(auto payload, index_writer->SerializedBytes());
+ ASSERT_OK(writer->WriteIndex(
+ type, type, [&payload](const std::shared_ptr<OutputStream>& stream) -> Result<bool> {
+ PAIMON_ASSIGN_OR_RAISE(int64_t written,
+ stream->Write(payload->data(), payload->size()));
+ if (written != static_cast<int64_t>(payload->size())) {
+ return Status::IOError("short write");
+ }
+ return true;
+ }));
+ }
+ ASSERT_OK(writer->Finish());
+ ASSERT_OK(writer->Close());
+ ASSERT_GT(state->length, gap_length);
+ auto input = std::make_shared<SparseIndexInput>(state);
+ ASSERT_OK_AND_ASSIGN(auto reader, FileIndexFormat::CreateReader(input, pool_, /*options=*/{}));
+ for (const auto& [type, indexer] : indexers) {
+ ASSERT_OK_AND_ASSIGN(
+ auto readers,
+ reader->ReadColumnIndex(
+ type,
+ CreateArrowSchema(arrow::schema({arrow::field(type, arrow::int32())})).get()));
+ ASSERT_EQ(1, readers.size()) << type;
+ ASSERT_OK_AND_ASSIGN(auto matches, readers[0]->VisitEqual(Literal(1)));
+ if (type == "bloom-filter") {
+ ASSERT_OK_AND_ASSIGN(bool remains, matches->IsRemain());
+ ASSERT_TRUE(remains);
+ } else {
+ ASSERT_EQ("{0,2}", matches->ToString()) << type;
+ }
+ }
+}
+
+TEST_F(FileIndexFormatTest, TestPluginRangeLimits) {
+ auto schema = arrow::schema({arrow::field("f0", arrow::int32())});
+ auto state = std::make_shared<SparseIndexState>();
+ state->length = int64_t{1} << 32;
+ auto input = std::make_shared<SparseIndexInput>(state);
+ BitmapFileIndex bitmap({});
+ RangeBitmapFileIndex range({});
+ const int64_t oversized = int64_t{1} << 31;
+ for (FileIndexer* indexer : std::vector<FileIndexer*>{&bitmap, &range}) {
+ auto c_schema = CreateArrowSchema(schema);
+ ASSERT_NOK_WITH_MSG(indexer->CreateReader(c_schema.get(), 0, oversized, input, pool_),
+ "INT32_MAX");
+ if (c_schema->release) {
+ c_schema->release(c_schema.get());
+ }
+ }
+}
+
+TEST_F(FileIndexFormatTest, TestV2EmptyContainer) {
+ auto output = NewIndexOutput(pool_);
+ ASSERT_OK_AND_ASSIGN(auto writer,
+ FileIndexFormat::CreateWriter(output, pool_, FileIndexFormat::V_2));
+ ASSERT_OK_AND_ASSIGN(int64_t prefix_length, output->GetPos());
+ ASSERT_EQ(12, prefix_length);
+ ASSERT_OK(writer->Finish());
+ ASSERT_OK(writer->Close());
+ ASSERT_OK_AND_ASSIGN(auto bytes, output->Finish(pool_.get()));
+ ASSERT_EQ(28, bytes->size());
+ auto input = std::make_shared<ByteArrayInputStream>(bytes->data(), bytes->size());
+ ASSERT_OK_AND_ASSIGN(auto reader, FileIndexFormat::CreateReader(input, pool_, /*options=*/{}));
+}
+
+TEST_F(FileIndexFormatTest, TestV1BuffersPayloadsUntilFinish) {
+ FileIndexFormat::ColumnIndexes indexes = {
+ {"f0", {{"bitmap", std::make_shared<Bytes>("payload", pool_.get())}, {"empty", nullptr}}}};
+ auto output = NewIndexOutput(pool_);
+ ASSERT_OK_AND_ASSIGN(auto writer, FileIndexFormat::CreateWriter(output, pool_));
+ for (const auto& [column, entries] : indexes) {
+ for (const auto& [type, bytes] : entries) {
+ FileIndexFormat::Writer::PayloadWriter payload;
+ if (bytes) {
+ payload = [bytes = bytes](
+ const std::shared_ptr<OutputStream>& destination) -> Result<bool> {
+ PAIMON_RETURN_NOT_OK(DataOutputStream(destination).WriteBytes(bytes));
+ return true;
+ };
+ }
+ ASSERT_OK(writer->WriteIndex(column, type, payload));
+ }
+ }
+ ASSERT_OK_AND_ASSIGN(int64_t before_finish, output->GetPos());
Review Comment:
review and refactor the test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(file-index): support V2 containers and streaming writes [paimon-cpp]
Posted by "lxy-9602 (via GitHub)" <gi...@apache.org>.
lxy-9602 merged PR #446:
URL: https://github.com/apache/paimon-cpp/pull/446
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(file-index): support V2 containers and streaming writes [paimon-cpp]
Posted by "lxy-9602 (via GitHub)" <gi...@apache.org>.
lxy-9602 commented on code in PR #446:
URL: https://github.com/apache/paimon-cpp/pull/446#discussion_r4323711145
##########
src/paimon/common/file_index/bitmap/bitmap_file_index.cpp:
##########
@@ -90,8 +98,8 @@ Result<std::shared_ptr<FileIndexReader>> BitmapFileIndex::CreateReader(
auto arrow_type = arrow_schema->field(0)->type();
PAIMON_ASSIGN_OR_RAISE(FieldType data_type,
FieldTypeUtils::ConvertToFieldType(arrow_type->id()));
- return std::make_shared<BitmapFileIndexReader>(arrow_type, data_type, start, length,
- input_stream, pool);
+ return std::make_shared<BitmapFileIndexReader>(
+ arrow_type, data_type, 0, static_cast<int32_t>(length), bounded_stream, pool);
Review Comment:
Why do we still need to pass `start` and `length` when using `bounded_stream` here?
##########
src/paimon/common/file_index/file_index_format_utils.h:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 <unordered_map>
+#include <utility>
+
+#include "paimon/result.h"
+
+namespace paimon {
+class DataInputStream;
+class InputStream;
+class MemoryPool;
+
+using FileIndexEntries =
+ std::unordered_map<std::string, std::unordered_map<std::string, std::pair<int64_t, int64_t>>>;
+
+class FileIndexFormatUtils {
+ public:
+ FileIndexFormatUtils() = delete;
+ ~FileIndexFormatUtils() = delete;
+
+ /// Creates a bounded, buffered view of the container directory.
+ static Result<std::shared_ptr<InputStream>> CreateFileIndexDirectory(
+ const std::shared_ptr<InputStream>& input, const std::shared_ptr<MemoryPool>& pool,
+ int64_t start, int64_t length, int64_t file_length);
Review Comment:
Please put `pool` as the last parameter.
##########
src/paimon/common/file_index/file_index_format_utils.h:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 <unordered_map>
+#include <utility>
+
+#include "paimon/result.h"
+
+namespace paimon {
+class DataInputStream;
+class InputStream;
+class MemoryPool;
+
+using FileIndexEntries =
+ std::unordered_map<std::string, std::unordered_map<std::string, std::pair<int64_t, int64_t>>>;
+
Review Comment:
mv `FileIndexEntries` to `FileIndexFormatUtils`
##########
include/paimon/file_index/file_index_format.h:
##########
@@ -109,35 +63,47 @@ class PAIMON_EXPORT FileIndexFormat {
const std::shared_ptr<InputStream>& input_stream, const std::shared_ptr<MemoryPool>& pool,
const std::map<std::string, std::string>& options);
- /// Creates a `Writer` which serializes a complete V1 file index container.
+ /// Creates a `Writer` for file index container.
///
/// @param output_stream Destination stream for serialized index data.
/// @param pool Memory pool for writer-side allocations.
+ /// @param version Container format version; defaults to V1.
/// @return A unique pointer to a `Writer` on success.
static Result<std::unique_ptr<Writer>> CreateWriter(
- const std::shared_ptr<OutputStream>& output_stream,
- const std::shared_ptr<MemoryPool>& pool);
+ const std::shared_ptr<OutputStream>& output_stream, const std::shared_ptr<MemoryPool>& pool,
+ int32_t version = V_1);
Review Comment:
Please remove the default parameters, and move the `version` parameter before the pool.
##########
src/paimon/common/file_index/file_index_format_utils.cpp:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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/common/file_index/file_index_format_utils.h"
+
+#include "paimon/common/io/offset_input_stream.h"
+#include "paimon/file_index/file_index_format.h"
+#include "paimon/io/buffered_input_stream.h"
+#include "paimon/io/data_input_stream.h"
+
+namespace paimon {
+Result<std::shared_ptr<InputStream>> FileIndexFormatUtils::CreateFileIndexDirectory(
+ const std::shared_ptr<InputStream>& input, const std::shared_ptr<MemoryPool>& pool,
+ int64_t start, int64_t length, int64_t file_length) {
Review Comment:
rename Directory
##########
src/paimon/core/io/spillable_index_output_stream.cpp:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/io/spillable_index_output_stream.h"
+
+#include <limits>
+
+#include "paimon/common/io/byte_array_output_stream.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+
+namespace paimon {
+SpillableIndexOutputStream::SpillableIndexOutputStream(
+ const std::shared_ptr<FileSystem>& file_system, const std::string& path, int64_t threshold,
+ const std::shared_ptr<MemoryPool>& pool)
+ : threshold_(threshold),
+ pool_(pool),
+ file_system_(file_system),
+ path_(path),
+ memory_(std::make_shared<ByteArrayOutputStream>(std::make_unique<MemorySegmentOutputStream>(
+ MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool))) {}
+
+Result<int64_t> SpillableIndexOutputStream::Write(const char* buffer, int64_t size) {
+ if (size < 0 || size > std::numeric_limits<int64_t>::max() - position_) {
+ return Status::Invalid("File index size overflow");
+ }
+ if (!external_ && position_ + size > threshold_) {
+ PAIMON_ASSIGN_OR_RAISE(external_, file_system_->Create(path_, /*overwrite=*/false));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> prefix, memory_->Finish(pool_.get()));
+ PAIMON_ASSIGN_OR_RAISE(int64_t written, external_->Write(prefix->data(), prefix->size()));
+ if (written != static_cast<int64_t>(prefix->size())) {
+ return Status::IOError("Short write for file index");
+ }
+ memory_.reset();
+ }
+ PAIMON_ASSIGN_OR_RAISE(int64_t written,
+ (external_ ? external_.get() : memory_.get())->Write(buffer, size));
+ if (written != size) {
+ return Status::IOError("Short write for file index");
+ }
+ position_ += written;
+ return written;
+}
+
+Result<int64_t> SpillableIndexOutputStream::GetPos() const {
+ return position_;
+}
+
+Result<std::string> SpillableIndexOutputStream::GetUri() const {
+ return path_;
+}
+
+Status SpillableIndexOutputStream::Flush() {
+ return external_ ? external_->Flush() : memory_->Flush();
+}
+
+Status SpillableIndexOutputStream::Close() {
+ if (closed_) {
+ return Status::OK();
+ }
+ closed_ = true;
+ return external_ ? external_->Close() : memory_->Close();
+}
+
+Result<std::shared_ptr<Bytes>> SpillableIndexOutputStream::EmbeddedBytes() {
+ if (external_) {
+ return std::shared_ptr<Bytes>();
Review Comment:
Should this return `nullptr`, or should it throw an error?
##########
src/paimon/core/io/data_file_index_writer.cpp:
##########
@@ -149,62 +151,36 @@ Status DataFileIndexWriter::AddBatch(const std::shared_ptr<arrow::StructArray>&
return Status::OK();
}
-Result<std::shared_ptr<Bytes>> DataFileIndexWriter::SerializeContainer() {
- FileIndexFormat::ColumnIndexes column_indexes;
- for (const IndexWriterEntry& entry : writers_) {
- PAIMON_ASSIGN_OR_RAISE(column_indexes[entry.column_name][entry.index_type],
- entry.writer->SerializedBytes());
- }
-
- auto segment_output = std::make_unique<MemorySegmentOutputStream>(
- MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool_);
- std::shared_ptr<ByteArrayOutputStream> output =
- std::make_shared<ByteArrayOutputStream>(std::move(segment_output));
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileIndexFormat::Writer> format_writer,
- FileIndexFormat::CreateWriter(output, pool_));
- PAIMON_RETURN_NOT_OK(format_writer->WriteColumnIndexes(column_indexes));
- PAIMON_RETURN_NOT_OK(format_writer->Close());
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes, output->Finish(pool_.get()));
- return bytes;
-}
-
Result<FileIndexWriteResult> DataFileIndexWriter::Finish(const std::string& data_file_path) {
if (finished_) {
return Status::Invalid("Data file index writer has already finished");
}
finished_ = true;
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes, SerializeContainer());
- if (static_cast<int64_t>(bytes->size()) <= in_manifest_threshold_) {
- return FileIndexWriteResult{bytes, {}};
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(in_manifest_threshold_, "file index inline threshold"));
Review Comment:
Why is the validation for `in_manifest_threshold_` placed so late?
##########
src/paimon/indexer/lumina/lumina_file_index.cpp:
##########
@@ -39,6 +39,42 @@
namespace paimon::lumina {
namespace {
+/// Reports positions relative to the wrapped stream's position at creation.
+/// Flush forwards to the wrapped stream; Close leaves it open so the container can continue
+/// writing.
+class OffsetOutputStream : public OutputStream {
+ public:
Review Comment:
Should this go into `lumina_file_writer.h` or into `utils`?
##########
src/paimon/common/file_index/file_index_format_v2.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 <map>
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "paimon/common/file_index/file_index_format_utils.h"
+#include "paimon/file_index/file_index_format.h"
+
+namespace paimon {
+/// V2 container layout and writer implementation.
+/// - All numeric fields use big-endian byte order.
+/// - Column metadata follows the index payloads.
+/// - Payload offsets and lengths are int64; an empty index uses (-1, 0).
+/// - Incremental writes stream payloads directly; Finish() writes the footer and trailer.
+/// - The trailer locates the footer, footer_length must fit within INT32_MAX bytes.
+///
+/// <pre>
+/// _______________________________________ _____________________
+/// | magic | version | PREFIX
+/// |--------------------------------------| ---------------------
+/// | index payload 1 |
+/// |--------------------------------------|
+/// | index payload 2 | BODY
+/// |--------------------------------------|
+/// | ... |
+/// |--------------------------------------| ---------------------
+/// | column number |
+/// |--------------------------------------|
+/// | column 1 | index number |
+/// |--------------------------------------|
+/// | index name 1 | start pos | length |
+/// |--------------------------------------|
+/// | index name 2 | start pos | length |
+/// |--------------------------------------|
+/// | ... | FOOTER
+/// |--------------------------------------|
+/// | column 2 | index number |
+/// |--------------------------------------|
+/// | index name 1 | start pos | length |
+/// |--------------------------------------|
+/// | ... |
+/// |--------------------------------------| ---------------------
+/// | footer length | tail magic | TRAILER
+/// |______________________________________| _____________________
+///
+/// magic: 8 bytes long
+/// version: 4 bytes int (2)
+/// column number: 4 bytes int
+/// column x: var bytes utf (length + bytes)
+/// index number: 4 bytes int (number of indexes for the column)
+/// index name x: var bytes utf (index type)
+/// start pos: 8 bytes long (absolute payload offset)
+/// length: 8 bytes long (payload length)
+/// footer length: 4 bytes int (footer bytes only, excludes trailer)
+/// tail magic: 8 bytes long (0x5041464944583032, "PAFIDX02")
+/// BODY: serialized index payloads
+/// </pre>
+///
+class FileIndexFormatV2 {
+ public:
+ FileIndexFormatV2() = delete;
+ ~FileIndexFormatV2() = delete;
+
+ class Writer;
+
+ /// Reads the directory after the magic/version prefix has been consumed.
+ static Result<FileIndexEntries> ReadIndexEntries(const std::shared_ptr<InputStream>& input,
+ const std::shared_ptr<MemoryPool>& pool,
+ int64_t file_length);
+};
+
+class FileIndexFormatV2::Writer : public FileIndexFormat::Writer {
+ public:
+ static Result<std::unique_ptr<Writer>> Create(const std::shared_ptr<OutputStream>& output);
+ Status Close() override;
+ Status WriteIndex(const std::string& column, const std::string& type,
+ const PayloadWriter& payload) override;
+ Status Finish() override;
+
+ private:
+ explicit Writer(const std::shared_ptr<OutputStream>& output);
+
+ std::shared_ptr<OutputStream> output_;
+ std::map<std::string, std::map<std::string, std::pair<int64_t, int64_t>>> entries_;
Review Comment:
Please add a comment to explain what is stored in the map.
##########
src/paimon/core/core_options.h:
##########
@@ -224,6 +222,7 @@ class PAIMON_EXPORT CoreOptions {
bool PrepareCommitWaitCompaction() const;
bool FileIndexReadEnabled() const;
int64_t FileIndexInManifestThreshold() const;
+ int32_t FileIndexFormatVersion() const;
Review Comment:
core_options_test.cpp
##########
src/paimon/common/file_index/file_index_format_v1.cpp:
##########
@@ -0,0 +1,196 @@
+/*
+ * 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/common/file_index/file_index_format_v1.h"
+
+#include <cassert>
+#include <utility>
+
+#include "paimon/common/io/byte_array_output_stream.h"
+#include "paimon/common/io/data_output_stream.h"
+#include "paimon/common/utils/math.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/io/data_input_stream.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/status.h"
+
+namespace paimon {
+namespace {
+template <typename T>
+Status AddChecked(T value, const char* name, int64_t* total) {
+ PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(value, name));
+ *total += static_cast<int64_t>(value);
+ return ValidateValueInRange<int32_t>(*total, name);
+}
+} // namespace
+
+FileIndexFormatV1::Writer::Writer(const std::shared_ptr<OutputStream>& output,
+ const std::shared_ptr<MemoryPool>& pool)
+ : output_stream_(output), pool_(pool) {
+ assert(output_stream_);
+}
+
+Status FileIndexFormatV1::Writer::WriteIndex(const std::string& column, const std::string& type,
+ const PayloadWriter& payload) {
+ if (finished_ || closed_ || failed_) {
+ return Status::Invalid("File index writer is not writable");
+ }
+ if (buffered_indexes_[column].count(type)) {
+ return Status::Invalid("Duplicate file index entry");
+ }
+ std::shared_ptr<Bytes> bytes;
+ if (payload) {
+ auto output =
+ std::make_shared<ByteArrayOutputStream>(std::make_unique<MemorySegmentOutputStream>(
+ MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool_));
+ failed_ = true;
+ PAIMON_ASSIGN_OR_RAISE(bool written, payload(output));
+ PAIMON_ASSIGN_OR_RAISE(int64_t length, output->GetPos());
+ if (!written && length != 0) {
+ return Status::Invalid("Absent file index payload must not write bytes");
+ }
+ if (written) {
+ PAIMON_ASSIGN_OR_RAISE(bytes, output->Finish(pool_.get()));
+ }
+ failed_ = false;
+ }
+ buffered_indexes_[column][type] = std::move(bytes);
+ return Status::OK();
+}
+
+Status FileIndexFormatV1::Writer::Finish() {
+ if (finished_ || closed_ || failed_) {
+ return Status::Invalid("File index writer is not writable");
+ }
+ Status status = WriteIndexes(buffered_indexes_);
+ failed_ = !status.ok();
+ if (status.ok()) {
+ buffered_indexes_.clear();
+ }
+ return status;
+}
+
+Status FileIndexFormatV1::Writer::WriteIndexes(const FileIndexFormat::ColumnIndexes& indexes) {
+ PAIMON_RETURN_NOT_OK(WriteHead(indexes));
+ // Write body.
+ DataOutputStream data_output(output_stream_);
+ for (const auto& [column_name, column_indexes] : indexes) {
+ for (const auto& [index_type, bytes] : column_indexes) {
+ if (bytes) {
+ PAIMON_RETURN_NOT_OK(data_output.WriteBytes(bytes));
+ }
+ }
+ }
+ finished_ = true;
+ return Status::OK();
+}
+
+Status FileIndexFormatV1::Writer::Close() {
+ if (closed_) {
+ return Status::OK();
+ }
+ closed_ = true;
+ PAIMON_RETURN_NOT_OK(output_stream_->Flush());
+ return output_stream_->Close();
+}
+
+Result<int32_t> FileIndexFormatV1::Writer::CalculateHeadLength(
+ const FileIndexFormat::ColumnIndexes& indexes) {
+ // magic(8), version(4), header length(4), and column count(4).
+ int64_t head_length = 8 + 4 + 4 + 4;
+ int64_t body_length = 0;
+ PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(indexes.size(), "file index column count"));
+ for (const auto& [column_name, column_indexes] : indexes) {
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<uint16_t>(column_name.size(), "file index column name length"));
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(column_indexes.size(), "column index count"));
+ // column name(2 + N) + index count(4)
+ head_length += 2 + static_cast<int64_t>(column_name.size()) + 4;
+ for (const auto& [index_type, bytes] : column_indexes) {
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<uint16_t>(index_type.size(), "file index type name length"));
+ // index type(2 + N) + body offset(4) + body length(4)
+ head_length += 2 + static_cast<int64_t>(index_type.size()) + 4 + 4;
+ if (bytes) {
+ PAIMON_RETURN_NOT_OK(AddChecked(bytes->size(), "index body", &body_length));
+ }
+ }
+ }
+
+ head_length += 4; // The trailing redundant-length field(4).
+ PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(head_length, "file index header length"));
+ int64_t container_length = head_length + body_length;
+ PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(container_length, "file index size"));
+ return static_cast<int32_t>(head_length);
+}
+
+Status FileIndexFormatV1::Writer::WriteHead(const FileIndexFormat::ColumnIndexes& indexes) {
+ PAIMON_ASSIGN_OR_RAISE(int32_t head_length, CalculateHeadLength(indexes));
+ DataOutputStream data_output(output_stream_);
+ // Write magic.
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int64_t>(FileIndexFormat::MAGIC));
+ // Write version.
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(FileIndexFormat::V_1));
+ // Write head length.
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(head_length));
+ // Write column count.
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(static_cast<int32_t>(indexes.size())));
+
+ int64_t body_offset = head_length;
+ for (const auto& [column_name, column_indexes] : indexes) {
+ // Write column name.
+ PAIMON_RETURN_NOT_OK(data_output.WriteString(column_name));
+ // Write index count for the column.
+ PAIMON_RETURN_NOT_OK(
+ data_output.WriteValue<int32_t>(static_cast<int32_t>(column_indexes.size())));
+ for (const auto& [index_type, bytes] : column_indexes) {
+ // Write index type.
+ PAIMON_RETURN_NOT_OK(data_output.WriteString(index_type));
+ // Write body offset and length.
+ if (bytes) {
+ PAIMON_RETURN_NOT_OK(
+ data_output.WriteValue<int32_t>(static_cast<int32_t>(body_offset)));
+ PAIMON_RETURN_NOT_OK(
+ data_output.WriteValue<int32_t>(static_cast<int32_t>(bytes->size())));
+ body_offset += static_cast<int64_t>(bytes->size());
+ } else {
+ PAIMON_RETURN_NOT_OK(
+ data_output.WriteValue<int32_t>(FileIndexFormat::EMPTY_INDEX_FLAG));
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(0));
+ }
+ }
+ }
+ // Write redundant length for future format extensions.
+ return data_output.WriteValue<int32_t>(kRedundantLength);
+}
+
+Result<FileIndexEntries> FileIndexFormatV1::ReadIndexEntries(
+ const std::shared_ptr<InputStream>& input, const std::shared_ptr<MemoryPool>& pool,
+ int64_t file_length) {
+ DataInputStream data_input(input);
+ PAIMON_ASSIGN_OR_RAISE(int32_t head_length, data_input.ReadValue<int32_t>());
+ constexpr int64_t kDirectoryStart = 16;
Review Comment:
rename Directory
##########
src/paimon/common/file_index/file_index_writer.cpp:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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/file_index/file_index_writer.h"
+
+#include "paimon/fs/file_system.h"
+#include "paimon/status.h"
+
+namespace paimon {
+Result<PAIMON_UNIQUE_PTR<Bytes>> FileIndexWriter::SerializedBytes() const {
+ return Status::NotImplemented("This file index writer requires streaming output");
+}
+
+Result<bool> FileIndexWriter::WriteTo(const std::shared_ptr<OutputStream>& output) const {
+ if (!output) {
Review Comment:
OutputStream*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org