github-actions[bot] commented on code in PR #25138:
URL: https://github.com/apache/doris/pull/25138#discussion_r1355303333
##########
be/src/vec/exec/format/parquet/byte_array_dict_decoder.cpp:
##########
@@ -125,55 +125,35 @@ Status
ByteArrayDictDecoder::_decode_values(MutableColumnPtr& doris_column, Data
return _decode_dict_values<has_filter>(doris_column, select_vector,
is_dict_filter);
}
- TypeIndex logical_type = remove_nullable(data_type)->get_type_id();
- switch (logical_type) {
- case TypeIndex::String:
- [[fallthrough]];
- case TypeIndex::FixedString: {
- size_t dict_index = 0;
+ size_t dict_index = 0;
- ColumnSelectVector::DataReadType read_type;
- while (size_t run_length =
select_vector.get_next_run<has_filter>(&read_type)) {
- switch (read_type) {
- case ColumnSelectVector::CONTENT: {
- std::vector<StringRef> string_values;
- string_values.reserve(run_length);
- for (size_t i = 0; i < run_length; ++i) {
-
string_values.emplace_back(_dict_items[_indexes[dict_index++]]);
- }
- doris_column->insert_many_strings_overflow(&string_values[0],
run_length,
- _max_value_length);
- break;
- }
- case ColumnSelectVector::NULL_DATA: {
- doris_column->insert_many_defaults(run_length);
- break;
- }
- case ColumnSelectVector::FILTERED_CONTENT: {
- dict_index += run_length;
- break;
- }
- case ColumnSelectVector::FILTERED_NULL: {
- // do nothing
- break;
- }
+ ColumnSelectVector::DataReadType read_type;
Review Comment:
warning: variable 'read_type' is not initialized
[cppcoreguidelines-init-variables]
```cpp
ColumnSelectVector::DataReadType read_type;
^
```
##########
be/src/vec/exec/format/parquet/byte_array_dict_decoder.cpp:
##########
@@ -125,55 +125,35 @@ Status
ByteArrayDictDecoder::_decode_values(MutableColumnPtr& doris_column, Data
return _decode_dict_values<has_filter>(doris_column, select_vector,
is_dict_filter);
}
- TypeIndex logical_type = remove_nullable(data_type)->get_type_id();
- switch (logical_type) {
- case TypeIndex::String:
- [[fallthrough]];
- case TypeIndex::FixedString: {
- size_t dict_index = 0;
+ size_t dict_index = 0;
- ColumnSelectVector::DataReadType read_type;
- while (size_t run_length =
select_vector.get_next_run<has_filter>(&read_type)) {
- switch (read_type) {
- case ColumnSelectVector::CONTENT: {
- std::vector<StringRef> string_values;
- string_values.reserve(run_length);
- for (size_t i = 0; i < run_length; ++i) {
-
string_values.emplace_back(_dict_items[_indexes[dict_index++]]);
- }
- doris_column->insert_many_strings_overflow(&string_values[0],
run_length,
- _max_value_length);
- break;
- }
- case ColumnSelectVector::NULL_DATA: {
- doris_column->insert_many_defaults(run_length);
- break;
- }
- case ColumnSelectVector::FILTERED_CONTENT: {
- dict_index += run_length;
- break;
- }
- case ColumnSelectVector::FILTERED_NULL: {
- // do nothing
- break;
- }
+ ColumnSelectVector::DataReadType read_type;
+ while (size_t run_length =
select_vector.get_next_run<has_filter>(&read_type)) {
+ switch (read_type) {
+ case ColumnSelectVector::CONTENT: {
+ std::vector<StringRef> string_values;
Review Comment:
warning: variable 'string_values' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
std::vector<StringRef> string_values = 0;
```
##########
be/src/vec/exec/format/parquet/byte_array_plain_decoder.cpp:
##########
@@ -56,73 +56,53 @@ template <bool has_filter>
Status ByteArrayPlainDecoder::_decode_values(MutableColumnPtr& doris_column,
DataTypePtr& data_type,
ColumnSelectVector& select_vector,
bool is_dict_filter) {
- TypeIndex logical_type = remove_nullable(data_type)->get_type_id();
- switch (logical_type) {
- case TypeIndex::String:
- [[fallthrough]];
- case TypeIndex::FixedString: {
- ColumnSelectVector::DataReadType read_type;
- while (size_t run_length =
select_vector.get_next_run<has_filter>(&read_type)) {
- switch (read_type) {
- case ColumnSelectVector::CONTENT: {
- std::vector<StringRef> string_values;
- string_values.reserve(run_length);
- for (size_t i = 0; i < run_length; ++i) {
- if (UNLIKELY(_offset + 4 > _data->size)) {
- return Status::IOError("Can't read byte array length
from plain decoder");
- }
- uint32_t length = decode_fixed32_le(
- reinterpret_cast<const uint8_t*>(_data->data) +
_offset);
- _offset += 4;
- if (UNLIKELY(_offset + length) > _data->size) {
- return Status::IOError("Can't read enough bytes in
plain decoder");
- }
- string_values.emplace_back(_data->data + _offset, length);
- _offset += length;
+ ColumnSelectVector::DataReadType read_type;
Review Comment:
warning: variable 'read_type' is not initialized
[cppcoreguidelines-init-variables]
```cpp
ColumnSelectVector::DataReadType read_type;
^
```
##########
be/src/vec/exec/format/parquet/byte_array_plain_decoder.cpp:
##########
@@ -56,73 +56,53 @@ template <bool has_filter>
Status ByteArrayPlainDecoder::_decode_values(MutableColumnPtr& doris_column,
DataTypePtr& data_type,
ColumnSelectVector& select_vector,
bool is_dict_filter) {
- TypeIndex logical_type = remove_nullable(data_type)->get_type_id();
- switch (logical_type) {
- case TypeIndex::String:
- [[fallthrough]];
- case TypeIndex::FixedString: {
- ColumnSelectVector::DataReadType read_type;
- while (size_t run_length =
select_vector.get_next_run<has_filter>(&read_type)) {
- switch (read_type) {
- case ColumnSelectVector::CONTENT: {
- std::vector<StringRef> string_values;
- string_values.reserve(run_length);
- for (size_t i = 0; i < run_length; ++i) {
- if (UNLIKELY(_offset + 4 > _data->size)) {
- return Status::IOError("Can't read byte array length
from plain decoder");
- }
- uint32_t length = decode_fixed32_le(
- reinterpret_cast<const uint8_t*>(_data->data) +
_offset);
- _offset += 4;
- if (UNLIKELY(_offset + length) > _data->size) {
- return Status::IOError("Can't read enough bytes in
plain decoder");
- }
- string_values.emplace_back(_data->data + _offset, length);
- _offset += length;
+ ColumnSelectVector::DataReadType read_type;
+ while (size_t run_length =
select_vector.get_next_run<has_filter>(&read_type)) {
+ switch (read_type) {
+ case ColumnSelectVector::CONTENT: {
+ std::vector<StringRef> string_values;
Review Comment:
warning: variable 'string_values' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
std::vector<StringRef> string_values = 0;
```
##########
be/src/vec/exec/format/parquet/delta_bit_pack_decoder.h:
##########
@@ -361,3 +346,260 @@ class DeltaByteArrayDecoder : public DeltaDecoder {
std::vector<char> _buffered_data;
};
} // namespace doris::vectorized
+
+namespace doris::vectorized {
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_header() {
+ if (!_bit_reader->GetVlqInt(&_values_per_block) ||
+ !_bit_reader->GetVlqInt(&_mini_blocks_per_block) ||
+ !_bit_reader->GetVlqInt(&_total_value_count) ||
+ !_bit_reader->GetZigZagVlqInt(&_last_value)) {
+ return Status::IOError("Init header eof");
+ }
+ if (_values_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per block");
+ }
+ if (_values_per_block % 128 != 0) {
+ return Status::InvalidArgument(
+ "the number of values in a block must be multiple of 128, but
it's " +
+ std::to_string(_values_per_block));
+ }
+ if (_mini_blocks_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero miniblock per block");
+ }
+ _values_per_mini_block = _values_per_block / _mini_blocks_per_block;
+ if (_values_per_mini_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per miniblock");
+ }
+ if (_values_per_mini_block % 32 != 0) {
+ return Status::InvalidArgument(
+ "The number of values in a miniblock must be multiple of 32,
but it's " +
+ std::to_string(_values_per_mini_block));
+ }
+ _total_values_remaining = _total_value_count;
+ _delta_bit_widths.resize(_mini_blocks_per_block);
+ // init as empty property
+ _block_initialized = false;
+ _values_remaining_current_mini_block = 0;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_block() {
+ DCHECK_GT(_total_values_remaining, 0) << "InitBlock called at EOF";
+ if (!_bit_reader->GetZigZagVlqInt(&_min_delta)) {
+ return Status::IOError("Init block eof");
+ }
+
+ // read the bitwidth of each miniblock
+ uint8_t* bit_width_data = _delta_bit_widths.data();
Review Comment:
warning: variable 'bit_width_data' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
uint8_t* bit_width_data = nullptr = _delta_bit_widths.data();
```
##########
be/src/vec/exec/format/parquet/delta_bit_pack_decoder.h:
##########
@@ -361,3 +346,260 @@ class DeltaByteArrayDecoder : public DeltaDecoder {
std::vector<char> _buffered_data;
};
} // namespace doris::vectorized
+
+namespace doris::vectorized {
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_header() {
+ if (!_bit_reader->GetVlqInt(&_values_per_block) ||
+ !_bit_reader->GetVlqInt(&_mini_blocks_per_block) ||
+ !_bit_reader->GetVlqInt(&_total_value_count) ||
+ !_bit_reader->GetZigZagVlqInt(&_last_value)) {
+ return Status::IOError("Init header eof");
+ }
+ if (_values_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per block");
+ }
+ if (_values_per_block % 128 != 0) {
+ return Status::InvalidArgument(
+ "the number of values in a block must be multiple of 128, but
it's " +
+ std::to_string(_values_per_block));
+ }
+ if (_mini_blocks_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero miniblock per block");
+ }
+ _values_per_mini_block = _values_per_block / _mini_blocks_per_block;
+ if (_values_per_mini_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per miniblock");
+ }
+ if (_values_per_mini_block % 32 != 0) {
+ return Status::InvalidArgument(
+ "The number of values in a miniblock must be multiple of 32,
but it's " +
+ std::to_string(_values_per_mini_block));
+ }
+ _total_values_remaining = _total_value_count;
+ _delta_bit_widths.resize(_mini_blocks_per_block);
+ // init as empty property
+ _block_initialized = false;
+ _values_remaining_current_mini_block = 0;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_block() {
+ DCHECK_GT(_total_values_remaining, 0) << "InitBlock called at EOF";
+ if (!_bit_reader->GetZigZagVlqInt(&_min_delta)) {
+ return Status::IOError("Init block eof");
+ }
+
+ // read the bitwidth of each miniblock
+ uint8_t* bit_width_data = _delta_bit_widths.data();
+ for (uint32_t i = 0; i < _mini_blocks_per_block; ++i) {
+ if (!_bit_reader->GetAligned<uint8_t>(1, bit_width_data + i)) {
+ return Status::IOError("Decode bit-width EOF");
+ }
+ // Note that non-conformant bitwidth entries are allowed by the
Parquet spec
+ // for extraneous miniblocks in the last block (GH-14923), so we check
+ // the bitwidths when actually using them (see InitMiniBlock()).
+ }
+ _mini_block_idx = 0;
+ _block_initialized = true;
+ RETURN_IF_ERROR(_init_mini_block(bit_width_data[0]));
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_mini_block(int bit_width) {
+ if (PREDICT_FALSE(bit_width > kMaxDeltaBitWidth)) {
+ return Status::InvalidArgument("delta bit width larger than integer
bit width");
+ }
+ _delta_bit_width = bit_width;
+ _values_remaining_current_mini_block = _values_per_mini_block;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_get_internal(T* buffer, int
num_values,
+ int*
out_num_values) {
+ num_values = static_cast<int>(std::min<int64_t>(num_values,
_total_values_remaining));
+ if (num_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+ int i = 0;
+ while (i < num_values) {
+ if (PREDICT_FALSE(_values_remaining_current_mini_block == 0)) {
+ if (PREDICT_FALSE(!_block_initialized)) {
+ buffer[i++] = _last_value;
+ DCHECK_EQ(i, 1); // we're at the beginning of the page
+ if (i == num_values) {
+ // When block is uninitialized and i reaches num_values we
have two
+ // different possibilities:
+ // 1. _total_value_count == 1, which means that the page
may have only
+ // one value (encoded in the header), and we should not
initialize
+ // any block.
+ // 2. _total_value_count != 1, which means we should
initialize the
+ // incoming block for subsequent reads.
+ if (_total_value_count != 1) {
+ RETURN_IF_ERROR(_init_block());
+ }
+ break;
+ }
+ RETURN_IF_ERROR(_init_block());
+ } else {
+ ++_mini_block_idx;
+ if (_mini_block_idx < _mini_blocks_per_block) {
+
RETURN_IF_ERROR(_init_mini_block(_delta_bit_widths.data()[_mini_block_idx]));
+ } else {
+ RETURN_IF_ERROR(_init_block());
+ }
+ }
+ }
+
+ int values_decode = std::min(_values_remaining_current_mini_block,
Review Comment:
warning: variable 'values_decode' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
int values_decode = 0 =
std::min(_values_remaining_current_mini_block,
```
##########
be/src/vec/exec/format/parquet/delta_bit_pack_decoder.h:
##########
@@ -361,3 +346,260 @@ class DeltaByteArrayDecoder : public DeltaDecoder {
std::vector<char> _buffered_data;
};
} // namespace doris::vectorized
+
+namespace doris::vectorized {
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_header() {
+ if (!_bit_reader->GetVlqInt(&_values_per_block) ||
+ !_bit_reader->GetVlqInt(&_mini_blocks_per_block) ||
+ !_bit_reader->GetVlqInt(&_total_value_count) ||
+ !_bit_reader->GetZigZagVlqInt(&_last_value)) {
+ return Status::IOError("Init header eof");
+ }
+ if (_values_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per block");
+ }
+ if (_values_per_block % 128 != 0) {
+ return Status::InvalidArgument(
+ "the number of values in a block must be multiple of 128, but
it's " +
+ std::to_string(_values_per_block));
+ }
+ if (_mini_blocks_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero miniblock per block");
+ }
+ _values_per_mini_block = _values_per_block / _mini_blocks_per_block;
+ if (_values_per_mini_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per miniblock");
+ }
+ if (_values_per_mini_block % 32 != 0) {
+ return Status::InvalidArgument(
+ "The number of values in a miniblock must be multiple of 32,
but it's " +
+ std::to_string(_values_per_mini_block));
+ }
+ _total_values_remaining = _total_value_count;
+ _delta_bit_widths.resize(_mini_blocks_per_block);
+ // init as empty property
+ _block_initialized = false;
+ _values_remaining_current_mini_block = 0;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_block() {
+ DCHECK_GT(_total_values_remaining, 0) << "InitBlock called at EOF";
+ if (!_bit_reader->GetZigZagVlqInt(&_min_delta)) {
+ return Status::IOError("Init block eof");
+ }
+
+ // read the bitwidth of each miniblock
+ uint8_t* bit_width_data = _delta_bit_widths.data();
+ for (uint32_t i = 0; i < _mini_blocks_per_block; ++i) {
+ if (!_bit_reader->GetAligned<uint8_t>(1, bit_width_data + i)) {
+ return Status::IOError("Decode bit-width EOF");
+ }
+ // Note that non-conformant bitwidth entries are allowed by the
Parquet spec
+ // for extraneous miniblocks in the last block (GH-14923), so we check
+ // the bitwidths when actually using them (see InitMiniBlock()).
+ }
+ _mini_block_idx = 0;
+ _block_initialized = true;
+ RETURN_IF_ERROR(_init_mini_block(bit_width_data[0]));
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_mini_block(int bit_width) {
+ if (PREDICT_FALSE(bit_width > kMaxDeltaBitWidth)) {
+ return Status::InvalidArgument("delta bit width larger than integer
bit width");
+ }
+ _delta_bit_width = bit_width;
+ _values_remaining_current_mini_block = _values_per_mini_block;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_get_internal(T* buffer, int
num_values,
+ int*
out_num_values) {
+ num_values = static_cast<int>(std::min<int64_t>(num_values,
_total_values_remaining));
+ if (num_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+ int i = 0;
+ while (i < num_values) {
+ if (PREDICT_FALSE(_values_remaining_current_mini_block == 0)) {
+ if (PREDICT_FALSE(!_block_initialized)) {
+ buffer[i++] = _last_value;
+ DCHECK_EQ(i, 1); // we're at the beginning of the page
+ if (i == num_values) {
+ // When block is uninitialized and i reaches num_values we
have two
+ // different possibilities:
+ // 1. _total_value_count == 1, which means that the page
may have only
+ // one value (encoded in the header), and we should not
initialize
+ // any block.
+ // 2. _total_value_count != 1, which means we should
initialize the
+ // incoming block for subsequent reads.
+ if (_total_value_count != 1) {
+ RETURN_IF_ERROR(_init_block());
+ }
+ break;
+ }
+ RETURN_IF_ERROR(_init_block());
+ } else {
+ ++_mini_block_idx;
+ if (_mini_block_idx < _mini_blocks_per_block) {
+
RETURN_IF_ERROR(_init_mini_block(_delta_bit_widths.data()[_mini_block_idx]));
+ } else {
+ RETURN_IF_ERROR(_init_block());
+ }
+ }
+ }
+
+ int values_decode = std::min(_values_remaining_current_mini_block,
+ static_cast<uint32_t>(num_values - i));
+ for (int j = 0; j < values_decode; ++j) {
+ if (!_bit_reader->GetValue(_delta_bit_width, buffer + i + j)) {
+ return Status::IOError("Get batch EOF");
+ }
+ }
+ for (int j = 0; j < values_decode; ++j) {
+ // Addition between min_delta, packed int and last_value should be
treated as
+ // unsigned addition. Overflow is as expected.
+ buffer[i + j] = static_cast<UT>(_min_delta) +
static_cast<UT>(buffer[i + j]) +
+ static_cast<UT>(_last_value);
+ _last_value = buffer[i + j];
+ }
+ _values_remaining_current_mini_block -= values_decode;
+ i += values_decode;
+ }
+ _total_values_remaining -= num_values;
+
+ if (PREDICT_FALSE(_total_values_remaining == 0)) {
+ if (!_bit_reader->Advance(_delta_bit_width *
_values_remaining_current_mini_block)) {
+ return Status::IOError("Skip padding EOF");
+ }
+ _values_remaining_current_mini_block = 0;
+ }
+ *out_num_values = num_values;
+ return Status::OK();
+}
+template <tparquet::Type::type PhysicalType>
+void DeltaLengthByteArrayDecoder<PhysicalType>::_decode_lengths() {
+ _len_decoder.set_bit_reader(_bit_reader);
+ // get the number of encoded lengths
+ int num_length = _len_decoder.valid_values_count();
+ _buffered_length.resize(num_length);
+
+ // decode all the lengths. all the lengths are buffered in
buffered_length_.
+ int ret;
+ Status st = _len_decoder.decode(_buffered_length.data(), num_length, &ret);
+ if (!st.ok()) {
+ LOG(FATAL) << "Fail to decode delta length, status: " << st;
+ }
+ DCHECK_EQ(ret, num_length);
+ _length_idx = 0;
+ _num_valid_values = num_length;
+}
+template <tparquet::Type::type PhysicalType>
+Status DeltaLengthByteArrayDecoder<PhysicalType>::_get_internal(Slice* buffer,
int max_values,
+ int*
out_num_values) {
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+
+ int32_t data_size = 0;
+ const int32_t* length_ptr = _buffered_length.data() + _length_idx;
+ for (int i = 0; i < max_values; ++i) {
+ int32_t len = length_ptr[i];
+ if (PREDICT_FALSE(len < 0)) {
+ return Status::InvalidArgument("Negative string delta length");
+ }
+ buffer[i].size = len;
+ if (common::add_overflow(data_size, len, data_size)) {
+ return Status::InvalidArgument("Excess expansion in
DELTA_(LENGTH_)BYTE_ARRAY");
+ }
+ }
+ _length_idx += max_values;
+
+ _buffered_data.resize(data_size);
+ char* data_ptr = _buffered_data.data();
Review Comment:
warning: variable 'data_ptr' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
char* data_ptr = nullptr = _buffered_data.data();
```
##########
be/src/vec/exec/format/parquet/delta_bit_pack_decoder.h:
##########
@@ -361,3 +346,260 @@ class DeltaByteArrayDecoder : public DeltaDecoder {
std::vector<char> _buffered_data;
};
} // namespace doris::vectorized
+
+namespace doris::vectorized {
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_header() {
+ if (!_bit_reader->GetVlqInt(&_values_per_block) ||
+ !_bit_reader->GetVlqInt(&_mini_blocks_per_block) ||
+ !_bit_reader->GetVlqInt(&_total_value_count) ||
+ !_bit_reader->GetZigZagVlqInt(&_last_value)) {
+ return Status::IOError("Init header eof");
+ }
+ if (_values_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per block");
+ }
+ if (_values_per_block % 128 != 0) {
+ return Status::InvalidArgument(
+ "the number of values in a block must be multiple of 128, but
it's " +
+ std::to_string(_values_per_block));
+ }
+ if (_mini_blocks_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero miniblock per block");
+ }
+ _values_per_mini_block = _values_per_block / _mini_blocks_per_block;
+ if (_values_per_mini_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per miniblock");
+ }
+ if (_values_per_mini_block % 32 != 0) {
+ return Status::InvalidArgument(
+ "The number of values in a miniblock must be multiple of 32,
but it's " +
+ std::to_string(_values_per_mini_block));
+ }
+ _total_values_remaining = _total_value_count;
+ _delta_bit_widths.resize(_mini_blocks_per_block);
+ // init as empty property
+ _block_initialized = false;
+ _values_remaining_current_mini_block = 0;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_block() {
+ DCHECK_GT(_total_values_remaining, 0) << "InitBlock called at EOF";
+ if (!_bit_reader->GetZigZagVlqInt(&_min_delta)) {
+ return Status::IOError("Init block eof");
+ }
+
+ // read the bitwidth of each miniblock
+ uint8_t* bit_width_data = _delta_bit_widths.data();
+ for (uint32_t i = 0; i < _mini_blocks_per_block; ++i) {
+ if (!_bit_reader->GetAligned<uint8_t>(1, bit_width_data + i)) {
+ return Status::IOError("Decode bit-width EOF");
+ }
+ // Note that non-conformant bitwidth entries are allowed by the
Parquet spec
+ // for extraneous miniblocks in the last block (GH-14923), so we check
+ // the bitwidths when actually using them (see InitMiniBlock()).
+ }
+ _mini_block_idx = 0;
+ _block_initialized = true;
+ RETURN_IF_ERROR(_init_mini_block(bit_width_data[0]));
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_mini_block(int bit_width) {
+ if (PREDICT_FALSE(bit_width > kMaxDeltaBitWidth)) {
+ return Status::InvalidArgument("delta bit width larger than integer
bit width");
+ }
+ _delta_bit_width = bit_width;
+ _values_remaining_current_mini_block = _values_per_mini_block;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_get_internal(T* buffer, int
num_values,
+ int*
out_num_values) {
+ num_values = static_cast<int>(std::min<int64_t>(num_values,
_total_values_remaining));
+ if (num_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+ int i = 0;
+ while (i < num_values) {
+ if (PREDICT_FALSE(_values_remaining_current_mini_block == 0)) {
+ if (PREDICT_FALSE(!_block_initialized)) {
+ buffer[i++] = _last_value;
+ DCHECK_EQ(i, 1); // we're at the beginning of the page
+ if (i == num_values) {
+ // When block is uninitialized and i reaches num_values we
have two
+ // different possibilities:
+ // 1. _total_value_count == 1, which means that the page
may have only
+ // one value (encoded in the header), and we should not
initialize
+ // any block.
+ // 2. _total_value_count != 1, which means we should
initialize the
+ // incoming block for subsequent reads.
+ if (_total_value_count != 1) {
+ RETURN_IF_ERROR(_init_block());
+ }
+ break;
+ }
+ RETURN_IF_ERROR(_init_block());
+ } else {
+ ++_mini_block_idx;
+ if (_mini_block_idx < _mini_blocks_per_block) {
+
RETURN_IF_ERROR(_init_mini_block(_delta_bit_widths.data()[_mini_block_idx]));
+ } else {
+ RETURN_IF_ERROR(_init_block());
+ }
+ }
+ }
+
+ int values_decode = std::min(_values_remaining_current_mini_block,
+ static_cast<uint32_t>(num_values - i));
+ for (int j = 0; j < values_decode; ++j) {
+ if (!_bit_reader->GetValue(_delta_bit_width, buffer + i + j)) {
+ return Status::IOError("Get batch EOF");
+ }
+ }
+ for (int j = 0; j < values_decode; ++j) {
+ // Addition between min_delta, packed int and last_value should be
treated as
+ // unsigned addition. Overflow is as expected.
+ buffer[i + j] = static_cast<UT>(_min_delta) +
static_cast<UT>(buffer[i + j]) +
+ static_cast<UT>(_last_value);
+ _last_value = buffer[i + j];
+ }
+ _values_remaining_current_mini_block -= values_decode;
+ i += values_decode;
+ }
+ _total_values_remaining -= num_values;
+
+ if (PREDICT_FALSE(_total_values_remaining == 0)) {
+ if (!_bit_reader->Advance(_delta_bit_width *
_values_remaining_current_mini_block)) {
+ return Status::IOError("Skip padding EOF");
+ }
+ _values_remaining_current_mini_block = 0;
+ }
+ *out_num_values = num_values;
+ return Status::OK();
+}
+template <tparquet::Type::type PhysicalType>
+void DeltaLengthByteArrayDecoder<PhysicalType>::_decode_lengths() {
+ _len_decoder.set_bit_reader(_bit_reader);
+ // get the number of encoded lengths
+ int num_length = _len_decoder.valid_values_count();
+ _buffered_length.resize(num_length);
+
+ // decode all the lengths. all the lengths are buffered in
buffered_length_.
+ int ret;
+ Status st = _len_decoder.decode(_buffered_length.data(), num_length, &ret);
+ if (!st.ok()) {
+ LOG(FATAL) << "Fail to decode delta length, status: " << st;
+ }
+ DCHECK_EQ(ret, num_length);
+ _length_idx = 0;
+ _num_valid_values = num_length;
+}
+template <tparquet::Type::type PhysicalType>
+Status DeltaLengthByteArrayDecoder<PhysicalType>::_get_internal(Slice* buffer,
int max_values,
+ int*
out_num_values) {
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+
+ int32_t data_size = 0;
+ const int32_t* length_ptr = _buffered_length.data() + _length_idx;
Review Comment:
warning: variable 'length_ptr' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
const int32_t* length_ptr = nullptr = _buffered_length.data() +
_length_idx;
```
##########
be/src/vec/exec/format/parquet/delta_bit_pack_decoder.h:
##########
@@ -361,3 +346,260 @@ class DeltaByteArrayDecoder : public DeltaDecoder {
std::vector<char> _buffered_data;
};
} // namespace doris::vectorized
+
+namespace doris::vectorized {
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_header() {
+ if (!_bit_reader->GetVlqInt(&_values_per_block) ||
+ !_bit_reader->GetVlqInt(&_mini_blocks_per_block) ||
+ !_bit_reader->GetVlqInt(&_total_value_count) ||
+ !_bit_reader->GetZigZagVlqInt(&_last_value)) {
+ return Status::IOError("Init header eof");
+ }
+ if (_values_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per block");
+ }
+ if (_values_per_block % 128 != 0) {
+ return Status::InvalidArgument(
+ "the number of values in a block must be multiple of 128, but
it's " +
+ std::to_string(_values_per_block));
+ }
+ if (_mini_blocks_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero miniblock per block");
+ }
+ _values_per_mini_block = _values_per_block / _mini_blocks_per_block;
+ if (_values_per_mini_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per miniblock");
+ }
+ if (_values_per_mini_block % 32 != 0) {
+ return Status::InvalidArgument(
+ "The number of values in a miniblock must be multiple of 32,
but it's " +
+ std::to_string(_values_per_mini_block));
+ }
+ _total_values_remaining = _total_value_count;
+ _delta_bit_widths.resize(_mini_blocks_per_block);
+ // init as empty property
+ _block_initialized = false;
+ _values_remaining_current_mini_block = 0;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_block() {
+ DCHECK_GT(_total_values_remaining, 0) << "InitBlock called at EOF";
+ if (!_bit_reader->GetZigZagVlqInt(&_min_delta)) {
+ return Status::IOError("Init block eof");
+ }
+
+ // read the bitwidth of each miniblock
+ uint8_t* bit_width_data = _delta_bit_widths.data();
+ for (uint32_t i = 0; i < _mini_blocks_per_block; ++i) {
+ if (!_bit_reader->GetAligned<uint8_t>(1, bit_width_data + i)) {
+ return Status::IOError("Decode bit-width EOF");
+ }
+ // Note that non-conformant bitwidth entries are allowed by the
Parquet spec
+ // for extraneous miniblocks in the last block (GH-14923), so we check
+ // the bitwidths when actually using them (see InitMiniBlock()).
+ }
+ _mini_block_idx = 0;
+ _block_initialized = true;
+ RETURN_IF_ERROR(_init_mini_block(bit_width_data[0]));
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_mini_block(int bit_width) {
+ if (PREDICT_FALSE(bit_width > kMaxDeltaBitWidth)) {
+ return Status::InvalidArgument("delta bit width larger than integer
bit width");
+ }
+ _delta_bit_width = bit_width;
+ _values_remaining_current_mini_block = _values_per_mini_block;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_get_internal(T* buffer, int
num_values,
+ int*
out_num_values) {
+ num_values = static_cast<int>(std::min<int64_t>(num_values,
_total_values_remaining));
+ if (num_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+ int i = 0;
+ while (i < num_values) {
+ if (PREDICT_FALSE(_values_remaining_current_mini_block == 0)) {
+ if (PREDICT_FALSE(!_block_initialized)) {
+ buffer[i++] = _last_value;
+ DCHECK_EQ(i, 1); // we're at the beginning of the page
+ if (i == num_values) {
+ // When block is uninitialized and i reaches num_values we
have two
+ // different possibilities:
+ // 1. _total_value_count == 1, which means that the page
may have only
+ // one value (encoded in the header), and we should not
initialize
+ // any block.
+ // 2. _total_value_count != 1, which means we should
initialize the
+ // incoming block for subsequent reads.
+ if (_total_value_count != 1) {
+ RETURN_IF_ERROR(_init_block());
+ }
+ break;
+ }
+ RETURN_IF_ERROR(_init_block());
+ } else {
+ ++_mini_block_idx;
+ if (_mini_block_idx < _mini_blocks_per_block) {
+
RETURN_IF_ERROR(_init_mini_block(_delta_bit_widths.data()[_mini_block_idx]));
+ } else {
+ RETURN_IF_ERROR(_init_block());
+ }
+ }
+ }
+
+ int values_decode = std::min(_values_remaining_current_mini_block,
+ static_cast<uint32_t>(num_values - i));
+ for (int j = 0; j < values_decode; ++j) {
+ if (!_bit_reader->GetValue(_delta_bit_width, buffer + i + j)) {
+ return Status::IOError("Get batch EOF");
+ }
+ }
+ for (int j = 0; j < values_decode; ++j) {
+ // Addition between min_delta, packed int and last_value should be
treated as
+ // unsigned addition. Overflow is as expected.
+ buffer[i + j] = static_cast<UT>(_min_delta) +
static_cast<UT>(buffer[i + j]) +
+ static_cast<UT>(_last_value);
+ _last_value = buffer[i + j];
+ }
+ _values_remaining_current_mini_block -= values_decode;
+ i += values_decode;
+ }
+ _total_values_remaining -= num_values;
+
+ if (PREDICT_FALSE(_total_values_remaining == 0)) {
+ if (!_bit_reader->Advance(_delta_bit_width *
_values_remaining_current_mini_block)) {
+ return Status::IOError("Skip padding EOF");
+ }
+ _values_remaining_current_mini_block = 0;
+ }
+ *out_num_values = num_values;
+ return Status::OK();
+}
+template <tparquet::Type::type PhysicalType>
+void DeltaLengthByteArrayDecoder<PhysicalType>::_decode_lengths() {
+ _len_decoder.set_bit_reader(_bit_reader);
+ // get the number of encoded lengths
+ int num_length = _len_decoder.valid_values_count();
+ _buffered_length.resize(num_length);
+
+ // decode all the lengths. all the lengths are buffered in
buffered_length_.
+ int ret;
Review Comment:
warning: variable 'ret' is not initialized [cppcoreguidelines-init-variables]
```suggestion
int ret = 0;
```
##########
be/src/vec/exec/format/parquet/parquet_column_convert.h:
##########
@@ -0,0 +1,548 @@
+// 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 <gen_cpp/PlanNodes_types.h>
Review Comment:
warning: 'gen_cpp/PlanNodes_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/PlanNodes_types.h>
^
```
##########
be/src/vec/exec/format/parquet/fix_length_plain_decoder.h:
##########
@@ -32,44 +33,144 @@ class ColumnSelectVector;
namespace doris::vectorized {
+template <tparquet::Type::type PhysicalType>
class FixLengthPlainDecoder final : public Decoder {
public:
- FixLengthPlainDecoder(tparquet::Type::type physical_type) :
_physical_type(physical_type) {};
+ FixLengthPlainDecoder() {};
~FixLengthPlainDecoder() override = default;
Status decode_values(MutableColumnPtr& doris_column, DataTypePtr&
data_type,
ColumnSelectVector& select_vector, bool
is_dict_filter) override;
- template <bool hasFilter>
+ template <bool has_filter>
Status _decode_values(MutableColumnPtr& doris_column, DataTypePtr&
data_type,
ColumnSelectVector& select_vector, bool
is_dict_filter);
Status skip_values(size_t num_values) override;
protected:
- template <typename Numeric, typename PhysicalType, bool has_filter>
+ template <bool has_filter>
Status _decode_numeric(MutableColumnPtr& doris_column, ColumnSelectVector&
select_vector);
- template <typename CppType, typename ColumnType, bool has_filter>
- Status _decode_date(MutableColumnPtr& doris_column, ColumnSelectVector&
select_vector);
-
- template <typename CppType, typename ColumnType, bool has_filter>
- Status _decode_datetime64(MutableColumnPtr& doris_column,
ColumnSelectVector& select_vector);
-
- template <typename CppType, typename ColumnType, bool has_filter>
- Status _decode_datetime96(MutableColumnPtr& doris_column,
ColumnSelectVector& select_vector);
-
- template <typename DecimalPrimitiveType, bool has_filter>
- Status _decode_binary_decimal(MutableColumnPtr& doris_column, DataTypePtr&
data_type,
- ColumnSelectVector& select_vector);
-
- template <typename DecimalPrimitiveType, typename DecimalPhysicalType,
bool has_filter>
- Status _decode_primitive_decimal(MutableColumnPtr& doris_column,
DataTypePtr& data_type,
- ColumnSelectVector& select_vector);
-
template <bool has_filter>
Status _decode_string(MutableColumnPtr& doris_column, ColumnSelectVector&
select_vector);
-
- tparquet::Type::type _physical_type;
};
+
+template <tparquet::Type::type PhysicalType>
+Status FixLengthPlainDecoder<PhysicalType>::skip_values(size_t num_values) {
+ _offset += _type_length * num_values;
+ if (UNLIKELY(_offset > _data->size)) {
+ return Status::IOError("Out-of-bounds access in parquet data decoder");
+ }
+ return Status::OK();
+}
+
+template <tparquet::Type::type PhysicalType>
+Status FixLengthPlainDecoder<PhysicalType>::decode_values(MutableColumnPtr&
doris_column,
+ DataTypePtr&
data_type,
+ ColumnSelectVector&
select_vector,
+ bool is_dict_filter)
{
+ if (select_vector.has_filter()) {
+ return _decode_values<true>(doris_column, data_type, select_vector,
is_dict_filter);
+ } else {
+ return _decode_values<false>(doris_column, data_type, select_vector,
is_dict_filter);
+ }
+}
+
+template <tparquet::Type::type PhysicalType>
+template <bool has_filter>
+Status FixLengthPlainDecoder<PhysicalType>::_decode_values(MutableColumnPtr&
doris_column,
+ DataTypePtr&
data_type,
+ ColumnSelectVector&
select_vector,
+ bool
is_dict_filter) {
+ size_t non_null_size = select_vector.num_values() -
select_vector.num_nulls();
+ if (UNLIKELY(_offset + _type_length * non_null_size > _data->size)) {
+ return Status::IOError("Out-of-bounds access in parquet data decoder");
+ }
+
+ if constexpr (PhysicalType == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
+ return _decode_string<has_filter>(doris_column, select_vector);
+ } else {
+ return _decode_numeric<has_filter>(doris_column, select_vector);
+ }
+}
+
+template <tparquet::Type::type PhysicalType>
+template <bool has_filter>
+Status FixLengthPlainDecoder<PhysicalType>::_decode_string(MutableColumnPtr&
doris_column,
+ ColumnSelectVector&
select_vector) {
+ ColumnSelectVector::DataReadType read_type;
Review Comment:
warning: variable 'read_type' is not initialized
[cppcoreguidelines-init-variables]
```cpp
ColumnSelectVector::DataReadType read_type;
^
```
##########
be/src/vec/exec/format/parquet/fix_length_plain_decoder.h:
##########
@@ -32,44 +33,144 @@ class ColumnSelectVector;
namespace doris::vectorized {
+template <tparquet::Type::type PhysicalType>
class FixLengthPlainDecoder final : public Decoder {
public:
- FixLengthPlainDecoder(tparquet::Type::type physical_type) :
_physical_type(physical_type) {};
+ FixLengthPlainDecoder() {};
~FixLengthPlainDecoder() override = default;
Status decode_values(MutableColumnPtr& doris_column, DataTypePtr&
data_type,
ColumnSelectVector& select_vector, bool
is_dict_filter) override;
- template <bool hasFilter>
+ template <bool has_filter>
Status _decode_values(MutableColumnPtr& doris_column, DataTypePtr&
data_type,
ColumnSelectVector& select_vector, bool
is_dict_filter);
Status skip_values(size_t num_values) override;
protected:
- template <typename Numeric, typename PhysicalType, bool has_filter>
+ template <bool has_filter>
Status _decode_numeric(MutableColumnPtr& doris_column, ColumnSelectVector&
select_vector);
- template <typename CppType, typename ColumnType, bool has_filter>
- Status _decode_date(MutableColumnPtr& doris_column, ColumnSelectVector&
select_vector);
-
- template <typename CppType, typename ColumnType, bool has_filter>
- Status _decode_datetime64(MutableColumnPtr& doris_column,
ColumnSelectVector& select_vector);
-
- template <typename CppType, typename ColumnType, bool has_filter>
- Status _decode_datetime96(MutableColumnPtr& doris_column,
ColumnSelectVector& select_vector);
-
- template <typename DecimalPrimitiveType, bool has_filter>
- Status _decode_binary_decimal(MutableColumnPtr& doris_column, DataTypePtr&
data_type,
- ColumnSelectVector& select_vector);
-
- template <typename DecimalPrimitiveType, typename DecimalPhysicalType,
bool has_filter>
- Status _decode_primitive_decimal(MutableColumnPtr& doris_column,
DataTypePtr& data_type,
- ColumnSelectVector& select_vector);
-
template <bool has_filter>
Status _decode_string(MutableColumnPtr& doris_column, ColumnSelectVector&
select_vector);
-
- tparquet::Type::type _physical_type;
};
+
+template <tparquet::Type::type PhysicalType>
+Status FixLengthPlainDecoder<PhysicalType>::skip_values(size_t num_values) {
+ _offset += _type_length * num_values;
+ if (UNLIKELY(_offset > _data->size)) {
+ return Status::IOError("Out-of-bounds access in parquet data decoder");
+ }
+ return Status::OK();
+}
+
+template <tparquet::Type::type PhysicalType>
+Status FixLengthPlainDecoder<PhysicalType>::decode_values(MutableColumnPtr&
doris_column,
+ DataTypePtr&
data_type,
+ ColumnSelectVector&
select_vector,
+ bool is_dict_filter)
{
+ if (select_vector.has_filter()) {
+ return _decode_values<true>(doris_column, data_type, select_vector,
is_dict_filter);
+ } else {
+ return _decode_values<false>(doris_column, data_type, select_vector,
is_dict_filter);
+ }
+}
+
+template <tparquet::Type::type PhysicalType>
+template <bool has_filter>
+Status FixLengthPlainDecoder<PhysicalType>::_decode_values(MutableColumnPtr&
doris_column,
+ DataTypePtr&
data_type,
+ ColumnSelectVector&
select_vector,
+ bool
is_dict_filter) {
+ size_t non_null_size = select_vector.num_values() -
select_vector.num_nulls();
+ if (UNLIKELY(_offset + _type_length * non_null_size > _data->size)) {
+ return Status::IOError("Out-of-bounds access in parquet data decoder");
+ }
+
+ if constexpr (PhysicalType == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
+ return _decode_string<has_filter>(doris_column, select_vector);
+ } else {
+ return _decode_numeric<has_filter>(doris_column, select_vector);
+ }
+}
+
+template <tparquet::Type::type PhysicalType>
+template <bool has_filter>
+Status FixLengthPlainDecoder<PhysicalType>::_decode_string(MutableColumnPtr&
doris_column,
+ ColumnSelectVector&
select_vector) {
+ ColumnSelectVector::DataReadType read_type;
+ while (size_t run_length =
select_vector.get_next_run<has_filter>(&read_type)) {
+ switch (read_type) {
+ case ColumnSelectVector::CONTENT: {
+ std::vector<StringRef> string_values;
Review Comment:
warning: variable 'string_values' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
std::vector<StringRef> string_values = 0;
```
##########
be/src/vec/exec/format/parquet/fix_length_plain_decoder.h:
##########
@@ -32,44 +33,144 @@ class ColumnSelectVector;
namespace doris::vectorized {
+template <tparquet::Type::type PhysicalType>
class FixLengthPlainDecoder final : public Decoder {
public:
- FixLengthPlainDecoder(tparquet::Type::type physical_type) :
_physical_type(physical_type) {};
+ FixLengthPlainDecoder() {};
~FixLengthPlainDecoder() override = default;
Status decode_values(MutableColumnPtr& doris_column, DataTypePtr&
data_type,
ColumnSelectVector& select_vector, bool
is_dict_filter) override;
- template <bool hasFilter>
+ template <bool has_filter>
Status _decode_values(MutableColumnPtr& doris_column, DataTypePtr&
data_type,
ColumnSelectVector& select_vector, bool
is_dict_filter);
Status skip_values(size_t num_values) override;
protected:
- template <typename Numeric, typename PhysicalType, bool has_filter>
+ template <bool has_filter>
Status _decode_numeric(MutableColumnPtr& doris_column, ColumnSelectVector&
select_vector);
- template <typename CppType, typename ColumnType, bool has_filter>
- Status _decode_date(MutableColumnPtr& doris_column, ColumnSelectVector&
select_vector);
-
- template <typename CppType, typename ColumnType, bool has_filter>
- Status _decode_datetime64(MutableColumnPtr& doris_column,
ColumnSelectVector& select_vector);
-
- template <typename CppType, typename ColumnType, bool has_filter>
- Status _decode_datetime96(MutableColumnPtr& doris_column,
ColumnSelectVector& select_vector);
-
- template <typename DecimalPrimitiveType, bool has_filter>
- Status _decode_binary_decimal(MutableColumnPtr& doris_column, DataTypePtr&
data_type,
- ColumnSelectVector& select_vector);
-
- template <typename DecimalPrimitiveType, typename DecimalPhysicalType,
bool has_filter>
- Status _decode_primitive_decimal(MutableColumnPtr& doris_column,
DataTypePtr& data_type,
- ColumnSelectVector& select_vector);
-
template <bool has_filter>
Status _decode_string(MutableColumnPtr& doris_column, ColumnSelectVector&
select_vector);
-
- tparquet::Type::type _physical_type;
};
+
+template <tparquet::Type::type PhysicalType>
+Status FixLengthPlainDecoder<PhysicalType>::skip_values(size_t num_values) {
+ _offset += _type_length * num_values;
+ if (UNLIKELY(_offset > _data->size)) {
+ return Status::IOError("Out-of-bounds access in parquet data decoder");
+ }
+ return Status::OK();
+}
+
+template <tparquet::Type::type PhysicalType>
+Status FixLengthPlainDecoder<PhysicalType>::decode_values(MutableColumnPtr&
doris_column,
+ DataTypePtr&
data_type,
+ ColumnSelectVector&
select_vector,
+ bool is_dict_filter)
{
+ if (select_vector.has_filter()) {
+ return _decode_values<true>(doris_column, data_type, select_vector,
is_dict_filter);
+ } else {
+ return _decode_values<false>(doris_column, data_type, select_vector,
is_dict_filter);
+ }
+}
+
+template <tparquet::Type::type PhysicalType>
+template <bool has_filter>
+Status FixLengthPlainDecoder<PhysicalType>::_decode_values(MutableColumnPtr&
doris_column,
+ DataTypePtr&
data_type,
+ ColumnSelectVector&
select_vector,
+ bool
is_dict_filter) {
+ size_t non_null_size = select_vector.num_values() -
select_vector.num_nulls();
+ if (UNLIKELY(_offset + _type_length * non_null_size > _data->size)) {
+ return Status::IOError("Out-of-bounds access in parquet data decoder");
+ }
+
+ if constexpr (PhysicalType == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
+ return _decode_string<has_filter>(doris_column, select_vector);
+ } else {
+ return _decode_numeric<has_filter>(doris_column, select_vector);
+ }
+}
+
+template <tparquet::Type::type PhysicalType>
+template <bool has_filter>
+Status FixLengthPlainDecoder<PhysicalType>::_decode_string(MutableColumnPtr&
doris_column,
+ ColumnSelectVector&
select_vector) {
+ ColumnSelectVector::DataReadType read_type;
+ while (size_t run_length =
select_vector.get_next_run<has_filter>(&read_type)) {
+ switch (read_type) {
+ case ColumnSelectVector::CONTENT: {
+ std::vector<StringRef> string_values;
+ string_values.reserve(run_length);
+ for (size_t i = 0; i < run_length; ++i) {
+ char* buf_start = _data->data + _offset;
+ string_values.emplace_back(buf_start, _type_length);
+ _offset += _type_length;
+ }
+ doris_column->insert_many_strings(&string_values[0], run_length);
+ break;
+ }
+ case ColumnSelectVector::NULL_DATA: {
+ doris_column->insert_many_defaults(run_length);
+ break;
+ }
+ case ColumnSelectVector::FILTERED_CONTENT: {
+ _offset += _type_length * run_length;
+ break;
+ }
+ case ColumnSelectVector::FILTERED_NULL: {
+ // do nothing
+ break;
+ }
+ }
+ }
+ return Status::OK();
+}
+
+template <tparquet::Type::type PhysicalType>
+template <bool has_filter>
+Status FixLengthPlainDecoder<PhysicalType>::_decode_numeric(MutableColumnPtr&
doris_column,
+
ColumnSelectVector& select_vector) {
+ if constexpr (PhysicalType == tparquet::Type::FIXED_LEN_BYTE_ARRAY ||
+ PhysicalType == tparquet::Type::BYTE_ARRAY) {
+ return Status::OK();
+ } else {
+ using ColumnType =
ParquetConvert::PhysicalTypeTraits<PhysicalType>::ColumnType;
+ using DataType =
ParquetConvert::PhysicalTypeTraits<PhysicalType>::DataType;
+
+ auto& column_data = static_cast<ColumnType&>(*doris_column).get_data();
+ size_t data_index = column_data.size();
+ column_data.resize(data_index + select_vector.num_values() -
select_vector.num_filtered());
+ ColumnSelectVector::DataReadType read_type;
Review Comment:
warning: variable 'read_type' is not initialized
[cppcoreguidelines-init-variables]
```cpp
ColumnSelectVector::DataReadType read_type;
^
```
##########
be/src/vec/exec/format/parquet/vparquet_column_reader.cpp:
##########
@@ -312,6 +316,16 @@ Status ScalarColumnReader::_read_nested_column(ColumnPtr&
doris_column, DataType
ColumnSelectVector&
select_vector, size_t batch_size,
size_t* read_rows, bool* eof,
bool is_dict_filter,
bool align_rows = false) {
+ bool need_convert = false;
+ auto& physical_type = _chunk_meta.meta_data.type;
+ DataTypePtr src_type;
+ ParquetConvert::convert_data_type_from_parquet(physical_type, src_type,
type, &need_convert);
+
+ ColumnPtr src_column = doris_column;
Review Comment:
warning: variable 'src_column' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
ColumnPtr src_column = 0 = doris_column;
```
##########
be/src/vec/exec/format/parquet/parquet_column_convert.h:
##########
@@ -0,0 +1,548 @@
+// 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 <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/Types_types.h>
+#include <gen_cpp/parquet_types.h>
+
+#include <algorithm>
+#include <functional>
+#include <ostream>
+#include <utility>
+
+#include "common/compiler_util.h" // IWYU pragma: keep
+#include "common/status.h"
+#include "gen_cpp/descriptors.pb.h"
+#include "gutil/endian.h"
+#include "io/file_factory.h"
+#include "olap/olap_common.h"
+#include "util/coding.h"
+#include "util/slice.h"
+#include "vec/columns/column_string.h"
+#include "vec/columns/column_vector.h"
+#include "vec/core/types.h"
+#include "vec/data_types/data_type.h"
+#include "vec/data_types/data_type_factory.hpp"
+#include "vec/data_types/data_type_nullable.h"
+#include "vec/data_types/data_type_number.h"
+#include "vec/data_types/data_type_string.h"
+#include "vec/exec/format/format_common.h"
+#include "vec/exec/format/parquet/decoder.h"
+#include "vec/exec/format/parquet/parquet_common.h"
+
+namespace doris::vectorized {
+
+namespace ParquetConvert {
+
+template <tparquet::Type::type ParquetType>
+struct PhysicalTypeTraits {};
+
+template <>
+struct PhysicalTypeTraits<tparquet::Type::INT32> {
+ using DataType = int32_t;
+ using ColumnType = ColumnVector<DataType>;
+};
+
+template <>
+struct PhysicalTypeTraits<tparquet::Type::INT64> {
+ using DataType = int64_t;
+ using ColumnType = ColumnVector<DataType>;
+};
+
+template <>
+struct PhysicalTypeTraits<tparquet::Type::FLOAT> {
+ using DataType = float;
+ using ColumnType = ColumnVector<DataType>;
+};
+
+template <>
+struct PhysicalTypeTraits<tparquet::Type::DOUBLE> {
+ using DataType = double;
+ using ColumnType = ColumnVector<DataType>;
+};
+
+template <>
+struct PhysicalTypeTraits<tparquet::Type::BYTE_ARRAY> {
+ using DataType = String;
+ using ColumnType = ColumnString;
+};
+
+template <>
+struct PhysicalTypeTraits<tparquet::Type::FIXED_LEN_BYTE_ARRAY> {
+ using DataType = String;
+ using ColumnType = ColumnString;
+};
+
+template <>
+struct PhysicalTypeTraits<tparquet::Type::INT96> {
+ using DataType = Int128;
+ using ColumnType = ColumnVector<DataType>;
+};
+
+#define FOR_LOGICAL_NUMERIC_TYPES(M) \
+ M(TypeIndex::Int8, Int8, Int32) \
+ M(TypeIndex::UInt8, UInt8, Int32) \
+ M(TypeIndex::Int16, Int16, Int32) \
+ M(TypeIndex::UInt16, UInt16, Int32) \
+ M(TypeIndex::Int32, Int32, Int32) \
+ M(TypeIndex::UInt32, UInt32, Int32) \
+ M(TypeIndex::Int64, Int64, Int64) \
+ M(TypeIndex::UInt64, UInt64, Int64) \
+ M(TypeIndex::Float32, Float32, Float32) \
+ M(TypeIndex::Float64, Float64, Float64)
+
+struct ConvertParams {
+ // schema.logicalType.TIMESTAMP.isAdjustedToUTC == false
+ static const cctz::time_zone utc0;
+ // schema.logicalType.TIMESTAMP.isAdjustedToUTC == true, we should set the
time zone
+ cctz::time_zone* ctz = nullptr;
+ size_t offset_days = 0;
+ int64_t second_mask = 1;
+ int64_t scale_to_nano_factor = 1;
+ DecimalScaleParams decimal_scale;
+ FieldSchema* field_schema = nullptr;
+
+ void init(FieldSchema* field_schema_, cctz::time_zone* ctz_) {
+ field_schema = field_schema_;
+ if (ctz_ != nullptr) {
+ ctz = ctz_;
+ }
+ const auto& schema = field_schema->parquet_schema;
+ if (schema.__isset.logicalType &&
schema.logicalType.__isset.TIMESTAMP) {
+ const auto& timestamp_info = schema.logicalType.TIMESTAMP;
+ if (!timestamp_info.isAdjustedToUTC) {
+ // should set timezone to utc+0
+ ctz = const_cast<cctz::time_zone*>(&utc0);
+ }
+ const auto& time_unit = timestamp_info.unit;
+ if (time_unit.__isset.MILLIS) {
+ second_mask = 1000;
+ scale_to_nano_factor = 1000000;
+ } else if (time_unit.__isset.MICROS) {
+ second_mask = 1000000;
+ scale_to_nano_factor = 1000;
+ } else if (time_unit.__isset.NANOS) {
+ second_mask = 1000000000;
+ scale_to_nano_factor = 1;
+ }
+ } else if (schema.__isset.converted_type) {
+ const auto& converted_type = schema.converted_type;
+ if (converted_type == tparquet::ConvertedType::TIMESTAMP_MILLIS) {
+ second_mask = 1000;
+ scale_to_nano_factor = 1000000;
+ } else if (converted_type ==
tparquet::ConvertedType::TIMESTAMP_MICROS) {
+ second_mask = 1000000;
+ scale_to_nano_factor = 1000;
+ }
+ }
+
+ if (ctz) {
+ VecDateTimeValue t;
+ t.from_unixtime(0, *ctz);
+ offset_days = t.day() == 31 ? 0 : 1;
+ }
+ }
+
+ template <typename DecimalPrimitiveType>
+ void init_decimal_converter(DataTypePtr& data_type) {
+ if (field_schema == nullptr || decimal_scale.scale_type !=
DecimalScaleParams::NOT_INIT) {
+ return;
+ }
+ auto scale = field_schema->parquet_schema.scale;
+ auto* decimal_type =
static_cast<DataTypeDecimal<DecimalPrimitiveType>*>(
+ const_cast<IDataType*>(remove_nullable(data_type).get()));
+ auto dest_scale = decimal_type->get_scale();
+ if (dest_scale > scale) {
+ decimal_scale.scale_type = DecimalScaleParams::SCALE_UP;
+ decimal_scale.scale_factor =
+
DecimalScaleParams::get_scale_factor<DecimalPrimitiveType>(dest_scale - scale);
+ } else if (dest_scale < scale) {
+ decimal_scale.scale_type = DecimalScaleParams::SCALE_DOWN;
+ decimal_scale.scale_factor =
+
DecimalScaleParams::get_scale_factor<DecimalPrimitiveType>(scale - dest_scale);
+ } else {
+ decimal_scale.scale_type = DecimalScaleParams::NO_SCALE;
+ decimal_scale.scale_factor = 1;
+ }
+ }
+};
+inline const cctz::time_zone ConvertParams::utc0 = cctz::utc_time_zone();
+inline Status convert_data_type_from_parquet(tparquet::Type::type parquet_type,
+ vectorized::DataTypePtr&
ans_data_type,
+ DataTypePtr& src_type, bool*
need_convert) {
+ std::cout << getTypeName(src_type->get_type_id()) << "\n";
+ if (is_complex_type(src_type)) {
+ *need_convert = false;
+ return Status::OK();
+ }
+ switch (parquet_type) {
+ case tparquet::Type::type::BOOLEAN:
+ ans_data_type = std::make_shared<DataTypeUInt8>();
+ break;
+ case tparquet::Type::type::INT32:
+ ans_data_type = std::make_shared<DataTypeInt32>();
+ break;
+ case tparquet::Type::type::INT64:
+ ans_data_type = std::make_shared<DataTypeInt64>();
+ break;
+ case tparquet::Type::type::FLOAT:
+ ans_data_type = std::make_shared<DataTypeFloat32>();
+ break;
+ case tparquet::Type::type::DOUBLE:
+ ans_data_type = std::make_shared<DataTypeFloat64>();
+ break;
+ case tparquet::Type::type::BYTE_ARRAY:
+ case tparquet::Type::type::FIXED_LEN_BYTE_ARRAY:
+ ans_data_type = std::make_shared<DataTypeString>();
+ break;
+ case tparquet::Type::type::INT96:
+ ans_data_type = std::make_shared<DataTypeInt128>();
+ break;
+ default:
+ return Status::IOError("Can't read parquet type : {}", parquet_type);
+ }
+ if (ans_data_type->get_type_id() == src_type->get_type_id()) {
+ *need_convert = false;
+ return Status::OK();
+ }
+ if (src_type->is_nullable()) {
+ auto& nested_src_type =
+ reinterpret_cast<const
DataTypeNullable*>(src_type.get())->get_nested_type();
+ auto sub = ans_data_type;
+ ans_data_type = std::make_shared<DataTypeNullable>(ans_data_type);
+
+ if (nested_src_type->get_type_id() == sub->get_type_id()) {
+ *need_convert = false;
+ return Status::OK();
+ }
+ }
+
+ *need_convert = true;
+ return Status::OK();
+}
+
+struct ColumnConvert {
+ virtual Status convert(const IColumn* src_col, IColumn* dst_col) { return
Status::OK(); }
+
+ virtual ~ColumnConvert() = default;
+
+public:
+ ConvertParams* _convert_params;
+};
+
+template <typename src_type, typename dst_type, bool is_nullable>
+struct NumberColumnConvert : public ColumnConvert {
+ Status convert(const IColumn* src_col, IColumn* dst_col) override;
+};
+
+inline void convert_null(const IColumn** src_col, IColumn** dst_col) {
+ size_t rows = (*src_col)->size();
+ if ((*src_col)->is_nullable()) {
+ auto src_nullable_column = reinterpret_cast<const
ColumnNullable*>(*src_col);
+ auto dst_nullable_column = reinterpret_cast<ColumnNullable*>(*dst_col);
+ auto& dst_null_col = dst_nullable_column->get_null_map_column();
+
+ for (auto j = 0; j < rows; j++) {
+ dst_null_col.insert(src_nullable_column->get_null_map_column()[j]);
+ }
+
+ *src_col = &src_nullable_column->get_nested_column();
+ *dst_col = &dst_nullable_column->get_nested_column();
+ }
+}
+
+template <typename src_type, typename dst_type, bool is_nullable>
+Status NumberColumnConvert<src_type, dst_type, is_nullable>::convert(const
IColumn* src_col,
+ IColumn*
dst_col) {
+ size_t rows = src_col->size();
+ if constexpr (is_nullable) {
+ convert_null(&src_col, &dst_col);
+ }
+ auto& src_data = static_cast<const
ColumnVector<src_type>*>(src_col)->get_data();
+ dst_col->resize(rows);
+ auto& data = static_cast<ColumnVector<dst_type>*>(dst_col)->get_data();
+
+ for (int i = 0; i < rows; i++) {
+ dst_type value = static_cast<dst_type>(src_data[i]);
+ data[i] = value;
+ }
+
+ return Status::OK();
+}
+template <typename src_type, bool is_nullable>
+struct NumberColumnToStringConvert : public ColumnConvert {
+ Status convert(const IColumn* src_col, IColumn* dst_col) override;
+};
+
+template <typename src_type, bool is_nullable>
+Status NumberColumnToStringConvert<src_type, is_nullable>::convert(const
IColumn* src_col,
+ IColumn*
dst_col) {
+ size_t rows = src_col->size();
+ if constexpr (is_nullable) {
+ convert_null(&src_col, &dst_col);
+ }
+ auto& src_data = static_cast<const
ColumnVector<src_type>*>(src_col)->get_data();
+ auto str_col = static_cast<ColumnString*>(dst_col);
+ for (int i = 0; i < rows; i++) {
+ std::string value = std::to_string(src_data[i]);
+ str_col->insert_data(value.data(), value.size());
+ }
+ return Status::OK();
+}
+
+template <bool is_nullable>
+struct int128totimestamp : public ColumnConvert {
+ [[nodiscard]] inline uint64_t to_timestamp_micros(uint32_t hi, uint64_t
lo) const {
+ return (hi - ParquetInt96::JULIAN_EPOCH_OFFSET_DAYS) *
ParquetInt96::MICROS_IN_DAY +
+ lo / ParquetInt96::NANOS_PER_MICROSECOND;
+ }
+ Status convert(const IColumn* src_col, IColumn* dst_col) override {
+ size_t rows = src_col->size();
+ if constexpr (is_nullable) {
+ convert_null(&src_col, &dst_col);
+ }
+ auto& src_data = static_cast<const
ColumnVector<Int128>*>(src_col)->get_data();
+ dst_col->resize(rows);
+ auto& data = static_cast<ColumnVector<UInt64>*>(dst_col)->get_data();
+
+ for (int i = 0; i < rows; i++) {
+ __int128 x = src_data[i];
+ uint32_t hi = x >> 64;
+ uint64_t lo = (x << 64) >> 64;
+ auto& num = data[i];
+ auto& value =
reinterpret_cast<DateV2Value<DateTimeV2ValueType>&>(num);
+ int64_t micros = to_timestamp_micros(hi, lo);
+ value.from_unixtime(micros / 1000000, *_convert_params->ctz);
+ value.set_microsecond(micros % 1000000);
+ }
+ return Status::OK();
+ }
+};
+
+template <bool is_nullable>
+struct int64totimestamp : public ColumnConvert {
+public:
+ Status convert(const IColumn* src_col, IColumn* dst_col) override {
+ size_t rows = src_col->size();
+ if constexpr (is_nullable) {
+ convert_null(&src_col, &dst_col);
+ }
+ dst_col->resize(rows);
+ auto& src_data = static_cast<const
ColumnVector<Int64>*>(src_col)->get_data();
+ auto& data = static_cast<ColumnVector<UInt64>*>(dst_col)->get_data();
+ for (int i = 0; i < rows; i++) {
+ int64 x = src_data[i];
+ dst_col = static_cast<ColumnVector<UInt64>*>(dst_col);
+ auto& num = data[i];
+ auto& value =
reinterpret_cast<DateV2Value<DateTimeV2ValueType>&>(num);
+ value.from_unixtime(x / _convert_params->second_mask,
*_convert_params->ctz);
+ value.set_microsecond((x % _convert_params->second_mask) *
+ _convert_params->scale_to_nano_factor /
1000);
+ std::cout << "value = " << value << "\n";
+ }
+ return Status::OK();
+ }
+};
+
+template <bool is_nullable>
+class int32todate : public ColumnConvert {
+public:
+ Status convert(const IColumn* src_col, IColumn* dst_col) override {
+ size_t rows = src_col->size();
+ if constexpr (is_nullable) {
+ convert_null(&src_col, &dst_col);
+ }
+ dst_col->resize(rows);
+ auto& src_data = reinterpret_cast<const
ColumnVector<uint32>*>(src_col)->get_data();
+ auto& data = static_cast<ColumnDateV2*>(dst_col)->get_data();
+ date_day_offset_dict& date_dict = date_day_offset_dict::get();
+
+ for (int i = 0; i < rows; i++) {
+ auto& value =
reinterpret_cast<DateV2Value<DateV2ValueType>&>(data[i]);
+ int64_t date_value = src_data[i] + _convert_params->offset_days;
+ value = date_dict[date_value];
+ }
+
+ return Status::OK();
+ }
+};
+
+template <typename DecimalType, bool is_nullable>
+class stringtodecimal : public ColumnConvert {
+public:
+ Status convert(const IColumn* src_col, IColumn* dst_col) override {
+ size_t rows = src_col->size();
+ if constexpr (is_nullable) {
+ convert_null(&src_col, &dst_col);
+ }
+ DecimalScaleParams& scale_params = _convert_params->decimal_scale;
+ auto buf = static_cast<const
ColumnString*>(src_col)->get_chars().data();
+ auto& offset = static_cast<const
ColumnString*>(src_col)->get_offsets();
+ dst_col->resize(rows);
+ auto& data =
static_cast<ColumnDecimal<DecimalType>*>(dst_col)->get_data();
+ for (int i = 0; i < rows; i++) {
+ int len = offset[i] - offset[i - 1];
+ Int128 value = buf[offset[i - 1]] & 0x80 ? -1 : 0;
+ memcpy(reinterpret_cast<char*>(&value) + sizeof(Int128) - len, buf
+ offset[i - 1],
+ len);
+ value = BigEndian::ToHost128(value);
+ if (scale_params.scale_type == DecimalScaleParams::SCALE_UP) {
+ value *= scale_params.scale_factor;
+ } else if (scale_params.scale_type ==
DecimalScaleParams::SCALE_DOWN) {
+ value /= scale_params.scale_factor;
+ }
+ auto& v = reinterpret_cast<DecimalType&>(data[i]);
+ v = (DecimalType)value;
+ }
+
+ return Status::OK();
+ }
+};
+template <typename NumberType, typename DecimalPhysicalType, bool is_nullable>
+class numbertodecimal : public ColumnConvert {
+public:
+ Status convert(const IColumn* src_col, IColumn* dst_col) override {
+ size_t rows = src_col->size();
+ if constexpr (is_nullable) {
+ convert_null(&src_col, &dst_col);
+ }
+ auto* src_data = static_cast<const
ColumnVector<NumberType>*>(src_col)->get_data().data();
+ dst_col->resize(rows);
+ DecimalScaleParams& scale_params = _convert_params->decimal_scale;
+ auto* data =
static_cast<ColumnDecimal<Decimal<Int64>>*>(dst_col)->get_data().data();
Review Comment:
warning: variable 'data' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
auto* data = nullptr =
static_cast<ColumnDecimal<Decimal<Int64>>*>(dst_col)->get_data().data();
```
##########
be/src/vec/exec/format/parquet/delta_bit_pack_decoder.h:
##########
@@ -361,3 +346,260 @@ class DeltaByteArrayDecoder : public DeltaDecoder {
std::vector<char> _buffered_data;
};
} // namespace doris::vectorized
+
+namespace doris::vectorized {
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_header() {
+ if (!_bit_reader->GetVlqInt(&_values_per_block) ||
+ !_bit_reader->GetVlqInt(&_mini_blocks_per_block) ||
+ !_bit_reader->GetVlqInt(&_total_value_count) ||
+ !_bit_reader->GetZigZagVlqInt(&_last_value)) {
+ return Status::IOError("Init header eof");
+ }
+ if (_values_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per block");
+ }
+ if (_values_per_block % 128 != 0) {
+ return Status::InvalidArgument(
+ "the number of values in a block must be multiple of 128, but
it's " +
+ std::to_string(_values_per_block));
+ }
+ if (_mini_blocks_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero miniblock per block");
+ }
+ _values_per_mini_block = _values_per_block / _mini_blocks_per_block;
+ if (_values_per_mini_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per miniblock");
+ }
+ if (_values_per_mini_block % 32 != 0) {
+ return Status::InvalidArgument(
+ "The number of values in a miniblock must be multiple of 32,
but it's " +
+ std::to_string(_values_per_mini_block));
+ }
+ _total_values_remaining = _total_value_count;
+ _delta_bit_widths.resize(_mini_blocks_per_block);
+ // init as empty property
+ _block_initialized = false;
+ _values_remaining_current_mini_block = 0;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_block() {
+ DCHECK_GT(_total_values_remaining, 0) << "InitBlock called at EOF";
+ if (!_bit_reader->GetZigZagVlqInt(&_min_delta)) {
+ return Status::IOError("Init block eof");
+ }
+
+ // read the bitwidth of each miniblock
+ uint8_t* bit_width_data = _delta_bit_widths.data();
+ for (uint32_t i = 0; i < _mini_blocks_per_block; ++i) {
+ if (!_bit_reader->GetAligned<uint8_t>(1, bit_width_data + i)) {
+ return Status::IOError("Decode bit-width EOF");
+ }
+ // Note that non-conformant bitwidth entries are allowed by the
Parquet spec
+ // for extraneous miniblocks in the last block (GH-14923), so we check
+ // the bitwidths when actually using them (see InitMiniBlock()).
+ }
+ _mini_block_idx = 0;
+ _block_initialized = true;
+ RETURN_IF_ERROR(_init_mini_block(bit_width_data[0]));
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_mini_block(int bit_width) {
+ if (PREDICT_FALSE(bit_width > kMaxDeltaBitWidth)) {
+ return Status::InvalidArgument("delta bit width larger than integer
bit width");
+ }
+ _delta_bit_width = bit_width;
+ _values_remaining_current_mini_block = _values_per_mini_block;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_get_internal(T* buffer, int
num_values,
+ int*
out_num_values) {
+ num_values = static_cast<int>(std::min<int64_t>(num_values,
_total_values_remaining));
+ if (num_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+ int i = 0;
+ while (i < num_values) {
+ if (PREDICT_FALSE(_values_remaining_current_mini_block == 0)) {
+ if (PREDICT_FALSE(!_block_initialized)) {
+ buffer[i++] = _last_value;
+ DCHECK_EQ(i, 1); // we're at the beginning of the page
+ if (i == num_values) {
+ // When block is uninitialized and i reaches num_values we
have two
+ // different possibilities:
+ // 1. _total_value_count == 1, which means that the page
may have only
+ // one value (encoded in the header), and we should not
initialize
+ // any block.
+ // 2. _total_value_count != 1, which means we should
initialize the
+ // incoming block for subsequent reads.
+ if (_total_value_count != 1) {
+ RETURN_IF_ERROR(_init_block());
+ }
+ break;
+ }
+ RETURN_IF_ERROR(_init_block());
+ } else {
+ ++_mini_block_idx;
+ if (_mini_block_idx < _mini_blocks_per_block) {
+
RETURN_IF_ERROR(_init_mini_block(_delta_bit_widths.data()[_mini_block_idx]));
+ } else {
+ RETURN_IF_ERROR(_init_block());
+ }
+ }
+ }
+
+ int values_decode = std::min(_values_remaining_current_mini_block,
+ static_cast<uint32_t>(num_values - i));
+ for (int j = 0; j < values_decode; ++j) {
+ if (!_bit_reader->GetValue(_delta_bit_width, buffer + i + j)) {
+ return Status::IOError("Get batch EOF");
+ }
+ }
+ for (int j = 0; j < values_decode; ++j) {
+ // Addition between min_delta, packed int and last_value should be
treated as
+ // unsigned addition. Overflow is as expected.
+ buffer[i + j] = static_cast<UT>(_min_delta) +
static_cast<UT>(buffer[i + j]) +
+ static_cast<UT>(_last_value);
+ _last_value = buffer[i + j];
+ }
+ _values_remaining_current_mini_block -= values_decode;
+ i += values_decode;
+ }
+ _total_values_remaining -= num_values;
+
+ if (PREDICT_FALSE(_total_values_remaining == 0)) {
+ if (!_bit_reader->Advance(_delta_bit_width *
_values_remaining_current_mini_block)) {
+ return Status::IOError("Skip padding EOF");
+ }
+ _values_remaining_current_mini_block = 0;
+ }
+ *out_num_values = num_values;
+ return Status::OK();
+}
+template <tparquet::Type::type PhysicalType>
+void DeltaLengthByteArrayDecoder<PhysicalType>::_decode_lengths() {
+ _len_decoder.set_bit_reader(_bit_reader);
+ // get the number of encoded lengths
+ int num_length = _len_decoder.valid_values_count();
+ _buffered_length.resize(num_length);
+
+ // decode all the lengths. all the lengths are buffered in
buffered_length_.
+ int ret;
+ Status st = _len_decoder.decode(_buffered_length.data(), num_length, &ret);
+ if (!st.ok()) {
+ LOG(FATAL) << "Fail to decode delta length, status: " << st;
+ }
+ DCHECK_EQ(ret, num_length);
+ _length_idx = 0;
+ _num_valid_values = num_length;
+}
+template <tparquet::Type::type PhysicalType>
+Status DeltaLengthByteArrayDecoder<PhysicalType>::_get_internal(Slice* buffer,
int max_values,
+ int*
out_num_values) {
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+
+ int32_t data_size = 0;
+ const int32_t* length_ptr = _buffered_length.data() + _length_idx;
+ for (int i = 0; i < max_values; ++i) {
+ int32_t len = length_ptr[i];
+ if (PREDICT_FALSE(len < 0)) {
+ return Status::InvalidArgument("Negative string delta length");
+ }
+ buffer[i].size = len;
+ if (common::add_overflow(data_size, len, data_size)) {
+ return Status::InvalidArgument("Excess expansion in
DELTA_(LENGTH_)BYTE_ARRAY");
+ }
+ }
+ _length_idx += max_values;
+
+ _buffered_data.resize(data_size);
+ char* data_ptr = _buffered_data.data();
+ for (int j = 0; j < data_size; j++) {
+ if (!_bit_reader->GetValue(8, data_ptr + j)) {
+ return Status::IOError("Get length bytes EOF");
+ }
+ }
+
+ for (int i = 0; i < max_values; ++i) {
+ buffer[i].data = data_ptr;
+ data_ptr += buffer[i].size;
+ }
+ // this->num_values_ -= max_values;
+ _num_valid_values -= max_values;
+ *out_num_values = max_values;
+ return Status::OK();
+}
+
+template <tparquet::Type::type PhysicalType>
+Status DeltaByteArrayDecoder<PhysicalType>::_get_internal(Slice* buffer, int
max_values,
+ int* out_num_values)
{
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = max_values;
+ return Status::OK();
+ }
+
+ int suffix_read;
Review Comment:
warning: variable 'suffix_read' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
int suffix_read = 0;
```
##########
be/src/vec/exec/format/parquet/delta_bit_pack_decoder.h:
##########
@@ -361,3 +346,260 @@ class DeltaByteArrayDecoder : public DeltaDecoder {
std::vector<char> _buffered_data;
};
} // namespace doris::vectorized
+
+namespace doris::vectorized {
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_header() {
+ if (!_bit_reader->GetVlqInt(&_values_per_block) ||
+ !_bit_reader->GetVlqInt(&_mini_blocks_per_block) ||
+ !_bit_reader->GetVlqInt(&_total_value_count) ||
+ !_bit_reader->GetZigZagVlqInt(&_last_value)) {
+ return Status::IOError("Init header eof");
+ }
+ if (_values_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per block");
+ }
+ if (_values_per_block % 128 != 0) {
+ return Status::InvalidArgument(
+ "the number of values in a block must be multiple of 128, but
it's " +
+ std::to_string(_values_per_block));
+ }
+ if (_mini_blocks_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero miniblock per block");
+ }
+ _values_per_mini_block = _values_per_block / _mini_blocks_per_block;
+ if (_values_per_mini_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per miniblock");
+ }
+ if (_values_per_mini_block % 32 != 0) {
+ return Status::InvalidArgument(
+ "The number of values in a miniblock must be multiple of 32,
but it's " +
+ std::to_string(_values_per_mini_block));
+ }
+ _total_values_remaining = _total_value_count;
+ _delta_bit_widths.resize(_mini_blocks_per_block);
+ // init as empty property
+ _block_initialized = false;
+ _values_remaining_current_mini_block = 0;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_block() {
+ DCHECK_GT(_total_values_remaining, 0) << "InitBlock called at EOF";
+ if (!_bit_reader->GetZigZagVlqInt(&_min_delta)) {
+ return Status::IOError("Init block eof");
+ }
+
+ // read the bitwidth of each miniblock
+ uint8_t* bit_width_data = _delta_bit_widths.data();
+ for (uint32_t i = 0; i < _mini_blocks_per_block; ++i) {
+ if (!_bit_reader->GetAligned<uint8_t>(1, bit_width_data + i)) {
+ return Status::IOError("Decode bit-width EOF");
+ }
+ // Note that non-conformant bitwidth entries are allowed by the
Parquet spec
+ // for extraneous miniblocks in the last block (GH-14923), so we check
+ // the bitwidths when actually using them (see InitMiniBlock()).
+ }
+ _mini_block_idx = 0;
+ _block_initialized = true;
+ RETURN_IF_ERROR(_init_mini_block(bit_width_data[0]));
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_mini_block(int bit_width) {
+ if (PREDICT_FALSE(bit_width > kMaxDeltaBitWidth)) {
+ return Status::InvalidArgument("delta bit width larger than integer
bit width");
+ }
+ _delta_bit_width = bit_width;
+ _values_remaining_current_mini_block = _values_per_mini_block;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_get_internal(T* buffer, int
num_values,
+ int*
out_num_values) {
+ num_values = static_cast<int>(std::min<int64_t>(num_values,
_total_values_remaining));
+ if (num_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+ int i = 0;
+ while (i < num_values) {
+ if (PREDICT_FALSE(_values_remaining_current_mini_block == 0)) {
+ if (PREDICT_FALSE(!_block_initialized)) {
+ buffer[i++] = _last_value;
+ DCHECK_EQ(i, 1); // we're at the beginning of the page
+ if (i == num_values) {
+ // When block is uninitialized and i reaches num_values we
have two
+ // different possibilities:
+ // 1. _total_value_count == 1, which means that the page
may have only
+ // one value (encoded in the header), and we should not
initialize
+ // any block.
+ // 2. _total_value_count != 1, which means we should
initialize the
+ // incoming block for subsequent reads.
+ if (_total_value_count != 1) {
+ RETURN_IF_ERROR(_init_block());
+ }
+ break;
+ }
+ RETURN_IF_ERROR(_init_block());
+ } else {
+ ++_mini_block_idx;
+ if (_mini_block_idx < _mini_blocks_per_block) {
+
RETURN_IF_ERROR(_init_mini_block(_delta_bit_widths.data()[_mini_block_idx]));
+ } else {
+ RETURN_IF_ERROR(_init_block());
+ }
+ }
+ }
+
+ int values_decode = std::min(_values_remaining_current_mini_block,
+ static_cast<uint32_t>(num_values - i));
+ for (int j = 0; j < values_decode; ++j) {
+ if (!_bit_reader->GetValue(_delta_bit_width, buffer + i + j)) {
+ return Status::IOError("Get batch EOF");
+ }
+ }
+ for (int j = 0; j < values_decode; ++j) {
+ // Addition between min_delta, packed int and last_value should be
treated as
+ // unsigned addition. Overflow is as expected.
+ buffer[i + j] = static_cast<UT>(_min_delta) +
static_cast<UT>(buffer[i + j]) +
+ static_cast<UT>(_last_value);
+ _last_value = buffer[i + j];
+ }
+ _values_remaining_current_mini_block -= values_decode;
+ i += values_decode;
+ }
+ _total_values_remaining -= num_values;
+
+ if (PREDICT_FALSE(_total_values_remaining == 0)) {
+ if (!_bit_reader->Advance(_delta_bit_width *
_values_remaining_current_mini_block)) {
+ return Status::IOError("Skip padding EOF");
+ }
+ _values_remaining_current_mini_block = 0;
+ }
+ *out_num_values = num_values;
+ return Status::OK();
+}
+template <tparquet::Type::type PhysicalType>
+void DeltaLengthByteArrayDecoder<PhysicalType>::_decode_lengths() {
+ _len_decoder.set_bit_reader(_bit_reader);
+ // get the number of encoded lengths
+ int num_length = _len_decoder.valid_values_count();
+ _buffered_length.resize(num_length);
+
+ // decode all the lengths. all the lengths are buffered in
buffered_length_.
+ int ret;
+ Status st = _len_decoder.decode(_buffered_length.data(), num_length, &ret);
+ if (!st.ok()) {
+ LOG(FATAL) << "Fail to decode delta length, status: " << st;
+ }
+ DCHECK_EQ(ret, num_length);
+ _length_idx = 0;
+ _num_valid_values = num_length;
+}
+template <tparquet::Type::type PhysicalType>
+Status DeltaLengthByteArrayDecoder<PhysicalType>::_get_internal(Slice* buffer,
int max_values,
+ int*
out_num_values) {
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+
+ int32_t data_size = 0;
+ const int32_t* length_ptr = _buffered_length.data() + _length_idx;
+ for (int i = 0; i < max_values; ++i) {
+ int32_t len = length_ptr[i];
+ if (PREDICT_FALSE(len < 0)) {
+ return Status::InvalidArgument("Negative string delta length");
+ }
+ buffer[i].size = len;
+ if (common::add_overflow(data_size, len, data_size)) {
+ return Status::InvalidArgument("Excess expansion in
DELTA_(LENGTH_)BYTE_ARRAY");
+ }
+ }
+ _length_idx += max_values;
+
+ _buffered_data.resize(data_size);
+ char* data_ptr = _buffered_data.data();
+ for (int j = 0; j < data_size; j++) {
+ if (!_bit_reader->GetValue(8, data_ptr + j)) {
+ return Status::IOError("Get length bytes EOF");
+ }
+ }
+
+ for (int i = 0; i < max_values; ++i) {
+ buffer[i].data = data_ptr;
+ data_ptr += buffer[i].size;
+ }
+ // this->num_values_ -= max_values;
+ _num_valid_values -= max_values;
+ *out_num_values = max_values;
+ return Status::OK();
+}
+
+template <tparquet::Type::type PhysicalType>
+Status DeltaByteArrayDecoder<PhysicalType>::_get_internal(Slice* buffer, int
max_values,
+ int* out_num_values)
{
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = max_values;
+ return Status::OK();
+ }
+
+ int suffix_read;
+ RETURN_IF_ERROR(_suffix_decoder.decode(buffer, max_values, &suffix_read));
+ if (PREDICT_FALSE(suffix_read != max_values)) {
+ return Status::IOError("Read {}, expecting {} from suffix decoder",
+ std::to_string(suffix_read),
std::to_string(max_values));
+ }
+
+ int64_t data_size = 0;
+ const int32_t* prefix_len_ptr = _buffered_prefix_length.data() +
_prefix_len_offset;
Review Comment:
warning: variable 'prefix_len_ptr' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
const int32_t* prefix_len_ptr = nullptr = _buffered_prefix_length.data()
+ _prefix_len_offset;
```
##########
be/src/vec/exec/format/parquet/fix_length_plain_decoder.h:
##########
@@ -32,44 +33,144 @@ class ColumnSelectVector;
namespace doris::vectorized {
+template <tparquet::Type::type PhysicalType>
class FixLengthPlainDecoder final : public Decoder {
public:
- FixLengthPlainDecoder(tparquet::Type::type physical_type) :
_physical_type(physical_type) {};
+ FixLengthPlainDecoder() {};
Review Comment:
warning: use '= default' to define a trivial default constructor
[modernize-use-equals-default]
```suggestion
FixLengthPlainDecoder() = default;;
```
##########
be/src/vec/exec/format/parquet/delta_bit_pack_decoder.h:
##########
@@ -361,3 +346,260 @@ class DeltaByteArrayDecoder : public DeltaDecoder {
std::vector<char> _buffered_data;
};
} // namespace doris::vectorized
+
+namespace doris::vectorized {
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_header() {
+ if (!_bit_reader->GetVlqInt(&_values_per_block) ||
+ !_bit_reader->GetVlqInt(&_mini_blocks_per_block) ||
+ !_bit_reader->GetVlqInt(&_total_value_count) ||
+ !_bit_reader->GetZigZagVlqInt(&_last_value)) {
+ return Status::IOError("Init header eof");
+ }
+ if (_values_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per block");
+ }
+ if (_values_per_block % 128 != 0) {
+ return Status::InvalidArgument(
+ "the number of values in a block must be multiple of 128, but
it's " +
+ std::to_string(_values_per_block));
+ }
+ if (_mini_blocks_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero miniblock per block");
+ }
+ _values_per_mini_block = _values_per_block / _mini_blocks_per_block;
+ if (_values_per_mini_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per miniblock");
+ }
+ if (_values_per_mini_block % 32 != 0) {
+ return Status::InvalidArgument(
+ "The number of values in a miniblock must be multiple of 32,
but it's " +
+ std::to_string(_values_per_mini_block));
+ }
+ _total_values_remaining = _total_value_count;
+ _delta_bit_widths.resize(_mini_blocks_per_block);
+ // init as empty property
+ _block_initialized = false;
+ _values_remaining_current_mini_block = 0;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_block() {
+ DCHECK_GT(_total_values_remaining, 0) << "InitBlock called at EOF";
+ if (!_bit_reader->GetZigZagVlqInt(&_min_delta)) {
+ return Status::IOError("Init block eof");
+ }
+
+ // read the bitwidth of each miniblock
+ uint8_t* bit_width_data = _delta_bit_widths.data();
+ for (uint32_t i = 0; i < _mini_blocks_per_block; ++i) {
+ if (!_bit_reader->GetAligned<uint8_t>(1, bit_width_data + i)) {
+ return Status::IOError("Decode bit-width EOF");
+ }
+ // Note that non-conformant bitwidth entries are allowed by the
Parquet spec
+ // for extraneous miniblocks in the last block (GH-14923), so we check
+ // the bitwidths when actually using them (see InitMiniBlock()).
+ }
+ _mini_block_idx = 0;
+ _block_initialized = true;
+ RETURN_IF_ERROR(_init_mini_block(bit_width_data[0]));
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_mini_block(int bit_width) {
+ if (PREDICT_FALSE(bit_width > kMaxDeltaBitWidth)) {
+ return Status::InvalidArgument("delta bit width larger than integer
bit width");
+ }
+ _delta_bit_width = bit_width;
+ _values_remaining_current_mini_block = _values_per_mini_block;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_get_internal(T* buffer, int
num_values,
+ int*
out_num_values) {
+ num_values = static_cast<int>(std::min<int64_t>(num_values,
_total_values_remaining));
+ if (num_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+ int i = 0;
+ while (i < num_values) {
+ if (PREDICT_FALSE(_values_remaining_current_mini_block == 0)) {
+ if (PREDICT_FALSE(!_block_initialized)) {
+ buffer[i++] = _last_value;
+ DCHECK_EQ(i, 1); // we're at the beginning of the page
+ if (i == num_values) {
+ // When block is uninitialized and i reaches num_values we
have two
+ // different possibilities:
+ // 1. _total_value_count == 1, which means that the page
may have only
+ // one value (encoded in the header), and we should not
initialize
+ // any block.
+ // 2. _total_value_count != 1, which means we should
initialize the
+ // incoming block for subsequent reads.
+ if (_total_value_count != 1) {
+ RETURN_IF_ERROR(_init_block());
+ }
+ break;
+ }
+ RETURN_IF_ERROR(_init_block());
+ } else {
+ ++_mini_block_idx;
+ if (_mini_block_idx < _mini_blocks_per_block) {
+
RETURN_IF_ERROR(_init_mini_block(_delta_bit_widths.data()[_mini_block_idx]));
+ } else {
+ RETURN_IF_ERROR(_init_block());
+ }
+ }
+ }
+
+ int values_decode = std::min(_values_remaining_current_mini_block,
+ static_cast<uint32_t>(num_values - i));
+ for (int j = 0; j < values_decode; ++j) {
+ if (!_bit_reader->GetValue(_delta_bit_width, buffer + i + j)) {
+ return Status::IOError("Get batch EOF");
+ }
+ }
+ for (int j = 0; j < values_decode; ++j) {
+ // Addition between min_delta, packed int and last_value should be
treated as
+ // unsigned addition. Overflow is as expected.
+ buffer[i + j] = static_cast<UT>(_min_delta) +
static_cast<UT>(buffer[i + j]) +
+ static_cast<UT>(_last_value);
+ _last_value = buffer[i + j];
+ }
+ _values_remaining_current_mini_block -= values_decode;
+ i += values_decode;
+ }
+ _total_values_remaining -= num_values;
+
+ if (PREDICT_FALSE(_total_values_remaining == 0)) {
+ if (!_bit_reader->Advance(_delta_bit_width *
_values_remaining_current_mini_block)) {
+ return Status::IOError("Skip padding EOF");
+ }
+ _values_remaining_current_mini_block = 0;
+ }
+ *out_num_values = num_values;
+ return Status::OK();
+}
+template <tparquet::Type::type PhysicalType>
+void DeltaLengthByteArrayDecoder<PhysicalType>::_decode_lengths() {
+ _len_decoder.set_bit_reader(_bit_reader);
+ // get the number of encoded lengths
+ int num_length = _len_decoder.valid_values_count();
+ _buffered_length.resize(num_length);
+
+ // decode all the lengths. all the lengths are buffered in
buffered_length_.
+ int ret;
+ Status st = _len_decoder.decode(_buffered_length.data(), num_length, &ret);
+ if (!st.ok()) {
+ LOG(FATAL) << "Fail to decode delta length, status: " << st;
+ }
+ DCHECK_EQ(ret, num_length);
+ _length_idx = 0;
+ _num_valid_values = num_length;
+}
+template <tparquet::Type::type PhysicalType>
+Status DeltaLengthByteArrayDecoder<PhysicalType>::_get_internal(Slice* buffer,
int max_values,
+ int*
out_num_values) {
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+
+ int32_t data_size = 0;
+ const int32_t* length_ptr = _buffered_length.data() + _length_idx;
+ for (int i = 0; i < max_values; ++i) {
+ int32_t len = length_ptr[i];
+ if (PREDICT_FALSE(len < 0)) {
+ return Status::InvalidArgument("Negative string delta length");
+ }
+ buffer[i].size = len;
+ if (common::add_overflow(data_size, len, data_size)) {
+ return Status::InvalidArgument("Excess expansion in
DELTA_(LENGTH_)BYTE_ARRAY");
+ }
+ }
+ _length_idx += max_values;
+
+ _buffered_data.resize(data_size);
+ char* data_ptr = _buffered_data.data();
+ for (int j = 0; j < data_size; j++) {
+ if (!_bit_reader->GetValue(8, data_ptr + j)) {
+ return Status::IOError("Get length bytes EOF");
+ }
+ }
+
+ for (int i = 0; i < max_values; ++i) {
+ buffer[i].data = data_ptr;
+ data_ptr += buffer[i].size;
+ }
+ // this->num_values_ -= max_values;
+ _num_valid_values -= max_values;
+ *out_num_values = max_values;
+ return Status::OK();
+}
+
+template <tparquet::Type::type PhysicalType>
+Status DeltaByteArrayDecoder<PhysicalType>::_get_internal(Slice* buffer, int
max_values,
+ int* out_num_values)
{
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = max_values;
+ return Status::OK();
+ }
+
+ int suffix_read;
+ RETURN_IF_ERROR(_suffix_decoder.decode(buffer, max_values, &suffix_read));
+ if (PREDICT_FALSE(suffix_read != max_values)) {
+ return Status::IOError("Read {}, expecting {} from suffix decoder",
+ std::to_string(suffix_read),
std::to_string(max_values));
+ }
+
+ int64_t data_size = 0;
+ const int32_t* prefix_len_ptr = _buffered_prefix_length.data() +
_prefix_len_offset;
+ for (int i = 0; i < max_values; ++i) {
+ if (PREDICT_FALSE(prefix_len_ptr[i] < 0)) {
+ return Status::InvalidArgument("negative prefix length in
DELTA_BYTE_ARRAY");
+ }
+ if (PREDICT_FALSE(common::add_overflow(data_size,
static_cast<int64_t>(prefix_len_ptr[i]),
+ data_size) ||
+ common::add_overflow(data_size,
static_cast<int64_t>(buffer[i].size),
+ data_size))) {
+ return Status::InvalidArgument("excess expansion in
DELTA_BYTE_ARRAY");
+ }
+ }
+ _buffered_data.resize(data_size);
+
+ std::string_view prefix {_last_value};
Review Comment:
warning: variable 'prefix' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
std::string_view prefix = 0 {_last_value};
```
##########
be/src/vec/exec/format/parquet/delta_bit_pack_decoder.h:
##########
@@ -361,3 +346,260 @@ class DeltaByteArrayDecoder : public DeltaDecoder {
std::vector<char> _buffered_data;
};
} // namespace doris::vectorized
+
+namespace doris::vectorized {
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_header() {
+ if (!_bit_reader->GetVlqInt(&_values_per_block) ||
+ !_bit_reader->GetVlqInt(&_mini_blocks_per_block) ||
+ !_bit_reader->GetVlqInt(&_total_value_count) ||
+ !_bit_reader->GetZigZagVlqInt(&_last_value)) {
+ return Status::IOError("Init header eof");
+ }
+ if (_values_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per block");
+ }
+ if (_values_per_block % 128 != 0) {
+ return Status::InvalidArgument(
+ "the number of values in a block must be multiple of 128, but
it's " +
+ std::to_string(_values_per_block));
+ }
+ if (_mini_blocks_per_block == 0) {
+ return Status::InvalidArgument("Cannot have zero miniblock per block");
+ }
+ _values_per_mini_block = _values_per_block / _mini_blocks_per_block;
+ if (_values_per_mini_block == 0) {
+ return Status::InvalidArgument("Cannot have zero value per miniblock");
+ }
+ if (_values_per_mini_block % 32 != 0) {
+ return Status::InvalidArgument(
+ "The number of values in a miniblock must be multiple of 32,
but it's " +
+ std::to_string(_values_per_mini_block));
+ }
+ _total_values_remaining = _total_value_count;
+ _delta_bit_widths.resize(_mini_blocks_per_block);
+ // init as empty property
+ _block_initialized = false;
+ _values_remaining_current_mini_block = 0;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_block() {
+ DCHECK_GT(_total_values_remaining, 0) << "InitBlock called at EOF";
+ if (!_bit_reader->GetZigZagVlqInt(&_min_delta)) {
+ return Status::IOError("Init block eof");
+ }
+
+ // read the bitwidth of each miniblock
+ uint8_t* bit_width_data = _delta_bit_widths.data();
+ for (uint32_t i = 0; i < _mini_blocks_per_block; ++i) {
+ if (!_bit_reader->GetAligned<uint8_t>(1, bit_width_data + i)) {
+ return Status::IOError("Decode bit-width EOF");
+ }
+ // Note that non-conformant bitwidth entries are allowed by the
Parquet spec
+ // for extraneous miniblocks in the last block (GH-14923), so we check
+ // the bitwidths when actually using them (see InitMiniBlock()).
+ }
+ _mini_block_idx = 0;
+ _block_initialized = true;
+ RETURN_IF_ERROR(_init_mini_block(bit_width_data[0]));
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_init_mini_block(int bit_width) {
+ if (PREDICT_FALSE(bit_width > kMaxDeltaBitWidth)) {
+ return Status::InvalidArgument("delta bit width larger than integer
bit width");
+ }
+ _delta_bit_width = bit_width;
+ _values_remaining_current_mini_block = _values_per_mini_block;
+ return Status::OK();
+}
+
+template <typename T, tparquet::Type::type PhysicalType>
+Status DeltaBitPackDecoder<T, PhysicalType>::_get_internal(T* buffer, int
num_values,
+ int*
out_num_values) {
+ num_values = static_cast<int>(std::min<int64_t>(num_values,
_total_values_remaining));
+ if (num_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+ int i = 0;
+ while (i < num_values) {
+ if (PREDICT_FALSE(_values_remaining_current_mini_block == 0)) {
+ if (PREDICT_FALSE(!_block_initialized)) {
+ buffer[i++] = _last_value;
+ DCHECK_EQ(i, 1); // we're at the beginning of the page
+ if (i == num_values) {
+ // When block is uninitialized and i reaches num_values we
have two
+ // different possibilities:
+ // 1. _total_value_count == 1, which means that the page
may have only
+ // one value (encoded in the header), and we should not
initialize
+ // any block.
+ // 2. _total_value_count != 1, which means we should
initialize the
+ // incoming block for subsequent reads.
+ if (_total_value_count != 1) {
+ RETURN_IF_ERROR(_init_block());
+ }
+ break;
+ }
+ RETURN_IF_ERROR(_init_block());
+ } else {
+ ++_mini_block_idx;
+ if (_mini_block_idx < _mini_blocks_per_block) {
+
RETURN_IF_ERROR(_init_mini_block(_delta_bit_widths.data()[_mini_block_idx]));
+ } else {
+ RETURN_IF_ERROR(_init_block());
+ }
+ }
+ }
+
+ int values_decode = std::min(_values_remaining_current_mini_block,
+ static_cast<uint32_t>(num_values - i));
+ for (int j = 0; j < values_decode; ++j) {
+ if (!_bit_reader->GetValue(_delta_bit_width, buffer + i + j)) {
+ return Status::IOError("Get batch EOF");
+ }
+ }
+ for (int j = 0; j < values_decode; ++j) {
+ // Addition between min_delta, packed int and last_value should be
treated as
+ // unsigned addition. Overflow is as expected.
+ buffer[i + j] = static_cast<UT>(_min_delta) +
static_cast<UT>(buffer[i + j]) +
+ static_cast<UT>(_last_value);
+ _last_value = buffer[i + j];
+ }
+ _values_remaining_current_mini_block -= values_decode;
+ i += values_decode;
+ }
+ _total_values_remaining -= num_values;
+
+ if (PREDICT_FALSE(_total_values_remaining == 0)) {
+ if (!_bit_reader->Advance(_delta_bit_width *
_values_remaining_current_mini_block)) {
+ return Status::IOError("Skip padding EOF");
+ }
+ _values_remaining_current_mini_block = 0;
+ }
+ *out_num_values = num_values;
+ return Status::OK();
+}
+template <tparquet::Type::type PhysicalType>
+void DeltaLengthByteArrayDecoder<PhysicalType>::_decode_lengths() {
+ _len_decoder.set_bit_reader(_bit_reader);
+ // get the number of encoded lengths
+ int num_length = _len_decoder.valid_values_count();
+ _buffered_length.resize(num_length);
+
+ // decode all the lengths. all the lengths are buffered in
buffered_length_.
+ int ret;
+ Status st = _len_decoder.decode(_buffered_length.data(), num_length, &ret);
+ if (!st.ok()) {
+ LOG(FATAL) << "Fail to decode delta length, status: " << st;
+ }
+ DCHECK_EQ(ret, num_length);
+ _length_idx = 0;
+ _num_valid_values = num_length;
+}
+template <tparquet::Type::type PhysicalType>
+Status DeltaLengthByteArrayDecoder<PhysicalType>::_get_internal(Slice* buffer,
int max_values,
+ int*
out_num_values) {
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = 0;
+ return Status::OK();
+ }
+
+ int32_t data_size = 0;
+ const int32_t* length_ptr = _buffered_length.data() + _length_idx;
+ for (int i = 0; i < max_values; ++i) {
+ int32_t len = length_ptr[i];
+ if (PREDICT_FALSE(len < 0)) {
+ return Status::InvalidArgument("Negative string delta length");
+ }
+ buffer[i].size = len;
+ if (common::add_overflow(data_size, len, data_size)) {
+ return Status::InvalidArgument("Excess expansion in
DELTA_(LENGTH_)BYTE_ARRAY");
+ }
+ }
+ _length_idx += max_values;
+
+ _buffered_data.resize(data_size);
+ char* data_ptr = _buffered_data.data();
+ for (int j = 0; j < data_size; j++) {
+ if (!_bit_reader->GetValue(8, data_ptr + j)) {
+ return Status::IOError("Get length bytes EOF");
+ }
+ }
+
+ for (int i = 0; i < max_values; ++i) {
+ buffer[i].data = data_ptr;
+ data_ptr += buffer[i].size;
+ }
+ // this->num_values_ -= max_values;
+ _num_valid_values -= max_values;
+ *out_num_values = max_values;
+ return Status::OK();
+}
+
+template <tparquet::Type::type PhysicalType>
+Status DeltaByteArrayDecoder<PhysicalType>::_get_internal(Slice* buffer, int
max_values,
+ int* out_num_values)
{
+ // Decode up to `max_values` strings into an internal buffer
+ // and reference them into `buffer`.
+ max_values = std::min(max_values, _num_valid_values);
+ if (max_values == 0) {
+ *out_num_values = max_values;
+ return Status::OK();
+ }
+
+ int suffix_read;
+ RETURN_IF_ERROR(_suffix_decoder.decode(buffer, max_values, &suffix_read));
+ if (PREDICT_FALSE(suffix_read != max_values)) {
+ return Status::IOError("Read {}, expecting {} from suffix decoder",
+ std::to_string(suffix_read),
std::to_string(max_values));
+ }
+
+ int64_t data_size = 0;
+ const int32_t* prefix_len_ptr = _buffered_prefix_length.data() +
_prefix_len_offset;
+ for (int i = 0; i < max_values; ++i) {
+ if (PREDICT_FALSE(prefix_len_ptr[i] < 0)) {
+ return Status::InvalidArgument("negative prefix length in
DELTA_BYTE_ARRAY");
+ }
+ if (PREDICT_FALSE(common::add_overflow(data_size,
static_cast<int64_t>(prefix_len_ptr[i]),
+ data_size) ||
+ common::add_overflow(data_size,
static_cast<int64_t>(buffer[i].size),
+ data_size))) {
+ return Status::InvalidArgument("excess expansion in
DELTA_BYTE_ARRAY");
+ }
+ }
+ _buffered_data.resize(data_size);
+
+ std::string_view prefix {_last_value};
+
+ char* data_ptr = _buffered_data.data();
Review Comment:
warning: variable 'data_ptr' is not initialized
[cppcoreguidelines-init-variables]
```suggestion
char* data_ptr = nullptr = _buffered_data.data();
```
--
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-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
