Skip to content
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

[BugFix] Fix memory limit exceeded problem when writing a partitioned Parquet table #43672

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
+39 −1
Open

[BugFix] Fix memory limit exceeded problem when writing a partitioned Parquet table #43672

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1752079
flush parquet file
mxdzs0612 Apr 7, 2024
b9a0eb3
fix ut
mxdzs0612 Apr 9, 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
Next Next commit
flush parquet file 
Signed-off-by: Jiao Mingye <[email protected]>
  • Loading branch information
@mxdzs0612
mxdzs0612 committed Apr 10, 2024
commit 175207924e98e797162c7cd68bc0f1e967b6926b
3 changes: 3 additions & 0 deletions be/src/common/config.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1258,6 +1258,9 @@ CONF_mInt64(pk_dump_interval_seconds, "3600"); // 1 hour
// Min data processed when scaling connector sink writers, default value is the same as Trino
CONF_mInt64(writer_scaling_min_size_mb, "128");

// Memory threshold for Parquet writer before a flush, should be greater than 0 and less than 1
CONF_mDouble(parquet_writer_mem_usage_threshold, "0.8");

// whether enable query profile for queries initiated by spark or flink
CONF_mBool(enable_profile_for_external_plan, "false");

Expand Down
9 changes: 8 additions & 1 deletion be/src/formats/parquet/parquet_file_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
#include "column/struct_column.h"
#include "column/vectorized_fwd.h"
#include "common/logging.h"
#include "exec/pipeline/query_context.h"
#include "exprs/expr.h"
#include "formats/file_writer.h"
#include "formats/utils.h"
Expand All @@ -48,7 +49,13 @@ std::future<Status> ParquetFileWriter::write(ChunkPtr chunk) {
if (auto status = _rowgroup_writer->write(chunk.get()); !status.ok()) {
return make_ready_future(std::move(status));
}
if (_rowgroup_writer->estimated_buffered_bytes() >= _writer_options->rowgroup_size) {
double mem_usage = 0.0;
Copy link
Contributor

Choose a reason for hiding this comment

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

It's better to implement this in ConnectorSinkOperator so all file formats can benefit from this

if (_runtime_state != nullptr) {
mem_usage = static_cast<double>(_runtime_state->query_mem_tracker_ptr()->consumption()) /
_runtime_state->query_ctx()->get_static_query_mem_limit();
}
if (mem_usage > config::parquet_writer_mem_usage_threshold ||
_rowgroup_writer->estimated_buffered_bytes() >= _writer_options->rowgroup_size) {
return _flush_row_group();
}
return make_ready_future(Status::OK());
Expand Down