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) #50415

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
dataroaring merged 3 commits into from
May 6, 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) #50415

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8fcd74a
[fix](parquet)Fixed the problem that when Parquert reader use index t…
hubgeter Apr 18, 2025
e90f602
fix format
hubgeter Apr 25, 2025
0a00faf
fix build
hubgeter Apr 27, 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 @@ -55,7 +55,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 @@ -71,7 +71,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();

Expand All @@ -85,7 +85,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;
};

} // namespace doris::vectorized
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 @@ -79,7 +79,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;
};

} // namespace doris::vectorized
4 changes: 2 additions & 2 deletions 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 @@ -279,7 +279,7 @@ Status OrcReader::_create_file_reader() {

Status OrcReader::init_reader(
const std::vector<std::string>* 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 @@ -692,7 +692,7 @@ bool static build_search_argument(std::vector<OrcPredicate>& predicates, int ind
}

bool OrcReader::_init_search_argument(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
if ((!_enable_filter_by_min_max) || colname_to_value_range->empty()) {
return false;
}
Expand Down
6 changes: 3 additions & 3 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 @@ -142,7 +142,7 @@ class OrcReader : public GenericReader {
//If you want to read the file by index instead of column name, set hive_use_column_names to false.
Status init_reader(
const std::vector<std::string>* 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 @@ -291,7 +291,7 @@ class OrcReader : public GenericReader {
static bool _check_acid_schema(const orc::Type& type);
static const orc::Type& _remove_acid(const orc::Type& type);
bool _init_search_argument(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range);
const 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();
Expand Down Expand Up @@ -598,7 +598,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, int 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 @@ -295,7 +295,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 @@ -346,7 +346,6 @@ Status ParquetReader::init_reader(
_missing_cols.emplace_back(name);
}
} 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 @@ -360,21 +359,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 {
int32_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 @@ -96,7 +96,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;
};

} // namespace doris::vectorized
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 @@ -530,7 +530,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<int, 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 @@ -603,7 +603,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<int, 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
6 changes: 3 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 @@ -150,7 +150,7 @@ class IcebergTableReader : public TableFormatReader {
std::unordered_map<std::string, std::string> _file_col_to_table_col;
// 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;
const std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
// copy from _colname_to_value_range with new column name that is in parquet/orc file, to support schema evolution.
std::unordered_map<std::string, ColumnValueRangeType> _new_colname_to_value_range;
// column id to name map. Collect from FE slot descriptor.
Expand Down Expand Up @@ -205,7 +205,7 @@ class IcebergParquetReader final : public IcebergTableReader {
Status init_reader(
const std::vector<std::string>& file_col_names,
const std::unordered_map<int, 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 @@ -251,7 +251,7 @@ class IcebergOrcReader final : public IcebergTableReader {
Status init_reader(
const std::vector<std::string>& file_col_names,
const std::unordered_map<int, 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 @@ -74,7 +74,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 @@ -57,14 +57,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;
};
} // namespace doris::vectorized
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 @@ -100,7 +100,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 @@ -65,13 +65,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;
};

} // namespace doris::vectorized
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 @@ -92,7 +92,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 @@ -64,10 +64,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;
};

} // namespace doris::vectorized
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/table/transactional_hive_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ TransactionalHiveReader::TransactionalHiveReader(std::unique_ptr<GenericReader>

Status TransactionalHiveReader::init_reader(
const std::vector<std::string>& 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 VExprContextSPtrs* not_single_slot_filter_conjuncts,
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/table/transactional_hive_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ class TransactionalHiveReader : public TableFormatReader {

Status init_reader(
const std::vector<std::string>& 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 VExprContextSPtrs* not_single_slot_filter_conjuncts,
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/exec/format/table/trino_connector_jni_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ TrinoConnectorJniReader::TrinoConnectorJniReader(
}

Status TrinoConnectorJniReader::init_reader(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
const std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
RETURN_IF_ERROR(_jni_connector->init(colname_to_value_range));
RETURN_IF_ERROR(_set_spi_plugins_dir());
return _jni_connector->open(_state, _profile);
Expand Down
Loading
Loading