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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
