lxy-9602 commented on code in PR #210:
URL: https://github.com/apache/paimon-cpp/pull/210#discussion_r3796458971
##########
include/paimon/file_index/file_index_format.h:
##########
@@ -88,6 +91,8 @@ class MemoryPool;
class PAIMON_EXPORT FileIndexFormat {
public:
class Reader;
+ class Writer;
+ using ColumnIndexes = std::map<std::string, std::map<std::string,
std::shared_ptr<Bytes>>>;
/// Creates a `Reader` to parse a index file (may contain multiple
indexes) from the given input
Review Comment:
Please add a comment explaining what `ColumnIndexes` represents.
##########
src/paimon/common/file_index/file_index_format.cpp:
##########
@@ -39,6 +41,102 @@ namespace paimon {
class InputStream;
class MemoryPool;
+class FileIndexFormatWriterImpl : public FileIndexFormat::Writer {
+ public:
+ explicit FileIndexFormatWriterImpl(const std::shared_ptr<OutputStream>&
output_stream)
+ : output_stream_(output_stream) {
+ assert(output_stream_);
+ }
+
+ Status WriteColumnIndexes(const FileIndexFormat::ColumnIndexes& indexes)
override {
+ if (written_) {
+ return Status::Invalid("File index column indexes have already
been written");
+ }
+ int64_t header_length = sizeof(int64_t) + sizeof(int32_t) * 3 +
sizeof(int32_t);
Review Comment:
Please add comments for header_length, why not `sizeof(int32_t) * 4`
##########
src/paimon/common/file_index/file_index_format_test.cpp:
##########
Review Comment:
The bytes in this test look very similar to those in
`TestWriteEmptyIndexGoldenBytes`; both are for the empty case. Can we reuse
them instead of duplicating them?
##########
src/paimon/common/io/memory_segment_output_stream.cpp:
##########
@@ -54,12 +54,20 @@ void MemorySegmentOutputStream::WriteString(const
std::string& str) {
}
void MemorySegmentOutputStream::Write(const char* data, uint32_t size) {
- auto bytes = std::make_shared<Bytes>(size, pool_.get());
- if (size != 0) {
- memcpy(bytes->data(), data, size);
Review Comment:
Why doesn’t `Write(const char* data, uint32_t size)` delegate to
`Write(const MemorySegment& segment, int32_t offset, int32_t len)`?
`MemorySegment` can be constructed via `WrapView`.
##########
src/paimon/common/io/byte_array_output_stream.cpp:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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/io/byte_array_output_stream.h"
+
+#include <algorithm>
+#include <limits>
+#include <vector>
+
+#include "paimon/common/memory/memory_segment_utils.h"
+#include "paimon/common/utils/math.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+
+namespace paimon {
+
+ByteArrayOutputStream::ByteArrayOutputStream(int32_t initial_capacity,
+ const
std::shared_ptr<MemoryPool>& pool)
+ : pool_(pool), output_(initial_capacity, pool_) {}
+
+Result<int64_t> ByteArrayOutputStream::Write(const char* buffer, int64_t size)
{
+ if (closed_) {
+ return Status::Invalid("Byte array output stream is closed");
+ }
+ PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(size, "write length"));
+ if (buffer == nullptr && size > 0) {
+ return Status::Invalid("Write buffer must not be null when size is
positive");
+ }
+ int64_t remaining = size;
+ while (remaining > 0) {
+ uint32_t to_write = static_cast<uint32_t>(std::min<int64_t>(
+ remaining,
static_cast<int64_t>(std::numeric_limits<uint32_t>::max())));
+ output_.Write(buffer, to_write);
+ buffer += to_write;
+ remaining -= to_write;
+ }
+ position_ += size;
+ return size;
+}
+
+Status ByteArrayOutputStream::Close() {
+ closed_ = true;
+ return Status::OK();
+}
+
+Result<std::shared_ptr<Bytes>> ByteArrayOutputStream::Finish() {
+ PAIMON_RETURN_NOT_OK(Close());
+ if (result_) {
+ return result_;
+ }
+ // TODO(jinli.zjw): Support int64_t lengths in
MemorySegmentUtils::CopyToBytes and remove this
+ // limit.
+ if (position_ > std::numeric_limits<int32_t>::max()) {
+ return Status::Invalid("Byte array output stream size exceeds
INT32_MAX");
Review Comment:
Why isn’t `ValidateValueInRange` used here?
##########
src/paimon/common/file_index/file_index_format.cpp:
##########
@@ -39,6 +41,102 @@ namespace paimon {
class InputStream;
class MemoryPool;
+class FileIndexFormatWriterImpl : public FileIndexFormat::Writer {
+ public:
+ explicit FileIndexFormatWriterImpl(const std::shared_ptr<OutputStream>&
output_stream)
+ : output_stream_(output_stream) {
+ assert(output_stream_);
+ }
+
+ Status WriteColumnIndexes(const FileIndexFormat::ColumnIndexes& indexes)
override {
+ if (written_) {
+ return Status::Invalid("File index column indexes have already
been written");
+ }
+ int64_t header_length = sizeof(int64_t) + sizeof(int32_t) * 3 +
sizeof(int32_t);
+ int64_t body_length = 0;
+ for (const auto& [column_name, column_indexes] : indexes) {
+
PAIMON_RETURN_NOT_OK(ValidateValueInRange<uint16_t>(column_name.size(),
+ "file index
column name length"));
+ header_length +=
+ sizeof(uint16_t) + static_cast<int64_t>(column_name.size()) +
sizeof(int32_t);
+ for (const auto& [index_type, bytes] : column_indexes) {
+
PAIMON_RETURN_NOT_OK(ValidateValueInRange<uint16_t>(index_type.size(),
+ "file
index type name length"));
+ header_length += sizeof(uint16_t) +
static_cast<int64_t>(index_type.size()) +
+ sizeof(int32_t) * 2;
+ if (bytes) {
+ PAIMON_RETURN_NOT_OK(AddChecked(bytes->size(),
&body_length, "index body"));
+ }
+ }
+ }
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(header_length, "file index header
length"));
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(indexes.size(), "file index column
count"));
+ PAIMON_RETURN_NOT_OK(AddChecked(header_length, &body_length, "file
index size"));
+
+ DataOutputStream data_output(output_stream_);
+
PAIMON_RETURN_NOT_OK(data_output.WriteValue<int64_t>(FileIndexFormat::MAGIC));
+
PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(FileIndexFormat::V_1));
+
PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(static_cast<int32_t>(header_length)));
+
PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(static_cast<int32_t>(indexes.size())));
+
+ int64_t body_offset = header_length;
+ for (const auto& [column_name, column_indexes] : indexes) {
+ PAIMON_RETURN_NOT_OK(data_output.WriteString(column_name));
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(column_indexes.size(), "column
index count"));
+ PAIMON_RETURN_NOT_OK(
+
data_output.WriteValue<int32_t>(static_cast<int32_t>(column_indexes.size())));
+ for (const auto& [index_type, bytes] : column_indexes) {
+ PAIMON_RETURN_NOT_OK(data_output.WriteString(index_type));
+ if (bytes == nullptr) {
+ PAIMON_RETURN_NOT_OK(
+
data_output.WriteValue<int32_t>(FileIndexFormat::EMPTY_INDEX_FLAG));
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(0));
+ continue;
+ }
+ 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());
+ }
+ }
+ PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(0));
+ 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));
+ }
+ }
+ }
+ written_ = true;
+ return Status::OK();
+ }
+
+ Status Close() override {
+ if (closed_) {
+ return Status::OK();
+ }
+ closed_ = true;
+ PAIMON_RETURN_NOT_OK(output_stream_->Flush());
+ return output_stream_->Close();
+ }
+
+ private:
+ template <typename T>
+ static Status AddChecked(T value, int64_t* total, const char* name) {
+ PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(value, name));
Review Comment:
If `total` is an output param, please move to the last.
##########
src/paimon/common/io/byte_array_output_stream.cpp:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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/io/byte_array_output_stream.h"
+
+#include <algorithm>
+#include <limits>
+#include <vector>
+
+#include "paimon/common/memory/memory_segment_utils.h"
+#include "paimon/common/utils/math.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+
+namespace paimon {
+
+ByteArrayOutputStream::ByteArrayOutputStream(int32_t initial_capacity,
+ const
std::shared_ptr<MemoryPool>& pool)
+ : pool_(pool), output_(initial_capacity, pool_) {}
+
+Result<int64_t> ByteArrayOutputStream::Write(const char* buffer, int64_t size)
{
+ if (closed_) {
+ return Status::Invalid("Byte array output stream is closed");
+ }
+ PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(size, "write length"));
+ if (buffer == nullptr && size > 0) {
+ return Status::Invalid("Write buffer must not be null when size is
positive");
+ }
+ int64_t remaining = size;
+ while (remaining > 0) {
+ uint32_t to_write = static_cast<uint32_t>(std::min<int64_t>(
+ remaining,
static_cast<int64_t>(std::numeric_limits<uint32_t>::max())));
+ output_.Write(buffer, to_write);
+ buffer += to_write;
+ remaining -= to_write;
+ }
+ position_ += size;
+ return size;
+}
+
+Status ByteArrayOutputStream::Close() {
+ closed_ = true;
+ return Status::OK();
+}
+
+Result<std::shared_ptr<Bytes>> ByteArrayOutputStream::Finish() {
+ PAIMON_RETURN_NOT_OK(Close());
+ if (result_) {
+ return result_;
+ }
+ // TODO(jinli.zjw): Support int64_t lengths in
MemorySegmentUtils::CopyToBytes and remove this
+ // limit.
+ if (position_ > std::numeric_limits<int32_t>::max()) {
+ return Status::Invalid("Byte array output stream size exceeds
INT32_MAX");
+ }
+ const std::vector<MemorySegment>& segments = output_.Segments();
+ result_ = std::shared_ptr<Bytes>(new Bytes(static_cast<size_t>(position_),
pool_.get()),
+ [pool = pool_](Bytes* bytes) { delete
bytes; });
+ MemorySegmentUtils::CopyToBytes(segments, /*offset=*/0, result_.get(),
Review Comment:
This implementation feels a bit unusual. Would it make sense to add a helper
on `Bytes` instead, for example:
```cpp
static std::shared_ptr<Bytes> AllocateShared(
size_t size, std::shared_ptr<MemoryPool> pool);
```
--
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]
