This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1b226ff8a2 [refactor](load) remove FlushContext from SegmentWriter
(#21596)
1b226ff8a2 is described below
commit 1b226ff8a2c80870a1e71155e54df6178158d753
Author: Kaijie Chen <c...@apache.org>
AuthorDate: Sat Jul 8 22:44:56 2023 +0800
[refactor](load) remove FlushContext from SegmentWriter (#21596)
* [refactor](load) remove FlushContext from SegmentWriter
* remove unused imports
---
be/src/olap/rowset/beta_rowset_writer.cpp | 13 +++++++----
be/src/olap/rowset/segment_v2/segment_writer.cpp | 28 +++++++-----------------
be/src/olap/rowset/segment_v2/segment_writer.h | 7 ++----
3 files changed, 19 insertions(+), 29 deletions(-)
diff --git a/be/src/olap/rowset/beta_rowset_writer.cpp
b/be/src/olap/rowset/beta_rowset_writer.cpp
index d2b314ff3a..cc2aea5ac5 100644
--- a/be/src/olap/rowset/beta_rowset_writer.cpp
+++ b/be/src/olap/rowset/beta_rowset_writer.cpp
@@ -818,15 +818,20 @@ Status BetaRowsetWriter::_do_create_segment_writer(
}
_segcompaction_worker.get_file_writer().reset(file_writer.release());
} else {
+ const auto& tablet_schema = flush_ctx && flush_ctx->flush_schema ?
flush_ctx->flush_schema
+ :
_context.tablet_schema;
+ if (flush_ctx && flush_ctx->block &&
+ flush_ctx->block->bytes() <=
config::segment_compression_threshold_kb * 1024) {
+ writer_options.compression_type = NO_COMPRESSION;
+ }
writer->reset(new segment_v2::SegmentWriter(
- file_writer.get(), segment_id, _context.tablet_schema,
_context.tablet,
- _context.data_dir, _context.max_rows_per_segment,
writer_options,
- _context.mow_context));
+ file_writer.get(), segment_id, tablet_schema, _context.tablet,
_context.data_dir,
+ _context.max_rows_per_segment, writer_options,
_context.mow_context));
{
std::lock_guard<SpinLock> l(_lock);
_file_writers.push_back(std::move(file_writer));
}
- auto s = (*writer)->init(flush_ctx);
+ auto s = (*writer)->init();
if (!s.ok()) {
LOG(WARNING) << "failed to init segment writer: " << s.to_string();
writer->reset(nullptr);
diff --git a/be/src/olap/rowset/segment_v2/segment_writer.cpp
b/be/src/olap/rowset/segment_v2/segment_writer.cpp
index d339b324a7..372cd66352 100644
--- a/be/src/olap/rowset/segment_v2/segment_writer.cpp
+++ b/be/src/olap/rowset/segment_v2/segment_writer.cpp
@@ -36,9 +36,7 @@
#include "olap/key_coder.h"
#include "olap/olap_common.h"
#include "olap/primary_key_index.h"
-#include "olap/row_cursor.h" // IWYU pragma: keep
-#include "olap/row_cursor.h" // RowCursor
-#include "olap/rowset/rowset_writer.h"
+#include "olap/row_cursor.h" // RowCursor // IWYU pragma:
keep
#include "olap/rowset/rowset_writer_context.h" // RowsetWriterContext
#include "olap/rowset/segment_v2/column_writer.h" // ColumnWriter
#include "olap/rowset/segment_v2/page_io.h"
@@ -118,31 +116,25 @@ void SegmentWriter::init_column_meta(ColumnMetaPB* meta,
uint32_t column_id,
}
}
-Status SegmentWriter::init(const FlushContext* flush_ctx) {
+Status SegmentWriter::init() {
std::vector<uint32_t> column_ids;
int column_cnt = _tablet_schema->num_columns();
- if (flush_ctx && flush_ctx->flush_schema) {
- column_cnt = flush_ctx->flush_schema->num_columns();
- }
for (uint32_t i = 0; i < column_cnt; ++i) {
column_ids.emplace_back(i);
}
- return init(column_ids, true, flush_ctx);
+ return init(column_ids, true);
}
-Status SegmentWriter::init(const std::vector<uint32_t>& col_ids, bool has_key,
- const FlushContext* flush_ctx) {
+Status SegmentWriter::init(const std::vector<uint32_t>& col_ids, bool has_key)
{
DCHECK(_column_writers.empty());
DCHECK(_column_ids.empty());
_has_key = has_key;
_column_writers.reserve(_tablet_schema->columns().size());
_column_ids.insert(_column_ids.end(), col_ids.begin(), col_ids.end());
_olap_data_convertor =
std::make_unique<vectorized::OlapBlockDataConvertor>();
- _opts.compression_type =
- (flush_ctx == nullptr || flush_ctx->block == nullptr ||
- flush_ctx->block->bytes() >
config::segment_compression_threshold_kb * 1024)
- ? _tablet_schema->compression_type()
- : NO_COMPRESSION;
+ if (_opts.compression_type == UNKNOWN_COMPRESSION) {
+ _opts.compression_type = _tablet_schema->compression_type();
+ }
auto create_column_writer = [&](uint32_t cid, const auto& column) -> auto {
ColumnWriterOptions opts;
opts.meta = _footer.add_columns();
@@ -242,11 +234,7 @@ Status SegmentWriter::init(const std::vector<uint32_t>&
col_ids, bool has_key,
return Status::OK();
};
- if (flush_ctx && flush_ctx->flush_schema) {
- RETURN_IF_ERROR(_create_writers(*flush_ctx->flush_schema, col_ids,
create_column_writer));
- } else {
- RETURN_IF_ERROR(_create_writers(*_tablet_schema, col_ids,
create_column_writer));
- }
+ RETURN_IF_ERROR(_create_writers(*_tablet_schema, col_ids,
create_column_writer));
// we don't need the short key index for unique key merge on write table.
if (_has_key) {
diff --git a/be/src/olap/rowset/segment_v2/segment_writer.h
b/be/src/olap/rowset/segment_v2/segment_writer.h
index 31f652190c..222977c28a 100644
--- a/be/src/olap/rowset/segment_v2/segment_writer.h
+++ b/be/src/olap/rowset/segment_v2/segment_writer.h
@@ -35,7 +35,6 @@
#include "gutil/macros.h"
#include "gutil/strings/substitute.h"
#include "olap/olap_define.h"
-#include "olap/rowset/rowset_writer.h"
#include "olap/rowset/segment_v2/column_writer.h"
#include "olap/tablet.h"
#include "olap/tablet_schema.h"
@@ -58,7 +57,6 @@ class ShortKeyIndexBuilder;
class PrimaryKeyIndexBuilder;
class KeyCoder;
struct RowsetWriterContext;
-struct FlushContext;
namespace io {
class FileWriter;
@@ -89,11 +87,10 @@ public:
std::shared_ptr<MowContext> mow_context);
~SegmentWriter();
- Status init(const FlushContext* flush_ctx = nullptr);
+ Status init();
// for vertical compaction
- Status init(const std::vector<uint32_t>& col_ids, bool has_key,
- const FlushContext* flush_ctx = nullptr);
+ Status init(const std::vector<uint32_t>& col_ids, bool has_key);
template <typename RowType>
Status append_row(const RowType& row);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
