You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2022/08/04 09:44:58 UTC
[doris] branch master updated: [feature-wip](parquet-reader)get compressed parquet page data (#11493)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new aed0282046 [feature-wip](parquet-reader)get compressed parquet page data (#11493)
aed0282046 is described below
commit aed0282046cfa00ad560ae898bc33cc23d5624d8
Author: Ashin Gau <As...@users.noreply.github.com>
AuthorDate: Thu Aug 4 17:44:52 2022 +0800
[feature-wip](parquet-reader)get compressed parquet page data (#11493)
---
be/src/io/buffered_reader.cpp | 61 ++++++++++++++++++++++
be/src/io/buffered_reader.h | 41 +++++++++++++++
be/src/vec/exec/format/parquet/schema_desc.cpp | 16 +++---
.../exec/format/parquet/vparquet_page_reader.cpp | 50 +++++++++++++++---
.../vec/exec/format/parquet/vparquet_page_reader.h | 37 ++++++++++---
5 files changed, 182 insertions(+), 23 deletions(-)
diff --git a/be/src/io/buffered_reader.cpp b/be/src/io/buffered_reader.cpp
index bd0deb823c..8e2446b9e1 100644
--- a/be/src/io/buffered_reader.cpp
+++ b/be/src/io/buffered_reader.cpp
@@ -22,6 +22,7 @@
#include "common/config.h"
#include "olap/olap_define.h"
+#include "util/bit_util.h"
namespace doris {
@@ -184,4 +185,64 @@ bool BufferedReader::closed() {
return _reader->closed();
}
+BufferedFileStreamReader::BufferedFileStreamReader(FileReader* file, int64_t offset, int64_t length)
+ : _file(file), _file_start_offset(offset), _file_end_offset(offset + length) {}
+
+Status BufferedFileStreamReader::seek(int64_t position) {
+ if (_file_position != position) {
+ RETURN_IF_ERROR(_file->seek(position));
+ _file_position = position;
+ }
+ return Status::OK();
+}
+
+Status BufferedFileStreamReader::read_bytes(const uint8_t** buf, int64_t offset,
+ int64_t* bytes_to_read) {
+ if (offset < _file_start_offset) {
+ return Status::IOError("Out-of-bounds Access");
+ }
+ if (offset >= _file_end_offset) {
+ *bytes_to_read = 0;
+ return Status::OK();
+ }
+ int64_t end_offset = offset + *bytes_to_read;
+ if (_buf_start_offset <= offset && _buf_end_offset >= end_offset) {
+ *buf = _buf.get() + offset - _buf_start_offset;
+ return Status::OK();
+ }
+ if (_buf_size < *bytes_to_read) {
+ size_t new_size = BitUtil::next_power_of_two(*bytes_to_read);
+ std::unique_ptr<uint8_t[]> new_buf(new uint8_t[new_size]);
+ if (offset >= _buf_start_offset && offset < _buf_end_offset) {
+ memcpy(new_buf.get(), _buf.get() + offset - _buf_start_offset,
+ _buf_end_offset - offset);
+ }
+ _buf = std::move(new_buf);
+ _buf_size = new_size;
+ } else if (offset > _buf_start_offset && offset < _buf_end_offset) {
+ memmove(_buf.get(), _buf.get() + offset - _buf_start_offset, _buf_end_offset - offset);
+ }
+ if (offset < _buf_start_offset || offset >= _buf_end_offset) {
+ _buf_end_offset = offset;
+ }
+ _buf_start_offset = offset;
+ int64_t to_read = end_offset - _buf_end_offset;
+ RETURN_IF_ERROR(seek(_buf_end_offset));
+ bool eof = false;
+ int64_t buf_remaining = _buf_end_offset - _buf_start_offset;
+ RETURN_IF_ERROR(
+ _file->read(_buf.get() + buf_remaining, _buf_size - buf_remaining, &to_read, &eof));
+ *bytes_to_read = buf_remaining + to_read;
+ _buf_end_offset += to_read;
+ *buf = _buf.get();
+ return Status::OK();
+}
+
+Status BufferedFileStreamReader::read_bytes(Slice& slice, int64_t offset) {
+ int64_t bytes_to_read = slice.size;
+ Status st = read_bytes((const uint8_t**)&slice.data, offset, &bytes_to_read);
+ slice.size = bytes_to_read;
+ return st;
+}
+
} // namespace doris
diff --git a/be/src/io/buffered_reader.h b/be/src/io/buffered_reader.h
index 75dec5cf7f..d4a5f37927 100644
--- a/be/src/io/buffered_reader.h
+++ b/be/src/io/buffered_reader.h
@@ -82,4 +82,45 @@ private:
RuntimeProfile::Counter* _remote_read_rate = nullptr;
};
+/**
+ * Load all the needed data in underlying buffer, so the caller does not need to prepare the data container.
+ */
+class BufferedStreamReader {
+public:
+ /**
+ * Return the address of underlying buffer that locates the start of data between [offset, offset + bytes_to_read)
+ * @param buf the buffer address to save the start address of data
+ * @param offset start offset ot read in stream
+ * @param bytes_to_read bytes to read
+ */
+ virtual Status read_bytes(const uint8_t** buf, int64_t offset, int64_t* bytes_to_read) = 0;
+ /**
+ * Save the data address to slice.data, and the slice.size is the bytes to read.
+ */
+ virtual Status read_bytes(Slice& slice, int64_t offset) = 0;
+ virtual ~BufferedStreamReader() = default;
+};
+
+class BufferedFileStreamReader : public BufferedStreamReader {
+public:
+ BufferedFileStreamReader(FileReader* file, int64_t offset, int64_t length);
+ ~BufferedFileStreamReader() override = default;
+
+ Status read_bytes(const uint8_t** buf, int64_t stream_offset, int64_t* bytes_to_read) override;
+ Status read_bytes(Slice& slice, int64_t stream_offset) override;
+
+private:
+ std::unique_ptr<uint8_t[]> _buf;
+ FileReader* _file;
+ int64_t _file_start_offset;
+ int64_t _file_end_offset;
+
+ int64_t _file_position = -1;
+ int64_t _buf_start_offset = 0;
+ int64_t _buf_end_offset = 0;
+ int64_t _buf_size = 0;
+
+ Status seek(int64_t position);
+};
+
} // namespace doris
diff --git a/be/src/vec/exec/format/parquet/schema_desc.cpp b/be/src/vec/exec/format/parquet/schema_desc.cpp
index 88b3210341..51969220ed 100644
--- a/be/src/vec/exec/format/parquet/schema_desc.cpp
+++ b/be/src/vec/exec/format/parquet/schema_desc.cpp
@@ -99,13 +99,13 @@ Status FieldDescriptor::parse_from_thrift(const std::vector<tparquet::SchemaElem
RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, &_fields[i]));
if (_name_to_field.find(_fields[i].name) != _name_to_field.end()) {
return Status::InvalidArgument(
- strings::Substitute("Duplicated field name: $0", _fields[i].name));
+ strings::Substitute("Duplicated field name: {}", _fields[i].name));
}
_name_to_field.emplace(_fields[i].name, &_fields[i]);
}
if (_next_schema_pos != t_schemas.size()) {
- return Status::InvalidArgument(strings::Substitute("Remaining $0 unparsed schema elements",
+ return Status::InvalidArgument(strings::Substitute("Remaining {} unparsed schema elements",
t_schemas.size() - _next_schema_pos));
}
@@ -204,27 +204,27 @@ Status FieldDescriptor::parse_list_field(const std::vector<tparquet::SchemaEleme
size_t curr_pos, FieldSchema* list_field) {
// the list definition:
// spark and hive have three level schemas but with different schema name
- // spark: <column-name> - "list" - "list_child"
+ // spark: <column-name> - "list" - "element"
// hive: <column-name> - "bag" - "array_element"
// parse three level schemas to two level primitive like: LIST<INT>,
// or nested structure like: LIST<MAP<INT, INT>>
auto& first_level = t_schemas[curr_pos];
if (first_level.num_children != 1) {
- return Status::InvalidArgument("List list_child should have only one child");
+ return Status::InvalidArgument("List element should have only one child");
}
if (curr_pos + 1 >= t_schemas.size()) {
- return Status::InvalidArgument("List list_child should have the second level schema");
+ return Status::InvalidArgument("List element should have the second level schema");
}
if (first_level.repetition_type == tparquet::FieldRepetitionType::REPEATED) {
- return Status::InvalidArgument("List list_child can't be a repeated schema");
+ return Status::InvalidArgument("List element can't be a repeated schema");
}
- // the repeated schema list_child
+ // the repeated schema element
auto& second_level = t_schemas[curr_pos + 1];
if (second_level.repetition_type != tparquet::FieldRepetitionType::REPEATED) {
- return Status::InvalidArgument("The second level of list list_child should be repeated");
+ return Status::InvalidArgument("The second level of list element should be repeated");
}
// This indicates if this list is nullable.
diff --git a/be/src/vec/exec/format/parquet/vparquet_page_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_page_reader.cpp
index d781159cab..533caa3db0 100644
--- a/be/src/vec/exec/format/parquet/vparquet_page_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_page_reader.cpp
@@ -17,17 +17,51 @@
#include "vparquet_page_reader.h"
+#include <stdint.h>
+
+#include "util/thrift_util.h"
+
namespace doris::vectorized {
-Status PageReader::read_page_header() {
- return Status();
-}
+static constexpr int64_t initPageHeaderSize = 1024;
-Status PageReader::read_page_data() {
- return Status();
-}
+PageReader::PageReader(BufferedStreamReader* reader, int64_t start_offset, int64_t length)
+ : _reader(reader), _start_offset(start_offset), _end_offset(start_offset + length) {}
+
+Status PageReader::next_page(Slice& slice) {
+ if (_offset < _start_offset || _offset >= _end_offset) {
+ return Status::IOError("Out-of-bounds Access");
+ }
+ if (_offset != _next_header_offset) {
+ return Status::IOError("Wrong header position");
+ }
+
+ const uint8_t* page_header_buf = nullptr;
+ int64_t max_size = _end_offset - _offset;
+ int64_t header_size = std::min(initPageHeaderSize, max_size);
+ uint32_t real_header_size = 0;
+ while (true) {
+ header_size = std::min(header_size, max_size);
+ RETURN_IF_ERROR(_reader->read_bytes(&page_header_buf, _offset, &header_size));
+ real_header_size = header_size;
+ auto st =
+ deserialize_thrift_msg(page_header_buf, &real_header_size, true, &_cur_page_header);
+ if (st.ok()) {
+ break;
+ }
+ if (_offset + header_size >= _end_offset) {
+ return Status::IOError("Failed to deserialize parquet page header");
+ }
+ header_size <<= 2;
+ }
+
+ _offset += real_header_size;
+ _next_header_offset = _offset + _cur_page_header.compressed_page_size;
-Status PageReader::init() {
- return Status();
+ slice.size = _cur_page_header.compressed_page_size;
+ RETURN_IF_ERROR(_reader->read_bytes(slice, _offset));
+ DCHECK_EQ(slice.size, _cur_page_header.compressed_page_size);
+ _offset += slice.size;
+ return Status::OK();
}
} // namespace doris::vectorized
diff --git a/be/src/vec/exec/format/parquet/vparquet_page_reader.h b/be/src/vec/exec/format/parquet/vparquet_page_reader.h
index 9b66896aaf..523098cc44 100644
--- a/be/src/vec/exec/format/parquet/vparquet_page_reader.h
+++ b/be/src/vec/exec/format/parquet/vparquet_page_reader.h
@@ -16,19 +16,42 @@
// under the License.
#pragma once
-#include <common/status.h>
-#include <gen_cpp/parquet_types.h>
+
+#include "common/status.h"
+#include "gen_cpp/parquet_types.h"
+#include "io/buffered_reader.h"
namespace doris::vectorized {
+/**
+ * Use to deserialize parquet page header, and get the page data in iterator interface.
+ */
class PageReader {
public:
- Status init();
- Status read_page_header();
- Status read_page_data();
+public:
+ PageReader(BufferedStreamReader* reader, int64_t start_offset, int64_t length);
+ ~PageReader() = default;
+
+ bool has_next_page() const { return _offset < _end_offset; }
+
+ Status next_page(Slice& slice);
+
+ const tparquet::PageHeader* get_page_header() const { return &_cur_page_header; }
+
+ void seek_to_page(int64_t page_header_offset) {
+ _offset = page_header_offset;
+ _next_header_offset = page_header_offset;
+ }
+
+private:
+ BufferedStreamReader* _reader;
+ tparquet::PageHeader _cur_page_header;
+
+ int64_t _offset = 0;
+ int64_t _next_header_offset = 0;
- //private:
- // tparquet::PageHeader* _page_header;
+ int64_t _start_offset = 0;
+ int64_t _end_offset = 0;
};
} // namespace doris::vectorized
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org