Skip to content

[fix](parquet)Fixed the problem that when Parquert reader use index to read files, there will be multiple threads modify same object #50161

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
morningman merged 4 commits into from
Apr 23, 2025
Merged

[fix](parquet)Fixed the problem that when Parquert reader use index to read files, there will be multiple threads modify same object #50161

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
04f246d
[fix](parquet)Fixed the problem that when Parquert reader use index t…
hubgeter Apr 18, 2025
f6db881
fix build
hubgeter Apr 18, 2025
598ec8e
fix be ur
hubgeter Apr 18, 2025
22e44d4
fix build
hubgeter Apr 18, 2025
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
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/avro/avro_jni_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ Status AvroJNIReader::get_columns(std::unordered_map<std::string, TypeDescriptor
}

Status AvroJNIReader::init_fetch_table_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
_colname_to_value_range = colname_to_value_range;
std::ostringstream required_fields;
std::ostringstream columns_types;
Expand Down
4 changes: 2 additions & 2 deletions be/src/vec/exec/format/avro/avro_jni_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ class AvroJNIReader : public JniReader {
std::unordered_set<std::string>* missing_cols) override;

Status init_fetch_table_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);

TFileType::type get_file_type() const;

Expand All @@ -82,7 +82,7 @@ class AvroJNIReader : public JniReader {
private:
const TFileScanRangeParams _params;
const TFileRangeDesc _range;
std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range = nullptr;
const std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range = nullptr;
};

#include "common/compile_check_end.h"
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/jni_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ Status MockJniReader::get_columns(std::unordered_map<std::string, TypeDescriptor
}

Status MockJniReader::init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
_colname_to_value_range = colname_to_value_range;
RETURN_IF_ERROR(_jni_connector->init(colname_to_value_range));
return _jni_connector->open(_state, _profile);
Expand Down
4 changes: 2 additions & 2 deletions be/src/vec/exec/format/jni_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ class MockJniReader : public JniReader {
std::unordered_set<std::string>* missing_cols) override;

Status init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);

Status close() override {
if (_jni_connector) {
Expand All @@ -106,7 +106,7 @@ class MockJniReader : public JniReader {
}

private:
std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
const std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
};

#include "common/compile_check_end.h"
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/orc/vorc_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,7 @@ Status OrcReader::_create_file_reader() {
Status OrcReader::init_reader(
const std::vector<std::string>* column_names,
const std::vector<std::string>& missing_column_names,
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const VExprContextSPtrs& conjuncts, bool is_acid, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const VExprContextSPtrs* not_single_slot_filter_conjuncts,
Expand Down
4 changes: 2 additions & 2 deletions be/src/vec/exec/format/orc/vorc_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ class OrcReader : public GenericReader {
Status init_reader(
const std::vector<std::string>* column_names,
const std::vector<std::string>& missing_column_names,
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const VExprContextSPtrs& conjuncts, bool is_acid,
const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
const VExprContextSPtrs* not_single_slot_filter_conjuncts,
Expand Down Expand Up @@ -631,7 +631,7 @@ class OrcReader : public GenericReader {
std::vector<DecimalScaleParams> _decimal_scale_params;
size_t _decimal_scale_params_index;

std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
const std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range = nullptr;
bool _is_acid = false;
std::unique_ptr<IColumn::Filter> _filter;
LazyReadContext _lazy_read_ctx;
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/parquet/vparquet_page_index.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ Status PageIndex::create_skipped_row_range(tparquet::OffsetIndex& offset_index,
}

Status PageIndex::collect_skipped_page_range(tparquet::ColumnIndex* column_index,
ColumnValueRangeType& col_val_range,
const ColumnValueRangeType& col_val_range,
const FieldSchema* col_schema,
std::vector<int>& skipped_ranges,
const cctz::time_zone& ctz) {
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/parquet/vparquet_page_index.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ class PageIndex {
Status create_skipped_row_range(tparquet::OffsetIndex& offset_index,
int64_t total_rows_of_group, int page_idx, RowRange* row_range);
Status collect_skipped_page_range(tparquet::ColumnIndex* column_index,
ColumnValueRangeType& col_val_range,
const ColumnValueRangeType& col_val_range,
const FieldSchema* col_schema,
std::vector<int>& skipped_ranges, const cctz::time_zone& ctz);
bool check_and_get_page_index_ranges(const std::vector<tparquet::ColumnChunk>& columns);
Expand Down
19 changes: 6 additions & 13 deletions be/src/vec/exec/format/parquet/vparquet_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -297,7 +297,7 @@ void ParquetReader::iceberg_sanitize(const std::vector<std::string>& read_column
Status ParquetReader::init_reader(
const std::vector<std::string>& all_column_names,
const std::vector<std::string>& missing_column_names,
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const std::unordered_map<std::string, int>* colname_to_slot_id,
Expand Down Expand Up @@ -349,7 +349,6 @@ Status ParquetReader::init_reader(
}

} else {
std::unordered_map<std::string, ColumnValueRangeType> new_colname_to_value_range;
const auto& table_column_idxs = _scan_params.column_idxs;
std::map<int, int> table_col_id_to_idx;
for (int i = 0; i < table_column_idxs.size(); i++) {
Expand All @@ -363,21 +362,15 @@ Status ParquetReader::init_reader(
auto& table_col = all_column_names[idx];
auto file_col = schema_desc.get_column(id)->name;
_read_columns.emplace_back(file_col);
_table_col_to_file_col[table_col] = file_col;

if (table_col != file_col) {
_table_col_to_file_col[table_col] = file_col;
auto iter = _colname_to_value_range->find(table_col);
if (iter != _colname_to_value_range->end()) {
continue;
}
new_colname_to_value_range[file_col] = iter->second;
_colname_to_value_range->erase(iter->first);
auto iter = _colname_to_value_range->find(table_col);
if (iter != _colname_to_value_range->end()) {
_colname_to_value_range_index_read.emplace(file_col, iter->second);
}
}
}
for (auto it : new_colname_to_value_range) {
_colname_to_value_range->emplace(it.first, std::move(it.second));
}
_colname_to_value_range = &_colname_to_value_range_index_read;
}
// build column predicates for column lazy read
_lazy_read_ctx.conjuncts = conjuncts;
Expand Down
9 changes: 7 additions & 2 deletions be/src/vec/exec/format/parquet/vparquet_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ class ParquetReader : public GenericReader {
Status init_reader(
const std::vector<std::string>& all_column_names,
const std::vector<std::string>& missing_column_names,
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const std::unordered_map<std::string, int>* colname_to_slot_id,
Expand Down Expand Up @@ -250,7 +250,12 @@ class ParquetReader : public GenericReader {
size_t _total_groups; // num of groups(stripes) of a parquet(orc) file
// table column name to file column name map. For iceberg schema evolution.
std::unordered_map<std::string, std::string> _table_col_to_file_col;
std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range = nullptr;
const std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range = nullptr;

// During initialization, multiple vfile_scanner's _colname_to_value_range will point to the same object,
// so the content in the object cannot be modified (there is a multi-threading problem).
// _colname_to_value_range_index_read used when _hive_use_column_names = false.
std::unordered_map<std::string, ColumnValueRangeType> _colname_to_value_range_index_read;
std::vector<std::string> _read_columns;
RowRange _whole_range = RowRange(0, 0);
const std::vector<int64_t>* _delete_rows = nullptr;
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/table/hudi_jni_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ Status HudiJniReader::get_columns(std::unordered_map<std::string, TypeDescriptor
}

Status HudiJniReader::init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
_colname_to_value_range = colname_to_value_range;
RETURN_IF_ERROR(_jni_connector->init(colname_to_value_range));
return _jni_connector->open(_state, _profile);
Expand Down
4 changes: 2 additions & 2 deletions be/src/vec/exec/format/table/hudi_jni_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,12 +58,12 @@ class HudiJniReader : public JniReader {
std::unordered_set<std::string>* missing_cols) override;

Status init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);

private:
const TFileScanRangeParams& _scan_params;
const THudiFileDesc& _hudi_params;
std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
const std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
};

#include "common/compile_check_end.h"
Expand Down
4 changes: 2 additions & 2 deletions be/src/vec/exec/format/table/hudi_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ Status HudiReader::get_next_block_inner(Block* block, size_t* read_rows, bool* e
Status HudiOrcReader::init_reader(
const std::vector<std::string>& read_table_col_names,
const std::unordered_map<int32_t, std::string>& table_col_id_table_name_map,
std::unordered_map<std::string, ColumnValueRangeType>* table_col_name_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>* table_col_name_to_value_range,
const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const VExprContextSPtrs* not_single_slot_filter_conjuncts,
Expand All @@ -71,7 +71,7 @@ Status HudiOrcReader::init_reader(
Status HudiParquetReader::init_reader(
const std::vector<std::string>& read_table_col_names,
const std::unordered_map<int32_t, std::string>& table_col_id_table_name_map,
std::unordered_map<std::string, ColumnValueRangeType>* table_col_name_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>* table_col_name_to_value_range,
const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const std::unordered_map<std::string, int>* colname_to_slot_id,
Expand Down
6 changes: 4 additions & 2 deletions be/src/vec/exec/format/table/hudi_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@ class HudiOrcReader final : public HudiReader {
Status init_reader(
const std::vector<std::string>& read_table_col_names,
const std::unordered_map<int32_t, std::string>& table_col_id_table_name_map,
std::unordered_map<std::string, ColumnValueRangeType>* table_col_name_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>*
table_col_name_to_value_range,
const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const VExprContextSPtrs* not_single_slot_filter_conjuncts,
Expand All @@ -72,7 +73,8 @@ class HudiParquetReader final : public HudiReader {
Status init_reader(
const std::vector<std::string>& read_table_col_names,
const std::unordered_map<int32_t, std::string>& table_col_id_table_name_map,
std::unordered_map<std::string, ColumnValueRangeType>* table_col_name_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>*
table_col_name_to_value_range,
const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const std::unordered_map<std::string, int>* colname_to_slot_id,
Expand Down
4 changes: 2 additions & 2 deletions be/src/vec/exec/format/table/iceberg_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -420,7 +420,7 @@ void IcebergTableReader::_gen_position_delete_file_range(Block& block, DeleteFil
Status IcebergParquetReader::init_reader(
const std::vector<std::string>& file_col_names,
const std::unordered_map<int32_t, std::string>& col_id_name_map,
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const std::unordered_map<std::string, int>* colname_to_slot_id,
Expand Down Expand Up @@ -490,7 +490,7 @@ Status IcebergParquetReader ::_read_position_delete_file(const TFileRangeDesc* d
Status IcebergOrcReader::init_reader(
const std::vector<std::string>& file_col_names,
const std::unordered_map<int32_t, std::string>& col_id_name_map,
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const std::unordered_map<std::string, int>* colname_to_slot_id,
Expand Down
5 changes: 2 additions & 3 deletions be/src/vec/exec/format/table/iceberg_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,7 +131,6 @@ class IcebergTableReader : public TableFormatReader, public TableSchemaChangeHel
ShardedKVCache* _kv_cache;
IcebergProfile _iceberg_profile;
std::vector<int64_t> _iceberg_delete_rows;

std::vector<std::string> _expand_col_names;
std::vector<ColumnWithTypeAndName> _expand_columns;

Expand Down Expand Up @@ -170,7 +169,7 @@ class IcebergParquetReader final : public IcebergTableReader {
Status init_reader(
const std::vector<std::string>& file_col_names,
const std::unordered_map<int32_t, std::string>& col_id_name_map,
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const std::unordered_map<std::string, int>* colname_to_slot_id,
Expand Down Expand Up @@ -217,7 +216,7 @@ class IcebergOrcReader final : public IcebergTableReader {
Status init_reader(
const std::vector<std::string>& file_col_names,
const std::unordered_map<int32_t, std::string>& col_id_name_map,
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
const RowDescriptor* row_descriptor,
const std::unordered_map<std::string, int>* colname_to_slot_id,
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/table/lakesoul_jni_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ Status LakeSoulJniReader::get_columns(std::unordered_map<std::string, TypeDescri
}

Status LakeSoulJniReader::init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
_colname_to_value_range = colname_to_value_range;
RETURN_IF_ERROR(_jni_connector->init(colname_to_value_range));
return _jni_connector->open(_state, _profile);
Expand Down
4 changes: 2 additions & 2 deletions be/src/vec/exec/format/table/lakesoul_jni_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,14 +58,14 @@ class LakeSoulJniReader : public ::doris::vectorized::GenericReader {
std::unordered_set<std::string>* missing_cols) override;

Status init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);

private:
const TLakeSoulFileDesc& _lakesoul_params;
const std::vector<SlotDescriptor*>& _file_slot_descs;
RuntimeState* _state;
RuntimeProfile* _profile;
std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
const std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
std::unique_ptr<::doris::vectorized::JniConnector> _jni_connector;
};
#include "common/compile_check_end.h"
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/table/max_compute_jni_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ Status MaxComputeJniReader::get_columns(
}

Status MaxComputeJniReader::init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
_colname_to_value_range = colname_to_value_range;
RETURN_IF_ERROR(_jni_connector->init(colname_to_value_range));
return _jni_connector->open(_state, _profile);
Expand Down
4 changes: 2 additions & 2 deletions be/src/vec/exec/format/table/max_compute_jni_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,13 +63,13 @@ class MaxComputeJniReader : public JniReader {
std::unordered_set<std::string>* missing_cols) override;

Status init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);

private:
const MaxComputeTableDescriptor* _table_desc = nullptr;
const TMaxComputeFileDesc& _max_compute_params;
const TFileRangeDesc& _range;
std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range = nullptr;
const std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range = nullptr;
};

#include "common/compile_check_end.h"
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/table/paimon_jni_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,7 @@ Status PaimonJniReader::get_columns(std::unordered_map<std::string, TypeDescript
}

Status PaimonJniReader::init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
_colname_to_value_range = colname_to_value_range;
RETURN_IF_ERROR(_jni_connector->init(colname_to_value_range));
return _jni_connector->open(_state, _profile);
Expand Down
4 changes: 2 additions & 2 deletions be/src/vec/exec/format/table/paimon_jni_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,10 +63,10 @@ class PaimonJniReader : public JniReader {
std::unordered_set<std::string>* missing_cols) override;

Status init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);

private:
std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
const std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
int64_t _remaining_table_level_row_count;
};

Expand Down
Loading
Loading