You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by GitBox <gi...@apache.org> on 2022/08/08 12:58:26 UTC
[GitHub] [doris] wsjz opened a new pull request, #11601: [feature-wip](parquet-reader) use batch column read model
wsjz opened a new pull request, #11601:
URL: https://github.com/apache/doris/pull/11601
# Proposed changes
Issue Number: close #xxx
## Problem summary
Describe your changes.
## Checklist(Required)
1. Does it affect the original behavior:
- [ ] Yes
- [ ] No
- [ ] I don't know
2. Has unit tests been added:
- [ ] Yes
- [ ] No
- [ ] No Need
3. Has document been added or modified:
- [ ] Yes
- [ ] No
- [ ] No Need
4. Does it need to update dependencies:
- [ ] Yes
- [ ] No
5. Are there any changes that cannot be rolled back:
- [ ] Yes (If Yes, please explain WHY)
- [ ] No
## Further comments
If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] wsjz commented on a diff in pull request #11601: [feature-wip](parquet-reader) update column read model and add page index
Posted by GitBox <gi...@apache.org>.
wsjz commented on code in PR #11601:
URL: https://github.com/apache/doris/pull/11601#discussion_r945651384
##########
be/src/vec/exec/format/parquet/vparquet_reader.cpp:
##########
@@ -90,63 +95,240 @@ Status ParquetReader::_init_read_columns(const std::vector<SlotDescriptor*>& tup
auto physical_type = _file_metadata->schema().get_column(parquet_col_id)->physical_type;
column.parquet_type = physical_type;
_read_columns.emplace_back(column);
+ VLOG_DEBUG << "slot_desc " << slot_desc->debug_string();
}
return Status::OK();
}
-Status ParquetReader::read_next_batch(Block* block) {
- int32_t group_id = 0;
- RETURN_IF_ERROR(_row_group_reader->get_next_row_group(&group_id));
+Status ParquetReader::read_next_batch(Block* block, bool* eof) {
+ DCHECK(_total_groups == _row_group_readers.size());
+ if (_total_groups == 0) {
+ *eof = true;
+ }
+ bool _batch_eof = false;
+ auto row_group_reader = _row_group_readers[_current_row_group_id];
+ RETURN_IF_ERROR(row_group_reader->next_batch(block, _batch_size, &_batch_eof));
+ if (_batch_eof) {
+ _current_row_group_id++;
+ if (_current_row_group_id > _total_groups) {
+ *eof = true;
+ }
Review Comment:
The `_current_row_group_id++;` statement can do this，so it's implicit to mark `*eof = true`
##########
be/src/vec/exec/format/parquet/vparquet_reader.cpp:
##########
@@ -90,63 +95,240 @@ Status ParquetReader::_init_read_columns(const std::vector<SlotDescriptor*>& tup
auto physical_type = _file_metadata->schema().get_column(parquet_col_id)->physical_type;
column.parquet_type = physical_type;
_read_columns.emplace_back(column);
+ VLOG_DEBUG << "slot_desc " << slot_desc->debug_string();
}
return Status::OK();
}
-Status ParquetReader::read_next_batch(Block* block) {
- int32_t group_id = 0;
- RETURN_IF_ERROR(_row_group_reader->get_next_row_group(&group_id));
+Status ParquetReader::read_next_batch(Block* block, bool* eof) {
+ DCHECK(_total_groups == _row_group_readers.size());
+ if (_total_groups == 0) {
+ *eof = true;
+ }
+ bool _batch_eof = false;
+ auto row_group_reader = _row_group_readers[_current_row_group_id];
+ RETURN_IF_ERROR(row_group_reader->next_batch(block, _batch_size, &_batch_eof));
+ if (_batch_eof) {
+ _current_row_group_id++;
+ if (_current_row_group_id > _total_groups) {
+ *eof = true;
+ }
Review Comment:
The `_current_row_group_id++;` statement can do this，so it's implicit to mark `*eof =false`
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] wsjz commented on a diff in pull request #11601: [feature-wip](parquet-reader) update column read model and add page index
Posted by GitBox <gi...@apache.org>.
wsjz commented on code in PR #11601:
URL: https://github.com/apache/doris/pull/11601#discussion_r945657012
##########
be/src/vec/exec/format/parquet/vparquet_column_chunk_reader.cpp:
##########
@@ -43,11 +44,12 @@ Status ColumnChunkReader::init() {
// get the block compression codec
RETURN_IF_ERROR(get_block_compression_codec(_metadata.codec, _block_compress_codec));
+ VLOG_DEBUG << "initColumnChunkReader finish";
return Status::OK();
}
Status ColumnChunkReader::next_page() {
- RETURN_IF_ERROR(_page_reader->next_page());
+ RETURN_IF_ERROR(_page_reader->next_page_header());
_num_values = _page_reader->get_page_header()->data_page_header.num_values;
Review Comment:
remaining
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] wsjz commented on a diff in pull request #11601: [feature-wip](parquet-reader) update column read model and add page index
Posted by GitBox <gi...@apache.org>.
wsjz commented on code in PR #11601:
URL: https://github.com/apache/doris/pull/11601#discussion_r945651384
##########
be/src/vec/exec/format/parquet/vparquet_reader.cpp:
##########
@@ -90,63 +95,240 @@ Status ParquetReader::_init_read_columns(const std::vector<SlotDescriptor*>& tup
auto physical_type = _file_metadata->schema().get_column(parquet_col_id)->physical_type;
column.parquet_type = physical_type;
_read_columns.emplace_back(column);
+ VLOG_DEBUG << "slot_desc " << slot_desc->debug_string();
}
return Status::OK();
}
-Status ParquetReader::read_next_batch(Block* block) {
- int32_t group_id = 0;
- RETURN_IF_ERROR(_row_group_reader->get_next_row_group(&group_id));
+Status ParquetReader::read_next_batch(Block* block, bool* eof) {
+ DCHECK(_total_groups == _row_group_readers.size());
+ if (_total_groups == 0) {
+ *eof = true;
+ }
+ bool _batch_eof = false;
+ auto row_group_reader = _row_group_readers[_current_row_group_id];
+ RETURN_IF_ERROR(row_group_reader->next_batch(block, _batch_size, &_batch_eof));
+ if (_batch_eof) {
+ _current_row_group_id++;
+ if (_current_row_group_id > _total_groups) {
+ *eof = true;
+ }
Review Comment:
The `_current_row_group_id++;` statement can do this，so it's implicit to mark `*eof =false`, the `next_batch() ` will be called in the upper.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] wsjz commented on a diff in pull request #11601: [feature-wip](parquet-reader) update column read model and add page index
Posted by GitBox <gi...@apache.org>.
wsjz commented on code in PR #11601:
URL: https://github.com/apache/doris/pull/11601#discussion_r945666374
##########
be/src/vec/exec/format/parquet/vparquet_column_reader.cpp:
##########
@@ -19,50 +19,82 @@
#include <common/status.h>
#include <gen_cpp/parquet_types.h>
+#include <vec/columns/columns_number.h>
#include "schema_desc.h"
#include "vparquet_column_chunk_reader.h"
namespace doris::vectorized {
-Status ScalarColumnReader::init(const FileReader* file, const FieldSchema* field,
- const tparquet::ColumnChunk* chunk, const TypeDescriptor& col_type,
- int64_t chunk_size) {
- // todo1: init column chunk reader
- // BufferedFileStreamReader stream_reader(reader, 0, chunk_size);
- // _chunk_reader(&stream_reader, chunk, field);
- // _chunk_reader.init();
- return Status();
-}
-
-Status ParquetColumnReader::create(const FileReader* file, int64_t chunk_size,
- const FieldSchema* field, const ParquetReadColumn& column,
- const TypeDescriptor& col_type,
+Status ParquetColumnReader::create(FileReader* file, FieldSchema* field,
+ const ParquetReadColumn& column,
const tparquet::RowGroup& row_group,
- const ParquetColumnReader* reader) {
+ std::vector<RowRange>& row_ranges,
+ std::unique_ptr<ParquetColumnReader>& reader) {
if (field->type.type == TYPE_MAP || field->type.type == TYPE_STRUCT) {
return Status::Corruption("not supported type");
}
if (field->type.type == TYPE_ARRAY) {
return Status::Corruption("not supported array type yet");
} else {
+ VLOG_DEBUG << "field->physical_column_index: " << field->physical_column_index;
+ tparquet::ColumnChunk chunk = row_group.columns[field->physical_column_index];
ScalarColumnReader* scalar_reader = new ScalarColumnReader(column);
- RETURN_IF_ERROR(scalar_reader->init(file, field,
- &row_group.columns[field->physical_column_index],
- col_type, chunk_size));
- reader = scalar_reader;
+ scalar_reader->init_column_metadata(chunk);
+ RETURN_IF_ERROR(scalar_reader->init(file, field, &chunk, row_ranges));
+ reader.reset(scalar_reader);
}
return Status::OK();
}
-Status ScalarColumnReader::read_column_data(const tparquet::RowGroup& row_group_meta,
- ColumnPtr* data) {
- // todo2: read data with chunk reader to load page data
- // while (_chunk_reader.has_next) {
- // _chunk_reader.next_page();
- // _chunk_reader.load_page_data();
- // }
- return Status();
+void ParquetColumnReader::init_column_metadata(const tparquet::ColumnChunk& chunk) {
+ auto chunk_meta = chunk.meta_data;
+ int64_t chunk_start = chunk_meta.__isset.dictionary_page_offset
+ ? chunk_meta.dictionary_page_offset
+ : chunk_meta.data_page_offset;
+ size_t chunk_len = chunk_meta.total_compressed_size;
+ _metadata.reset(new ParquetColumnMetadata(chunk_start, chunk_len, chunk_meta));
+}
+
+void ParquetColumnReader::_skipped_pages() {}
+
+Status ScalarColumnReader::init(FileReader* file, FieldSchema* field, tparquet::ColumnChunk* chunk,
+ std::vector<RowRange>& row_ranges) {
+ BufferedFileStreamReader stream_reader(file, _metadata->start_offset(), _metadata->size());
+ _row_ranges.reset(&row_ranges);
+ _chunk_reader.reset(new ColumnChunkReader(&stream_reader, chunk, field));
+ _chunk_reader->init();
+ return Status::OK();
+}
+
+Status ScalarColumnReader::read_column_data(ColumnPtr& doris_column, DataTypePtr& type,
+ size_t batch_size, int64_t* read_rows, bool* eof) {
+ if (_chunk_reader->num_values() <= 0) {
+ // seek to next page header
+ _chunk_reader->next_page();
+ if (_row_ranges->size() != 0) {
+ _skipped_pages();
+ }
+ // load data to decoder
+ _chunk_reader->load_page_data();
+ }
+ size_t read_values =
+ _chunk_reader->num_values() < batch_size ? _chunk_reader->num_values() : batch_size;
Review Comment:
default size of a block is `batch_size`, I think we should limit it
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] morningman merged pull request #11601: [feature-wip](parquet-reader) update column read model and add page index
Posted by GitBox <gi...@apache.org>.
morningman merged PR #11601:
URL: https://github.com/apache/doris/pull/11601
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] morningman commented on a diff in pull request #11601: [feature-wip](parquet-reader) update column read model and add page index
Posted by GitBox <gi...@apache.org>.
morningman commented on code in PR #11601:
URL: https://github.com/apache/doris/pull/11601#discussion_r945588010
##########
be/src/vec/exec/format/parquet/vparquet_reader.h:
##########
@@ -58,18 +64,18 @@ class ParquetReadColumn {
class ParquetReader {
public:
- ParquetReader(FileReader* file_reader, int32_t num_of_columns_from_file,
+ ParquetReader(FileReader* file_reader, int32_t num_of_columns_from_file, size_t batch_size,
int64_t range_start_offset, int64_t range_size);
~ParquetReader();
Status init_reader(const TupleDescriptor* tuple_desc,
const std::vector<SlotDescriptor*>& tuple_slot_descs,
- const std::vector<ExprContext*>& conjunct_ctxs, const std::string& timezone);
+ std::vector<ExprContext*>& conjunct_ctxs, const std::string& timezone);
- Status read_next_batch(Block* block);
+ Status read_next_batch(Block* block, bool* eof);
- bool has_next() const { return !_batch_eof; };
+ bool has_next() const { return !*_file_eof; };
Review Comment:
the `eof` of `read_next_batch` already told you the end-of-file info. No need this `has_next()` method.
##########
be/src/vec/exec/format/parquet/vparquet_reader.cpp:
##########
@@ -45,26 +50,26 @@ void ParquetReader::close() {
Status ParquetReader::init_reader(const TupleDescriptor* tuple_desc,
const std::vector<SlotDescriptor*>& tuple_slot_descs,
- const std::vector<ExprContext*>& conjunct_ctxs,
+ std::vector<ExprContext*>& conjunct_ctxs,
const std::string& timezone) {
_file_reader->open();
+ _conjunct_ctxs.reset(&conjunct_ctxs);
RETURN_IF_ERROR(parse_thrift_footer(_file_reader, _file_metadata));
auto metadata = _file_metadata->to_thrift_metadata();
Review Comment:
This `metadata` is unused?
You can save it as a field of this class.
##########
be/src/vec/exec/format/parquet/vparquet_group_reader.cpp:
##########
@@ -24,235 +24,57 @@
namespace doris::vectorized {
RowGroupReader::RowGroupReader(doris::FileReader* file_reader,
- const std::shared_ptr<FileMetaData>& file_metadata,
const std::vector<ParquetReadColumn>& read_columns,
- const std::map<std::string, int>& map_column,
- const std::vector<ExprContext*>& conjunct_ctxs)
+ const int32_t row_group_id, tparquet::RowGroup& row_group)
: _file_reader(file_reader),
- _file_metadata(file_metadata),
_read_columns(read_columns),
- _map_column(map_column),
- _conjunct_ctxs(conjunct_ctxs),
- _current_row_group(-1) {}
+ _row_group_id(row_group_id),
+ _row_group_meta(row_group),
+ _total_rows(row_group.num_rows) {}
RowGroupReader::~RowGroupReader() {
- for (auto& column_reader : _column_readers) {
- auto reader = column_reader.second;
- reader->close();
- delete reader;
- reader = nullptr;
- }
_column_readers.clear();
}
-Status RowGroupReader::init(const TupleDescriptor* tuple_desc, int64_t split_start_offset,
- int64_t split_size) {
- _tuple_desc = tuple_desc;
- _split_start_offset = split_start_offset;
- _split_size = split_size;
- _init_conjuncts(tuple_desc, _conjunct_ctxs);
- RETURN_IF_ERROR(_init_column_readers());
+Status RowGroupReader::init(const FieldDescriptor& schema, std::vector<RowRange>& row_ranges) {
+ RETURN_IF_ERROR(_init_column_readers(schema, row_ranges));
return Status::OK();
}
-void RowGroupReader::_init_conjuncts(const TupleDescriptor* tuple_desc,
- const std::vector<ExprContext*>& conjunct_ctxs) {
- if (tuple_desc->slots().empty()) {
- return;
- }
- for (auto& read_col : _read_columns) {
- _parquet_column_ids.emplace(read_col.parquet_column_id);
- }
-
- for (int i = 0; i < tuple_desc->slots().size(); i++) {
- auto col_iter = _map_column.find(tuple_desc->slots()[i]->col_name());
- if (col_iter == _map_column.end()) {
- continue;
- }
- int parquet_col_id = col_iter->second;
- if (_parquet_column_ids.end() == _parquet_column_ids.find(parquet_col_id)) {
- continue;
- }
- for (int conj_idx = 0; conj_idx < conjunct_ctxs.size(); conj_idx++) {
- Expr* conjunct = conjunct_ctxs[conj_idx]->root();
- if (conjunct->get_num_children() == 0) {
- continue;
- }
- Expr* raw_slot = conjunct->get_child(0);
- if (TExprNodeType::SLOT_REF != raw_slot->node_type()) {
- continue;
- }
- SlotRef* slot_ref = (SlotRef*)raw_slot;
- SlotId conjunct_slot_id = slot_ref->slot_id();
- if (conjunct_slot_id == tuple_desc->slots()[i]->id()) {
- // Get conjuncts by conjunct_slot_id
- auto iter = _slot_conjuncts.find(conjunct_slot_id);
- if (_slot_conjuncts.end() == iter) {
- std::vector<ExprContext*> conjuncts;
- conjuncts.emplace_back(conjunct_ctxs[conj_idx]);
- _slot_conjuncts.emplace(std::make_pair(conjunct_slot_id, conjuncts));
- } else {
- std::vector<ExprContext*> conjuncts = iter->second;
- conjuncts.emplace_back(conjunct_ctxs[conj_idx]);
- }
- }
- }
- }
-}
-
-Status RowGroupReader::_init_column_readers() {
+Status RowGroupReader::_init_column_readers(const FieldDescriptor& schema,
+ std::vector<RowRange>& row_ranges) {
for (auto& read_col : _read_columns) {
SlotDescriptor* slot_desc = read_col.slot_desc;
- FieldDescriptor schema = _file_metadata->schema();
TypeDescriptor col_type = slot_desc->type();
- const auto& field = schema.get_column(slot_desc->col_name());
- const tparquet::RowGroup row_group =
- _file_metadata->to_thrift_metadata().row_groups[_current_row_group];
- ParquetColumnReader* reader = nullptr;
- RETURN_IF_ERROR(ParquetColumnReader::create(_file_reader, MAX_PARQUET_BLOCK_SIZE, field,
- read_col, slot_desc->type(), row_group,
- reader));
+ auto field = const_cast<FieldSchema*>(schema.get_column(slot_desc->col_name()));
+ VLOG_DEBUG << "field: " << field->debug_string();
+ std::unique_ptr<ParquetColumnReader> reader;
+ RETURN_IF_ERROR(ParquetColumnReader::create(_file_reader, field, read_col, _row_group_meta,
+ row_ranges, reader));
if (reader == nullptr) {
+ VLOG_DEBUG << "Init row group reader failed";
return Status::Corruption("Init row group reader failed");
}
- _column_readers[slot_desc->id()] = reader;
+ _column_readers[slot_desc->id()] = std::move(reader);
}
return Status::OK();
}
-Status RowGroupReader::fill_columns_data(Block* block, const int32_t group_id) {
- // get ColumnWithTypeAndName from src_block
+Status RowGroupReader::next_batch(Block* block, size_t batch_size, bool* _batch_eof) {
+ if (_read_rows >= _total_rows) {
+ *_batch_eof = true;
+ }
for (auto& read_col : _read_columns) {
- const tparquet::RowGroup row_group =
- _file_metadata->to_thrift_metadata().row_groups[_current_row_group];
- auto& column_with_type_and_name = block->get_by_name(read_col.slot_desc->col_name());
- RETURN_IF_ERROR(_column_readers[read_col.slot_desc->id()]->read_column_data(
- row_group, &column_with_type_and_name.column));
- VLOG_DEBUG << column_with_type_and_name.name;
+ auto slot_desc = read_col.slot_desc;
+ auto& column_with_type_and_name = block->get_by_name(slot_desc->col_name());
+ auto column_ptr = column_with_type_and_name.column;
+ auto column_type = column_with_type_and_name.type;
+ RETURN_IF_ERROR(_column_readers[slot_desc->id()]->read_column_data(
+ column_ptr, column_type, batch_size, &_read_rows, _batch_eof));
Review Comment:
`_read_rows` should be a cumulative value. But here it just be set to the number values of current batch.
##########
be/src/vec/exec/format/parquet/vparquet_column_reader.cpp:
##########
@@ -19,50 +19,82 @@
#include <common/status.h>
#include <gen_cpp/parquet_types.h>
+#include <vec/columns/columns_number.h>
#include "schema_desc.h"
#include "vparquet_column_chunk_reader.h"
namespace doris::vectorized {
-Status ScalarColumnReader::init(const FileReader* file, const FieldSchema* field,
- const tparquet::ColumnChunk* chunk, const TypeDescriptor& col_type,
- int64_t chunk_size) {
- // todo1: init column chunk reader
- // BufferedFileStreamReader stream_reader(reader, 0, chunk_size);
- // _chunk_reader(&stream_reader, chunk, field);
- // _chunk_reader.init();
- return Status();
-}
-
-Status ParquetColumnReader::create(const FileReader* file, int64_t chunk_size,
- const FieldSchema* field, const ParquetReadColumn& column,
- const TypeDescriptor& col_type,
+Status ParquetColumnReader::create(FileReader* file, FieldSchema* field,
+ const ParquetReadColumn& column,
const tparquet::RowGroup& row_group,
- const ParquetColumnReader* reader) {
+ std::vector<RowRange>& row_ranges,
+ std::unique_ptr<ParquetColumnReader>& reader) {
if (field->type.type == TYPE_MAP || field->type.type == TYPE_STRUCT) {
return Status::Corruption("not supported type");
}
if (field->type.type == TYPE_ARRAY) {
return Status::Corruption("not supported array type yet");
} else {
+ VLOG_DEBUG << "field->physical_column_index: " << field->physical_column_index;
+ tparquet::ColumnChunk chunk = row_group.columns[field->physical_column_index];
ScalarColumnReader* scalar_reader = new ScalarColumnReader(column);
- RETURN_IF_ERROR(scalar_reader->init(file, field,
- &row_group.columns[field->physical_column_index],
- col_type, chunk_size));
- reader = scalar_reader;
+ scalar_reader->init_column_metadata(chunk);
+ RETURN_IF_ERROR(scalar_reader->init(file, field, &chunk, row_ranges));
+ reader.reset(scalar_reader);
}
return Status::OK();
}
-Status ScalarColumnReader::read_column_data(const tparquet::RowGroup& row_group_meta,
- ColumnPtr* data) {
- // todo2: read data with chunk reader to load page data
- // while (_chunk_reader.has_next) {
- // _chunk_reader.next_page();
- // _chunk_reader.load_page_data();
- // }
- return Status();
+void ParquetColumnReader::init_column_metadata(const tparquet::ColumnChunk& chunk) {
+ auto chunk_meta = chunk.meta_data;
+ int64_t chunk_start = chunk_meta.__isset.dictionary_page_offset
+ ? chunk_meta.dictionary_page_offset
+ : chunk_meta.data_page_offset;
+ size_t chunk_len = chunk_meta.total_compressed_size;
+ _metadata.reset(new ParquetColumnMetadata(chunk_start, chunk_len, chunk_meta));
+}
+
+void ParquetColumnReader::_skipped_pages() {}
+
+Status ScalarColumnReader::init(FileReader* file, FieldSchema* field, tparquet::ColumnChunk* chunk,
+ std::vector<RowRange>& row_ranges) {
+ BufferedFileStreamReader stream_reader(file, _metadata->start_offset(), _metadata->size());
+ _row_ranges.reset(&row_ranges);
+ _chunk_reader.reset(new ColumnChunkReader(&stream_reader, chunk, field));
+ _chunk_reader->init();
+ return Status::OK();
+}
+
+Status ScalarColumnReader::read_column_data(ColumnPtr& doris_column, DataTypePtr& type,
+ size_t batch_size, int64_t* read_rows, bool* eof) {
+ if (_chunk_reader->num_values() <= 0) {
+ // seek to next page header
+ _chunk_reader->next_page();
+ if (_row_ranges->size() != 0) {
+ _skipped_pages();
+ }
+ // load data to decoder
+ _chunk_reader->load_page_data();
+ }
+ size_t read_values =
+ _chunk_reader->num_values() < batch_size ? _chunk_reader->num_values() : batch_size;
Review Comment:
I think we don't need to check the number of values of a `_chunk_reader`.
Just call `_chunk_reader->decode_values` and return the number of values read.
##########
be/src/vec/exec/format/parquet/vparquet_reader.cpp:
##########
@@ -90,63 +95,240 @@ Status ParquetReader::_init_read_columns(const std::vector<SlotDescriptor*>& tup
auto physical_type = _file_metadata->schema().get_column(parquet_col_id)->physical_type;
column.parquet_type = physical_type;
_read_columns.emplace_back(column);
+ VLOG_DEBUG << "slot_desc " << slot_desc->debug_string();
}
return Status::OK();
}
-Status ParquetReader::read_next_batch(Block* block) {
- int32_t group_id = 0;
- RETURN_IF_ERROR(_row_group_reader->get_next_row_group(&group_id));
+Status ParquetReader::read_next_batch(Block* block, bool* eof) {
+ DCHECK(_total_groups == _row_group_readers.size());
+ if (_total_groups == 0) {
+ *eof = true;
+ }
+ bool _batch_eof = false;
+ auto row_group_reader = _row_group_readers[_current_row_group_id];
+ RETURN_IF_ERROR(row_group_reader->next_batch(block, _batch_size, &_batch_eof));
+ if (_batch_eof) {
+ _current_row_group_id++;
+ if (_current_row_group_id > _total_groups) {
+ *eof = true;
+ }
Review Comment:
missing `else`?
In `else` block, you should open the next row group reader and call `next_batch()`
##########
be/src/vec/exec/format/parquet/vparquet_column_chunk_reader.cpp:
##########
@@ -43,11 +44,12 @@ Status ColumnChunkReader::init() {
// get the block compression codec
RETURN_IF_ERROR(get_block_compression_codec(_metadata.codec, _block_compress_codec));
+ VLOG_DEBUG << "initColumnChunkReader finish";
return Status::OK();
}
Status ColumnChunkReader::next_page() {
- RETURN_IF_ERROR(_page_reader->next_page());
+ RETURN_IF_ERROR(_page_reader->next_page_header());
_num_values = _page_reader->get_page_header()->data_page_header.num_values;
Review Comment:
Better rename `_num_values` to `_left_num_values`.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org