Skip to content

[feature](schema change) unified schema change for parquet and orc reader#32873

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
AshinGau merged 8 commits into from
Apr 10, 2024
Merged

[feature](schema change) unified schema change for parquet and orc reader #32873

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
42e2107
[feature](schema change)
AshinGau Mar 20, 2024
bdbd9f7
fix test
AshinGau Apr 8, 2024
75edfb6
fix2
AshinGau Apr 8, 2024
96822be
fix decimal
AshinGau Apr 9, 2024
68eb80c
fix time to string
AshinGau Apr 10, 2024
b74138b
Merge branch 'master' into uni_change
AshinGau Apr 10, 2024
6a3e4cb
Merge branch 'master' into uni_change
AshinGau Apr 10, 2024
6c5b607
Merge branch 'master' into uni_change
AshinGau Apr 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 1 addition & 8 deletions be/src/vec/data_types/data_type_struct.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -355,14 +355,7 @@ std::optional<size_t> DataTypeStruct::try_get_position_by_name(const String& nam
}

String DataTypeStruct::get_name_by_position(size_t i) const {
if (i == 0 || i > names.size()) {

@morningmanmorningmanApr 7, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why removing this check?

@AshinGauAshinGauApr 8, 2024

Copy link
Copy Markdown
MemberAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This function is wrong, and there's no check like Block::get_by_position

fmt::memory_buffer error_msg;
fmt::format_to(error_msg, "Index of tuple element ({}) if out range ([1, {}])", i,
names.size());
LOG(FATAL) << fmt::to_string(error_msg);
}

return names[i - 1];
return names[i];
}

int64_t DataTypeStruct::get_uncompressed_serialized_bytes(const IColumn& column,
Expand Down
332 changes: 332 additions & 0 deletions be/src/vec/exec/format/column_type_convert.cpp
View file
Open in desktop

Large diffs are not rendered by default.

537 changes: 537 additions & 0 deletions be/src/vec/exec/format/column_type_convert.h
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion be/src/vec/exec/format/format_common.h
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -30,7 +30,7 @@ struct DecimalScaleParams {
SCALE_DOWN,
};
ScaleType scale_type = ScaleType::NOT_INIT;
int32_t scale_factor = 1;
int64_t scale_factor = 1;

template <typename DecimalPrimitiveType>
static inline constexpr DecimalPrimitiveType get_scale_factor(int32_t n) {
Expand Down
187 changes: 130 additions & 57 deletions be/src/vec/exec/format/orc/vorc_reader.cpp
View file
Open in desktop

Large diffs are not rendered by default.

18 changes: 14 additions & 4 deletions be/src/vec/exec/format/orc/vorc_reader.h
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -46,6 +46,7 @@
#include "vec/columns/column_array.h"
#include "vec/data_types/data_type.h"
#include "vec/data_types/data_type_nullable.h"
#include "vec/exec/format/column_type_convert.h"
#include "vec/exec/format/format_common.h"
#include "vec/exec/format/generic_reader.h"
#include "vec/exec/format/table/transactional_hive_reader.h"
Expand DownExpand Up@@ -190,6 +191,9 @@ class OrcReader : public GenericReader {
std::unordered_map<std::string, orc::StringDictionary*>& column_name_to_dict_map,
bool* is_stripe_filtered);

static TypeDescriptor convert_to_doris_type(const orc::Type* orc_type);
static std::string get_field_name_lower_case(const orc::Type* orc_type, int pos);

protected:
void _collect_profile_before_close() override;

Expand DownExpand Up@@ -255,15 +259,20 @@ class OrcReader : public GenericReader {
bool* is_hive1_orc);
static bool _check_acid_schema(const orc::Type& type);
static const orc::Type& _remove_acid(const orc::Type& type);
TypeDescriptor _convert_to_doris_type(const orc::Type* orc_type);
bool _init_search_argument(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);
void _init_bloom_filter(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);
void _init_system_properties();
void _init_file_description();

template <bool is_filter = false>
Status _fill_doris_data_column(const std::string& col_name, MutableColumnPtr& data_column,
const DataTypePtr& data_type, const orc::Type* orc_column_type,
orc::ColumnVectorBatch* cvb, size_t num_values);

template <bool is_filter = false>
Status _orc_column_to_doris_column(const std::string& col_name, const ColumnPtr& doris_column,
Status _orc_column_to_doris_column(const std::string& col_name, ColumnPtr& doris_column,
const DataTypePtr& data_type,
const orc::Type* orc_column_type,
orc::ColumnVectorBatch* cvb, size_t num_values);
Expand DownExpand Up@@ -470,8 +479,6 @@ class OrcReader : public GenericReader {
orc::DataBuffer<int64_t>& orc_offsets, size_t num_values,
size_t* element_size);

std::string _get_field_name_lower_case(const orc::Type* orc_type, int pos);

void _collect_profile_on_close();

bool _can_filter_by_dict(int slot_id);
Expand DownExpand Up@@ -576,6 +583,9 @@ class OrcReader : public GenericReader {
bool _is_dict_cols_converted;
bool _has_complex_type = false;
std::vector<orc::TypeKind>* _unsupported_pushdown_types;

// resolve schema change
std::unordered_map<std::string, std::unique_ptr<converter::ColumnTypeConverter>> _converters;
};

class ORCFileInputStream : public orc::InputStream, public ProfileCollector {
Expand Down
41 changes: 11 additions & 30 deletions be/src/vec/exec/format/parquet/decoder.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -43,22 +43,12 @@ Status Decoder::get_decoder(tparquet::Type::type type, tparquet::Encoding::type
decoder.reset(new ByteArrayPlainDecoder());
break;
case tparquet::Type::INT32:
decoder.reset(new FixLengthPlainDecoder<tparquet::Type::INT32>());
break;
case tparquet::Type::INT64:
decoder.reset(new FixLengthPlainDecoder<tparquet::Type::INT64>());
break;
case tparquet::Type::INT96:
decoder.reset(new FixLengthPlainDecoder<tparquet::Type::INT96>());
break;
case tparquet::Type::FLOAT:
decoder.reset(new FixLengthPlainDecoder<tparquet::Type::FLOAT>());
break;
case tparquet::Type::DOUBLE:
decoder.reset(new FixLengthPlainDecoder<tparquet::Type::DOUBLE>());
break;
case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
decoder.reset(new FixLengthPlainDecoder<tparquet::Type::FIXED_LEN_BYTE_ARRAY>());
decoder.reset(new FixLengthPlainDecoder());
break;
default:
return Status::InternalError("Unsupported type {}(encoding={}) in parquet decoder",
Expand All@@ -73,22 +63,12 @@ Status Decoder::get_decoder(tparquet::Type::type type, tparquet::Encoding::type
decoder.reset(new ByteArrayDictDecoder());
break;
case tparquet::Type::INT32:
decoder.reset(new FixLengthDictDecoder<tparquet::Type::INT32>());
break;
case tparquet::Type::INT64:
decoder.reset(new FixLengthDictDecoder<tparquet::Type::INT64>());
break;
case tparquet::Type::INT96:
decoder.reset(new FixLengthDictDecoder<tparquet::Type::INT96>());
break;
case tparquet::Type::FLOAT:
decoder.reset(new FixLengthDictDecoder<tparquet::Type::FLOAT>());
break;
case tparquet::Type::DOUBLE:
decoder.reset(new FixLengthDictDecoder<tparquet::Type::DOUBLE>());
break;
case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
decoder.reset(new FixLengthDictDecoder<tparquet::Type::FIXED_LEN_BYTE_ARRAY>());
decoder.reset(new FixLengthDictDecoder());
break;
default:
return Status::InternalError("Unsupported type {}(encoding={}) in parquet decoder",
Expand All@@ -109,10 +89,10 @@ Status Decoder::get_decoder(tparquet::Type::type type, tparquet::Encoding::type
// Supports only INT32 and INT64.
switch (type) {
case tparquet::Type::INT32:
decoder.reset(new DeltaBitPackDecoder<int32, tparquet::Type::INT32>());
decoder.reset(new DeltaBitPackDecoder<int32>());
break;
case tparquet::Type::INT64:
decoder.reset(new DeltaBitPackDecoder<int64, tparquet::Type::INT64>());
decoder.reset(new DeltaBitPackDecoder<int64>());
break;
default:
return Status::InternalError("DELTA_BINARY_PACKED only supports INT32 and INT64");
Expand All@@ -121,20 +101,21 @@ Status Decoder::get_decoder(tparquet::Type::type type, tparquet::Encoding::type
case tparquet::Encoding::DELTA_BYTE_ARRAY:
switch (type) {
case tparquet::Type::BYTE_ARRAY:
decoder.reset(new DeltaByteArrayDecoder<tparquet::Type::BYTE_ARRAY>());
case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
decoder.reset(new DeltaByteArrayDecoder());
break;
default:
return Status::InternalError("DELTA_BYTE_ARRAY only supports BYTE_ARRAY.");
return Status::InternalError(
"DELTA_BYTE_ARRAY only supports BYTE_ARRAY, FIXED_LEN_BYTE_ARRAY.");
}
break;
case tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY:
switch (type) {
case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
decoder.reset(new DeltaLengthByteArrayDecoder<tparquet::Type::FIXED_LEN_BYTE_ARRAY>());
case tparquet::Type::BYTE_ARRAY:
decoder.reset(new DeltaLengthByteArrayDecoder());
break;
default:
return Status::InternalError(
"DELTA_LENGTH_BYTE_ARRAY only supports FIXED_LEN_BYTE_ARRAY.");
return Status::InternalError("DELTA_LENGTH_BYTE_ARRAY only supports BYTE_ARRAY.");
}
break;
default:
Expand Down
Loading