add defensive code for small sstable optimization

mittest/mtlenv/storage/blocksstable/test_index_tree.cpp

@@ -498,6 +498,7 @@ void TestIndexTree::prepare_index_desc(ObDataStoreDesc &index_desc)
 {
   int ret = OB_SUCCESS;
   ret = index_desc.init(index_schema_, ObLSID(1), ObTabletID(1), MAJOR_MERGE);
+  index_desc.major_working_cluster_version_ = DATA_VERSION_4_0_0_0;
   ASSERT_EQ(OB_SUCCESS, ret);
 }
 

src/storage/blocksstable/ob_index_block_builder.cpp

@@ -424,7 +424,8 @@ bool ObSSTableIndexBuilder::check_index_desc(const ObDataStoreDesc &index_desc)
   if (!index_desc.is_valid()
       || index_desc.need_prebuild_bloomfilter_
       || index_desc.merge_info_ != nullptr
-      || index_desc.row_column_count_ != index_desc.rowkey_column_count_ + 1) {
+      || index_desc.row_column_count_ != index_desc.rowkey_column_count_ + 1
+      || (index_desc.is_major_merge() && index_desc.major_working_cluster_version_ < DATA_VERSION_4_0_0_0)) {
     ret = false;
   }
   return ret;

src/storage/blocksstable/ob_index_block_builder.h

@@ -427,6 +427,7 @@ class ObSSTableIndexBuilder final
 public:
   enum ObSpaceOptimizationMode
   {
+    // TODO zhouxinlan.zxl : delete the mode DISABLE
     ENABLE = 0,   // enable the optimization for small sstable with given occupy_size
     DISABLE = 1,  // disable the optimization
     AUTO = 2      // enable the optimization without giving occupy_size (users don't know/assign occupy_size)

src/storage/ddl/ob_complement_data_task.cpp

@@ -245,7 +245,9 @@ int ObComplementDataContext::init(const ObComplementDataParam &param, const ObDa
   } else if (OB_ISNULL(index_builder_ = new (builder_buf) ObSSTableIndexBuilder())) {
     ret = OB_ERR_UNEXPECTED;
     LOG_WARN("failed to new ObSSTableIndexBuilder", K(ret));
-  } else if (OB_FAIL(index_builder_->init(desc))) {
+  } else if (OB_FAIL(index_builder_->init(desc,
+                                          nullptr, // macro block flush callback
+                                          ObSSTableIndexBuilder::DISABLE))) {
     LOG_WARN("failed to init index builder", K(ret), K(desc));
   } else {
     is_major_sstable_exist_ = nullptr != latest_major_sstable ? true : false;

src/storage/ddl/ob_direct_insert_sstable_ctx.cpp

@@ -737,7 +737,9 @@ int ObSSTableInsertTabletContext::prepare_index_builder_if_need(const ObTableSch
       } else if (OB_ISNULL(index_builder_ = new (builder_buf) ObSSTableIndexBuilder())) {
         ret = OB_ERR_UNEXPECTED;
         LOG_WARN("failed to new ObSSTableIndexBuilder", K(ret));
-      } else if (OB_FAIL(index_builder_->init(data_desc))) {
+      } else if (OB_FAIL(index_builder_->init(data_desc,
+                                              nullptr, // macro block flush callback
+                                              ObSSTableIndexBuilder::DISABLE))) {
         LOG_WARN("failed to init index builder", K(ret), K(data_desc));
       }
       if (OB_FAIL(ret)) {

unittest/storage/blocksstable/ob_multi_version_sstable_test.h

@@ -452,7 +452,7 @@ void ObMultiVersionSSTableTest::reset_writer(const int64_t snapshot_version)
   data_desc_.row_store_type_ = row_store_type_;
   ASSERT_TRUE(data_desc_.is_valid());
 
-  ASSERT_EQ(OB_SUCCESS, index_desc_.init(index_schema_, ls_id, tablet_id, merge_type_, snapshot_version, 1000000));
+  ASSERT_EQ(OB_SUCCESS, index_desc_.init(index_schema_, ls_id, tablet_id, merge_type_, snapshot_version, DATA_VERSION_4_1_0_0));
   ASSERT_TRUE(index_desc_.is_valid());
   ASSERT_EQ(OB_SUCCESS, root_index_builder_->init(index_desc_));
 

unittest/storage/test_major_rows_merger.cpp

@@ -156,6 +156,7 @@ void ObMajorRowsMergerTest::prepare_merge_context(const ObMergeType &merge_type,
 
   ASSERT_EQ(OB_SUCCESS, merge_context.init_merge_info());
   ASSERT_EQ(OB_SUCCESS, merge_context.merge_info_.prepare_index_builder(index_desc_));
+  index_desc_.major_working_cluster_version_ = DATA_VERSION_4_0_0_0;
 }