Merge branch 'dag_patch_lite' into 'master'

Co-authored-by: Charles0429 <[email protected]>
Co-authored-by: simonjoylet <[email protected]>
Co-authored-by: a1iive <[email protected]>

Author: 4 people

deps/oblib/src/lib/list/ob_dlist.h

@@ -106,6 +106,7 @@ class ObDList
   DLinkNode *remove_first();
 
   void push_range(ObDList<DLinkNode> &range);
+  void push_back_range(ObDList<DLinkNode> &range);
   void pop_range(int32_t num, ObDList<DLinkNode> &range);
 
   //the list is empty or not
@@ -292,6 +293,24 @@ void ObDList<DLinkNode>::push_range(ObDList<DLinkNode> &range)
   }
 }
 template <typename DLinkNode>
+void ObDList<DLinkNode>::push_back_range(ObDList<DLinkNode> &range)
+{
+  if (!range.is_empty()) {
+    if (is_empty()) {
+      push_range(range);
+    } else {
+      DLinkNode *tail = header_.prev_;
+      DLinkNode *first = range.header_.next_;
+      DLinkNode *last = range.header_.prev_;
+      first->prev_ = NULL;
+      last->next_ = NULL;
+      tail->add_range_after(first, last);
+      size_ += range.get_size();
+      range.reset();
+    }
+  }
+}
+template <typename DLinkNode>
 void ObDList<DLinkNode>::pop_range(int32_t num, ObDList<DLinkNode> &range)
 {
   DLinkNode *first = this->header_.next_;

deps/oblib/src/lib/ob_errno.h

@@ -438,6 +438,7 @@ constexpr int OB_ERR_DUPLICATE_INDEX = -9137;
 constexpr int OB_INVALID_KMS_DEST = -9139;
 constexpr int OB_OBJECT_STORAGE_OBJECT_LOCKED_BY_WORM = -9140;
 constexpr int OB_OBJECT_STORAGE_OVERWRITE_CONTENT_MISMATCH = -9142;
+constexpr int OB_DAG_TASK_IS_SUSPENDED = -9143;
 constexpr int OB_ERR_XML_PARSE = -9549;
 constexpr int OB_ERR_XSLT_PARSE = -9574;
 constexpr int OB_HDFS_CONNECT_FS_ERROR = -11019;

deps/oblib/src/lib/queue/ob_lighty_queue.h

@@ -99,6 +99,7 @@ class LightyQueue
   bool is_inited() const { return queue_.is_inited(); }
   int push(void *data, const int64_t timeout = 0);
   int pop(void *&data, const int64_t timeout = 0);
+  int64_t to_string(char* buf, const int64_t buf_len) const { return 0; }
 private:
   typedef ObOrderedFixedQueue<void> Queue;
   Queue queue_;

deps/oblib/src/lib/wait_event/ob_wait_event.h

@@ -125,6 +125,8 @@ WAIT_EVENT_DEF(ASYNC_COMMITTING_WAIT, 16018, "tx commiting wait", "", "", "", CO
 WAIT_EVENT_DEF(OBCDC_PART_MGR_SCHEMA_VERSION_WAIT, 18000, "oblog part mgr schema version wait", "", "", "", CONCURRENCY, true, true)
 WAIT_EVENT_DEF(BACKUP_TMP_FILE_WAIT, 18001, "backup tmp file wait", "", "", "", CONCURRENCY, true, true)
 WAIT_EVENT_DEF(BACKUP_TMP_FILE_QUEUE_WAIT, 18002, "backup tmp file queue wait", "", "", "", CONCURRENCY, true, true)
+WAIT_EVENT_DEF(CG_ROW_TMP_FILE_WAIT, 18003, "cg row file wait", "", "", "", CONCURRENCY, true, true)
+WAIT_EVENT_DEF(CG_BLOCK_TMP_FILE_WAIT, 18004, "cg block file wait", "", "", "", CONCURRENCY, true, true)
 WAIT_EVENT_DEF(SYNC_GET_GTS_WAIT, 18101, "sync get gts timestamp wait", "address", "", "", NETWORK, true, true)
 WAIT_EVENT_DEF(LOCK_FOR_READ_WAIT, 18102, "sleep: lock for read need wait for concurrency control", "sleep_interval", "", "", CONCURRENCY, true, true)
 WAIT_EVENT_DEF(DEADLOCK_MGR_BUCKET_LOCK, 18103, "latch: deadlock manager bucket access wait", "spin count", "", "", CONCURRENCY, true, true)

mittest/mtlenv/storage/access/test_co_merge.cpp

@@ -375,6 +375,7 @@ void TestCOMerge::prepare_co_sstable(
                           DATA_CURRENT_VERSION,
                           table_schema.get_micro_index_clustered(),
                           0 /*tablet_transfer_seq*/,
+                          0 /*concurrent_cnt*/,
                           share::SCN::invalid_scn(),
                           &cg_schema,
                           i));
@@ -2465,6 +2466,7 @@ TEST_F(TestCOMerge, test_rebuild_sstable)
 
   //prepare merge_ctx
   prepare_merge_context(MAJOR_MERGE, false, trans_version_range, merge_context);
+  merge_context.static_desc_.concurrent_cnt_ = 2;
 	merge_context.array_count_ = 3;
   alloc_merge_infos(merge_context);
   OK(merge_context.prepare_index_builder(0, 3));

mittest/mtlenv/storage/blocksstable/CMakeLists.txt

@@ -18,4 +18,6 @@ storage_dml_unittest(test_skip_index_sortedness)
 storage_dml_unittest(test_truncate_info_kv_cache)
 target_link_libraries(test_truncate_info_kv_cache PUBLIC truncate_info_helper)
 storage_dml_unittest(test_ob_block_writer_concurrent_guard)
+storage_dml_unittest(test_dag_macro_writer)
+storage_dml_unittest(test_macro_block_row_bare_iterator)
 storage_dml_unittest(test_micro_hash_index)

mittest/mtlenv/storage/blocksstable/ob_index_block_data_prepare.h

@@ -623,7 +623,8 @@ void TestIndexBlockDataPrepare::prepare_data(const int64_t micro_block_size)
   scn.convert_for_tx(SNAPSHOT_VERSION);
   ObWholeDataStoreDesc desc;
   ASSERT_EQ(OB_SUCCESS, desc.init(false/*is_ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_), merge_type_, SNAPSHOT_VERSION,
-                                  DATA_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/, scn));
+                                  DATA_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/,
+                                  0/*concurrent_cnt*/, scn));
   desc.get_desc().static_desc_->schema_version_ = 10;
   void *builder_buf = allocator_.alloc(sizeof(ObSSTableIndexBuilder));
   root_index_builder_ = new (builder_buf) ObSSTableIndexBuilder(false /* not need writer buffer*/);
@@ -672,7 +673,8 @@ void TestIndexBlockDataPrepare::prepare_discontinuous_data(const int64_t micro_b
   scn.convert_for_tx(SNAPSHOT_VERSION);
   ObWholeDataStoreDesc desc;
   ASSERT_EQ(OB_SUCCESS, desc.init(false/*is_ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_), merge_type_, SNAPSHOT_VERSION,
-                                  DATA_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/, scn));
+                                  DATA_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/, 0/*concurrent_cnt*/,
+                                  scn));
   desc.get_desc().static_desc_->schema_version_ = 10;
   void *builder_buf = allocator_.alloc(sizeof(ObSSTableIndexBuilder));
   root_index_builder_ = new (builder_buf) ObSSTableIndexBuilder(false /* not need writer buffer*/);
@@ -744,7 +746,8 @@ void TestIndexBlockDataPrepare::prepare_cg_data()
   ObWholeDataStoreDesc desc;
   share::SCN scn;
   scn.convert_for_tx(SNAPSHOT_VERSION);
-  ASSERT_EQ(OB_SUCCESS, desc.init(false/*is_ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_), merge_type_, SNAPSHOT_VERSION, DATA_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/, scn));
+  ASSERT_EQ(OB_SUCCESS, desc.init(false/*is_ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_), merge_type_,
+    SNAPSHOT_VERSION, DATA_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/, 0/*concurrent_cnt*/, scn));
   ObIArray<ObColDesc> &col_descs = desc.get_desc().col_desc_->col_desc_array_;
   for (int64_t i = 0; i < col_descs.count(); ++i) {
     if (col_descs.at(i).col_type_.type_ == ObIntType) {
@@ -766,8 +769,7 @@ void TestIndexBlockDataPrepare::prepare_cg_data()
   OK(data_desc.init(false/*is_ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_),
                     merge_type_, SNAPSHOT_VERSION, DATA_CURRENT_VERSION,
                     table_schema_.get_micro_index_clustered(),
-                    0 /*transfer_seq*/,
-                    scn, &cg_schema, 0));
+                    0 /*transfer_seq*/, 0/*concurrent_cnt*/, scn, &cg_schema, 0));
   data_desc.get_desc().static_desc_->schema_version_ = 10;
   void *builder_buf = allocator_.alloc(sizeof(ObSSTableIndexBuilder));
   root_index_builder_ = new (builder_buf) ObSSTableIndexBuilder(false /* not need writer buffer*/);
@@ -811,7 +813,9 @@ void TestIndexBlockDataPrepare::prepare_ddl_memtable()
 
   share::SCN ddl_start_scn;
   ddl_start_scn.convert_from_ts(ObTimeUtility::current_time());
-  ASSERT_EQ(OB_SUCCESS, ddl_memtable_.init(allocator_, *tablet_handle.get_obj(), sstable_.get_key(), ddl_start_scn, DATA_CURRENT_VERSION));
+  ObStorageSchema *storage_schema = nullptr;
+  ASSERT_EQ(OB_SUCCESS, tablet_handle.get_obj()->load_storage_schema(allocator_, storage_schema));
+  ASSERT_EQ(OB_SUCCESS, ddl_memtable_.init(allocator_, *tablet_handle.get_obj(), sstable_.get_key(), ddl_start_scn, DATA_CURRENT_VERSION, storage_schema));
 
   SMART_VAR(ObSSTableSecMetaIterator, meta_iter) {
     ObDatumRange query_range;
@@ -942,9 +946,12 @@ void TestIndexBlockDataPrepare::prepare_partial_ddl_data()
   ObMacroBlockWriter writer;
   row_generate_.reset();
   ObWholeDataStoreDesc desc;
+  const share::SCN reorganization_scn(share::SCN::min_scn());
   share::SCN end_scn;
   end_scn.convert_from_ts(ObTimeUtility::current_time());
-  ASSERT_EQ(OB_SUCCESS, desc.init(true/*is ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_), merge_type_, SNAPSHOT_VERSION, CLUSTER_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/, end_scn));
+  ASSERT_EQ(OB_SUCCESS, desc.init(true/*is ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_),
+      merge_type_, SNAPSHOT_VERSION, CLUSTER_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/,
+      0/*concurrent_cnt*/, end_scn));
   void *builder_buf = allocator_.alloc(sizeof(ObSSTableIndexBuilder));
   merge_root_index_builder_ = new (builder_buf) ObSSTableIndexBuilder(false /* not need writer buffer */);
   ASSERT_NE(nullptr, merge_root_index_builder_);
@@ -1019,7 +1026,9 @@ void TestIndexBlockDataPrepare::prepare_partial_cg_data()
   ObWholeDataStoreDesc desc;
   share::SCN end_scn;
   end_scn.convert_from_ts(ObTimeUtility::current_time());
-  ASSERT_EQ(OB_SUCCESS, desc.init(true/*is ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_), merge_type_, SNAPSHOT_VERSION, CLUSTER_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/, end_scn));
+  ASSERT_EQ(OB_SUCCESS, desc.init(true/*is ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_),
+      merge_type_, SNAPSHOT_VERSION, CLUSTER_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/,
+      0/*concurrent_cnt*/, end_scn));
   void *builder_buf = allocator_.alloc(sizeof(ObSSTableIndexBuilder));
   merge_root_index_builder_ = new (builder_buf) ObSSTableIndexBuilder(false /* not need index buffer */);
   ASSERT_NE(nullptr, merge_root_index_builder_);
@@ -1256,7 +1265,9 @@ void TestIndexBlockDataPrepare::prepare_contrastive_sstable()
   ObWholeDataStoreDesc desc;
   share::SCN end_scn;
   end_scn.convert_from_ts(ObTimeUtility::current_time());
-  ASSERT_EQ(OB_SUCCESS, desc.init(false/*is_ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_), merge_type_, SNAPSHOT_VERSION, CLUSTER_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/, end_scn));
+  ASSERT_EQ(OB_SUCCESS, desc.init(false/*is_ddl*/, table_schema_, ObLSID(ls_id_), ObTabletID(tablet_id_),
+      merge_type_, SNAPSHOT_VERSION, CLUSTER_CURRENT_VERSION, table_schema_.get_micro_index_clustered(), 0 /*transfer_seq*/,
+      0/*concurrent_cnt*/, end_scn));
   void *builder_buf = allocator_.alloc(sizeof(ObSSTableIndexBuilder));
   root_index_builder_ = new (builder_buf) ObSSTableIndexBuilder(false /* not need writer buffer*/);
   ASSERT_NE(nullptr, root_index_builder_);
@@ -1317,13 +1328,19 @@ void TestIndexBlockDataPrepare::prepare_merge_ddl_kvs()
   ObITable::TableKey ddl_key = sstable_.get_key();
   ddl_key.table_type_ = ObITable::TableType::MAJOR_SSTABLE;
   for (int64_t i = 0; i < DDL_KVS_CNT; ++i) {
-    void *buf = allocator_.alloc(sizeof(ObDDLMemtable));
+    void *buf = ddl_kv_handle_.get_obj()->ddl_memtable_allocator_.alloc(sizeof(ObDDLMemtable));
     ASSERT_NE(nullptr, buf);
     ObDDLMemtable *new_ddl_table = new (buf) ObDDLMemtable;
     ddl_key.slice_range_.start_slice_idx_ = 0;
     ddl_key.slice_range_.end_slice_idx_ = 0;
-    ASSERT_EQ(OB_SUCCESS, new_ddl_table->init(allocator_, *tablet_handle.get_obj(), ddl_key, ddl_start_scn, 4000));
+    ObStorageSchema *storage_schema = nullptr;
+    void *buf2 = ddl_kv_handle_.get_obj()->ddl_memtable_allocator_.alloc(sizeof(ObArenaAllocator));
+    ASSERT_NE(nullptr, buf2);
+    ObArenaAllocator *allocator = new (buf2) ObArenaAllocator();
+    ASSERT_EQ(OB_SUCCESS, tablet_handle.get_obj()->load_storage_schema(allocator_, storage_schema));
+    ASSERT_EQ(OB_SUCCESS, new_ddl_table->init(*allocator, *tablet_handle.get_obj(), ddl_key, ddl_start_scn, 4000, storage_schema));
     ASSERT_EQ(OB_SUCCESS, ddl_kv_handle_.get_obj()->get_ddl_memtables().push_back(new_ddl_table));
+    ASSERT_EQ(OB_SUCCESS, ddl_kv_handle_.get_obj()->get_ddl_memtable_allocators().push_back(allocator));
   }
   ObDDLKVHandle kv_handle;
   ObDDLKvMgrHandle ddl_kv_mgr_handle;

mittest/mtlenv/storage/blocksstable/test_clustered_index_writer.cpp

@@ -85,7 +85,7 @@ void TestClusteredIndexWriter::prepare_data_store_desc(
                        ObTimeUtility::fast_current_time() /*snapshot_version*/,
                        DATA_CURRENT_VERSION,
                        table_schema_.get_micro_index_clustered(),
-                       0 /*transfer_seq*/);
+                       0 /*transfer_seq*/, 0/*concurrent_cnt*/);
   data_desc.get_desc().sstable_index_builder_ = sstable_index_builder;
   ASSERT_EQ(OB_SUCCESS, ret);
 }