Mow master 0922 tmp 4 rebase

.github/workflows/auto-pr-reply.yml

@@ -1,3 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
 name: Auto Reply to PR
 
 on:
@@ -8,11 +25,16 @@ jobs:
  comment:
  runs-on: ubuntu-latest
  steps:
- - name: Comment on PR
- uses: ./.github/actions/create-or-update-comment
- with:
- issue-number: ${{ github.event.pull_request.number }}
- body: |
- Thank you for your contribution to Apache Doris.
-
- Don't know what should be done next? See [How to process your PR](https://cwiki.apache.org/confluence/display/DORIS/How+to+process+your+PR)
+ - name: Checkout
+ uses: actions/checkout@v3
+ with:
+ persist-credentials: false
+ submodules: recursive
+ - name: Comment on PR
+ uses: ./.github/actions/create-or-update-comment
+ with:
+ issue-number: ${{ github.event.pull_request.number }}
+ body: |
+ Thank you for your contribution to Apache Doris.
+
+ Don't know what should be done next? See [How to process your PR](https://cwiki.apache.org/confluence/display/DORIS/How+to+process+your+PR)

.gitmodules

@@ -26,4 +26,4 @@
  branch = clucene
 [submodule ".github/actions/create-or-update-comment"]
  path = .github/actions/create-or-update-comment
- url = https://github.com/peter-evans/create-or-update-comment
+ url = https://github.com/peter-evans/create-or-update-comment.git

be/src/common/config.cpp

@@ -505,7 +505,8 @@ DEFINE_Int64(stream_tvf_buffer_size, "1048576"); // 1MB
 
 // OlapTableSink sender's send interval, should be less than the real response time of a tablet writer rpc.
 // You may need to lower the speed when the sink receiver bes are too busy.
-DEFINE_mInt32(olap_table_sink_send_interval_ms, "1");
+DEFINE_mInt32(olap_table_sink_send_interval_microseconds, "1000");
+DEFINE_mDouble(olap_table_sink_send_interval_auto_partition_factor, "0.001");
 
 // Fragment thread pool
 DEFINE_Int32(fragment_pool_thread_num_min, "64");
@@ -1067,6 +1068,8 @@ DEFINE_mInt64(lookup_connection_cache_bytes_limit, "4294967296");
 DEFINE_mInt64(LZ4_HC_compression_level, "9");
 
 DEFINE_mBool(enable_merge_on_write_correctness_check, "true");
+// rowid conversion correctness check when compaction for mow table
+DEFINE_mBool(enable_rowid_conversion_correctness_check, "false");
 
 // The secure path with user files, used in the `local` table function.
 DEFINE_mString(user_files_secure_path, "${DORIS_HOME}");

be/src/common/config.h

@@ -559,7 +559,9 @@ DECLARE_Int64(stream_tvf_buffer_size);
 
 // OlapTableSink sender's send interval, should be less than the real response time of a tablet writer rpc.
 // You may need to lower the speed when the sink receiver bes are too busy.
-DECLARE_mInt32(olap_table_sink_send_interval_ms);
+DECLARE_mInt32(olap_table_sink_send_interval_microseconds);
+// For auto partition, the send interval will multiply the factor
+DECLARE_mDouble(olap_table_sink_send_interval_auto_partition_factor);
 
 // Fragment thread pool
 DECLARE_Int32(fragment_pool_thread_num_min);
@@ -1128,6 +1130,8 @@ DECLARE_mBool(enable_flatten_nested_for_variant);
 DECLARE_mDouble(ratio_of_defaults_as_sparse_column);
 
 DECLARE_mBool(enable_merge_on_write_correctness_check);
+// rowid conversion correctness check when compaction for mow table
+DECLARE_mBool(enable_rowid_conversion_correctness_check);
 
 // The secure path with user files, used in the `local` table function.
 DECLARE_mString(user_files_secure_path);

be/src/io/fs/buffered_reader.cpp

@@ -54,7 +54,7 @@ Status MergeRangeFileReader::read_at_impl(size_t offset, Slice result, size_t* b
  Status st = _reader->read_at(offset, result, bytes_read, io_ctx);
  _statistics.merged_io++;
  _statistics.request_bytes += *bytes_read;
- _statistics.read_bytes += *bytes_read;
+ _statistics.merged_bytes += *bytes_read;
  return st;
  }
  if (offset + result.size > _random_access_ranges[range_index].end_offset) {
@@ -68,10 +68,10 @@ Status MergeRangeFileReader::read_at_impl(size_t offset, Slice result, size_t* b
  if (cached_data.contains(offset)) {
  // has cached data in box
  _read_in_box(cached_data, offset, result, &has_read);
+ _statistics.request_bytes += has_read;
  if (has_read == result.size) {
  // all data is read in cache
  *bytes_read = has_read;
- _statistics.request_bytes += has_read;
  return Status::OK();
  }
  } else if (!cached_data.empty()) {
@@ -91,7 +91,7 @@ Status MergeRangeFileReader::read_at_impl(size_t offset, Slice result, size_t* b
  *bytes_read = has_read + read_size;
  _statistics.merged_io++;
  _statistics.request_bytes += read_size;
- _statistics.read_bytes += read_size;
+ _statistics.merged_bytes += read_size;
  return Status::OK();
  }
 
@@ -186,7 +186,7 @@ Status MergeRangeFileReader::read_at_impl(size_t offset, Slice result, size_t* b
  *bytes_read = has_read + read_size;
  _statistics.merged_io++;
  _statistics.request_bytes += read_size;
- _statistics.read_bytes += read_size;
+ _statistics.merged_bytes += read_size;
  return Status::OK();
  }
 
@@ -314,7 +314,7 @@ Status MergeRangeFileReader::_fill_box(int range_index, size_t start_offset, siz
  RETURN_IF_ERROR(
  _reader->read_at(start_offset, Slice(_read_slice, to_read), bytes_read, io_ctx));
  _statistics.merged_io++;
- _statistics.read_bytes += *bytes_read;
+ _statistics.merged_bytes += *bytes_read;
  }
 
  SCOPED_RAW_TIMER(&_statistics.copy_time);

be/src/io/fs/buffered_reader.h

@@ -80,7 +80,8 @@ class MergeRangeFileReader : public io::FileReader {
  int64_t request_io = 0;
  int64_t merged_io = 0;
  int64_t request_bytes = 0;
- int64_t read_bytes = 0;
+ int64_t merged_bytes = 0;
+ int64_t apply_bytes = 0;
  };
 
  struct RangeCachedData {
@@ -147,6 +148,9 @@ class MergeRangeFileReader : public io::FileReader {
  // Equivalent min size of each IO that can reach the maximum storage speed limit:
  // 512KB for oss, 4KB for hdfs
  _equivalent_io_size = _is_oss ? OSS_MIN_IO_SIZE : HDFS_MIN_IO_SIZE;
+ for (const PrefetchRange& range : _random_access_ranges) {
+ _statistics.apply_bytes += range.end_offset - range.start_offset;
+ }
  if (_profile != nullptr) {
  const char* random_profile = "MergedSmallIO";
  ADD_TIMER_WITH_LEVEL(_profile, random_profile, 1);
@@ -158,8 +162,10 @@ class MergeRangeFileReader : public io::FileReader {
  random_profile, 1);
  _request_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "RequestBytes", TUnit::BYTES,
  random_profile, 1);
- _read_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "MergedBytes", TUnit::BYTES,
- random_profile, 1);
+ _merged_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "MergedBytes", TUnit::BYTES,
+ random_profile, 1);
+ _apply_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "ApplyBytes", TUnit::BYTES,
+ random_profile, 1);
  }
  }
 
@@ -184,7 +190,8 @@ class MergeRangeFileReader : public io::FileReader {
  COUNTER_UPDATE(_request_io, _statistics.request_io);
  COUNTER_UPDATE(_merged_io, _statistics.merged_io);
  COUNTER_UPDATE(_request_bytes, _statistics.request_bytes);
- COUNTER_UPDATE(_read_bytes, _statistics.read_bytes);
+ COUNTER_UPDATE(_merged_bytes, _statistics.merged_bytes);
+ COUNTER_UPDATE(_apply_bytes, _statistics.apply_bytes);
  }
  }
  return Status::OK();
@@ -220,7 +227,8 @@ class MergeRangeFileReader : public io::FileReader {
  RuntimeProfile::Counter* _request_io;
  RuntimeProfile::Counter* _merged_io;
  RuntimeProfile::Counter* _request_bytes;
- RuntimeProfile::Counter* _read_bytes;
+ RuntimeProfile::Counter* _merged_bytes;
+ RuntimeProfile::Counter* _apply_bytes;
 
  int _search_read_range(size_t start_offset, size_t end_offset);
  void _clean_cached_data(RangeCachedData& cached_data);

be/src/olap/compaction.cpp

@@ -660,7 +660,8 @@ Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
  output_rowsets.push_back(_output_rowset);
 
  if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
- _tablet->enable_unique_key_merge_on_write()) {
+ _tablet->enable_unique_key_merge_on_write() &&
+ _tablet->tablet_schema()->cluster_key_idxes().empty()) {
  Version version = _tablet->max_version();
  DeleteBitmap output_rowset_delete_bitmap(_tablet->tablet_id());
  std::set<RowLocation> missed_rows;
@@ -689,7 +690,9 @@ Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
  }
  }
 
- RETURN_IF_ERROR(_tablet->check_rowid_conversion(_output_rowset, location_map));
+ if (config::enable_rowid_conversion_correctness_check) {
+ RETURN_IF_ERROR(_tablet->check_rowid_conversion(_output_rowset, location_map));
+ }
  location_map.clear();
 
  {
@@ -750,7 +753,9 @@ Status Compaction::modify_rowsets(const Merger::Statistics* stats) {
  }
  }
 
- RETURN_IF_ERROR(_tablet->check_rowid_conversion(_output_rowset, location_map));
+ if (config::enable_rowid_conversion_correctness_check) {
+ RETURN_IF_ERROR(_tablet->check_rowid_conversion(_output_rowset, location_map));
+ }
 
  _tablet->merge_delete_bitmap(output_rowset_delete_bitmap);
  RETURN_IF_ERROR(_tablet->modify_rowsets(output_rowsets, _input_rowsets, true));

be/src/olap/delete_bitmap_calculator.cpp

@@ -92,47 +92,54 @@ bool MergeIndexDeleteBitmapCalculatorContext::Comparator::operator()(
  Slice key1, key2;
  RETURN_IF_ERROR(lhs->get_current_key(key1));
  RETURN_IF_ERROR(rhs->get_current_key(key2));
- if (_sequence_length == 0) {
+ if (_sequence_length == 0 && _rowid_length == 0) {
  auto cmp_result = key1.compare(key2);
  // when key1 is the same as key2,
  // we want the one with greater segment id to be popped first
  return cmp_result ? (cmp_result > 0) : (lhs->segment_id() < rhs->segment_id());
  }
  // smaller key popped first
- auto key1_without_seq = Slice(key1.get_data(), key1.get_size() - _sequence_length);
- auto key2_without_seq = Slice(key2.get_data(), key2.get_size() - _sequence_length);
+ auto key1_without_seq =
+ Slice(key1.get_data(), key1.get_size() - _sequence_length - _rowid_length);
+ auto key2_without_seq =
+ Slice(key2.get_data(), key2.get_size() - _sequence_length - _rowid_length);
  auto cmp_result = key1_without_seq.compare(key2_without_seq);
  if (cmp_result != 0) {
  return cmp_result > 0;
  }
- // greater sequence value popped first
- auto key1_sequence_val =
- Slice(key1.get_data() + key1.get_size() - _sequence_length, _sequence_length);
- auto key2_sequence_val =
- Slice(key2.get_data() + key2.get_size() - _sequence_length, _sequence_length);
- cmp_result = key1_sequence_val.compare(key2_sequence_val);
- if (cmp_result != 0) {
- return cmp_result < 0;
+ if (_sequence_length > 0) {
+ // greater sequence value popped first
+ auto key1_sequence_val =
+ Slice(key1.get_data() + key1_without_seq.get_size() + 1, _sequence_length - 1);
+ auto key2_sequence_val =
+ Slice(key2.get_data() + key2_without_seq.get_size() + 1, _sequence_length - 1);
+ cmp_result = key1_sequence_val.compare(key2_sequence_val);
+ if (cmp_result != 0) {
+ return cmp_result < 0;
+ }
  }
  // greater segment id popped first
  return lhs->segment_id() < rhs->segment_id();
 }
 
 bool MergeIndexDeleteBitmapCalculatorContext::Comparator::is_key_same(Slice const& lhs,
  Slice const& rhs) const {
- DCHECK(lhs.get_size() >= _sequence_length);
- DCHECK(rhs.get_size() >= _sequence_length);
- auto lhs_without_seq = Slice(lhs.get_data(), lhs.get_size() - _sequence_length);
- auto rhs_without_seq = Slice(rhs.get_data(), rhs.get_size() - _sequence_length);
+ DCHECK(lhs.get_size() >= _sequence_length + _rowid_length);
+ DCHECK(rhs.get_size() >= _sequence_length + _rowid_length);
+ auto lhs_without_seq = Slice(lhs.get_data(), lhs.get_size() - _sequence_length - _rowid_length);
+ auto rhs_without_seq = Slice(rhs.get_data(), rhs.get_size() - _sequence_length - _rowid_length);
  return lhs_without_seq.compare(rhs_without_seq) == 0;
 }
 
 Status MergeIndexDeleteBitmapCalculator::init(RowsetId rowset_id,
  std::vector<SegmentSharedPtr> const& segments,
- size_t seq_col_length, size_t max_batch_size) {
+ size_t seq_col_length, size_t rowdid_length,
+ size_t max_batch_size) {
  _rowset_id = rowset_id;
  _seq_col_length = seq_col_length;
- _comparator = MergeIndexDeleteBitmapCalculatorContext::Comparator(seq_col_length);
+ _rowid_length = rowdid_length;
+ _comparator =
+ MergeIndexDeleteBitmapCalculatorContext::Comparator(seq_col_length, _rowid_length);
  _contexts.reserve(segments.size());
  _heap = std::make_unique<Heap>(_comparator);
 
@@ -146,6 +153,10 @@ Status MergeIndexDeleteBitmapCalculator::init(RowsetId rowset_id,
  _contexts.emplace_back(std::move(index), index_type, segment->id(), pk_idx->num_rows());
  _heap->push(&_contexts.back());
  }
+ if (_rowid_length > 0) {
+ _rowid_coder = get_key_coder(
+ get_scalar_type_info<FieldType::OLAP_FIELD_TYPE_UNSIGNED_INT>()->type());
+ }
  return Status::OK();
 }
 
@@ -159,6 +170,15 @@ Status MergeIndexDeleteBitmapCalculator::calculate_one(RowLocation& loc) {
  if (!_last_key.empty() && _comparator.is_key_same(cur_key, _last_key)) {
  loc.segment_id = cur_ctx->segment_id();
  loc.row_id = cur_ctx->row_id();
+ if (_rowid_length > 0) {
+ Slice key_without_seq = Slice(cur_key.get_data(),
+ cur_key.get_size() - _seq_col_length - _rowid_length);
+ Slice rowid_slice =
+ Slice(cur_key.get_data() + key_without_seq.get_size() + _seq_col_length + 1,
+ _rowid_length - 1);
+ RETURN_IF_ERROR(_rowid_coder->decode_ascending(&rowid_slice, _rowid_length,
+ (uint8_t*)&loc.row_id));
+ }
  auto st = cur_ctx->advance();
  if (st.ok()) {
  _heap->push(cur_ctx);
@@ -176,8 +196,9 @@ Status MergeIndexDeleteBitmapCalculator::calculate_one(RowLocation& loc) {
  RETURN_IF_ERROR(nxt_ctx->get_current_key(nxt_key));
  Status st = _comparator.is_key_same(cur_key, nxt_key)
  ? cur_ctx->advance()
- : cur_ctx->seek_at_or_after(Slice(
- nxt_key.get_data(), nxt_key.get_size() - _seq_col_length));
+ : cur_ctx->seek_at_or_after(
+ Slice(nxt_key.get_data(),
+ nxt_key.get_size() - _seq_col_length - _rowid_length));
  if (st.is<ErrorCode::END_OF_FILE>()) {
  continue;
  }

be/src/olap/delete_bitmap_calculator.h

@@ -29,6 +29,7 @@
 #include "olap/base_tablet.h"
 #include "olap/binlog_config.h"
 #include "olap/data_dir.h"
+#include "olap/key_coder.h"
 #include "olap/olap_common.h"
 #include "olap/rowset/rowset.h"
 #include "olap/rowset/rowset_meta.h"
@@ -47,13 +48,15 @@ class MergeIndexDeleteBitmapCalculatorContext {
 public:
  class Comparator {
  public:
- Comparator(size_t sequence_length) : _sequence_length(sequence_length) {}
+ Comparator(size_t sequence_length, size_t rowid_length)
+ : _sequence_length(sequence_length), _rowid_length(rowid_length) {}
  bool operator()(MergeIndexDeleteBitmapCalculatorContext* lhs,
  MergeIndexDeleteBitmapCalculatorContext* rhs) const;
  bool is_key_same(Slice const& lhs, Slice const& rhs) const;
 
  private:
  size_t _sequence_length;
+ size_t _rowid_length;
  };
 
  MergeIndexDeleteBitmapCalculatorContext(std::unique_ptr<segment_v2::IndexedColumnIterator> iter,
@@ -90,7 +93,7 @@ class MergeIndexDeleteBitmapCalculator {
  MergeIndexDeleteBitmapCalculator() = default;
 
  Status init(RowsetId rowset_id, std::vector<SegmentSharedPtr> const& segments,
- size_t seq_col_length = 0, size_t max_batch_size = 1024);
+ size_t seq_col_length = 0, size_t rowid_length = 0, size_t max_batch_size = 1024);
 
  Status calculate_one(RowLocation& loc);
 
@@ -101,11 +104,13 @@ class MergeIndexDeleteBitmapCalculator {
  std::vector<MergeIndexDeleteBitmapCalculatorContext*>,
  MergeIndexDeleteBitmapCalculatorContext::Comparator>;
  std::vector<MergeIndexDeleteBitmapCalculatorContext> _contexts;
- MergeIndexDeleteBitmapCalculatorContext::Comparator _comparator {0};
+ MergeIndexDeleteBitmapCalculatorContext::Comparator _comparator {0, 0};
  RowsetId _rowset_id;
  std::unique_ptr<Heap> _heap;
  std::string _last_key;
  size_t _seq_col_length;
+ size_t _rowid_length;
+ const KeyCoder* _rowid_coder = nullptr;
 };
 
 } // namespace doris