diff --git a/src/meta/src/hummock/manager/commit_epoch.rs b/src/meta/src/hummock/manager/commit_epoch.rs
index cc06f65ffc8b1..6836e238ac284 100644
--- a/src/meta/src/hummock/manager/commit_epoch.rs
+++ b/src/meta/src/hummock/manager/commit_epoch.rs
@@ -12,13 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::ops::Deref;
 use std::sync::Arc;
 
 use risingwave_common::catalog::TableId;
 use risingwave_hummock_sdk::change_log::ChangeLogDelta;
-use risingwave_hummock_sdk::compaction_group::group_split::{self, split_sst};
+use risingwave_hummock_sdk::compaction_group::group_split::split_sst;
 use risingwave_hummock_sdk::sstable_info::SstableInfo;
 use risingwave_hummock_sdk::table_stats::{
     add_prost_table_stats_map, purge_prost_table_stats, to_prost_table_stats_map, PbTableStatsMap,
@@ -233,24 +233,8 @@ impl HummockManager {
             ));
         }
 
-        // Add new table
-        let mut group_members_table_ids: HashMap<u64, BTreeSet<TableId>> = HashMap::new();
-        {
-            // expand group_members_table_ids
-            for (table_id, group_id) in &table_compaction_group_mapping {
-                group_members_table_ids
-                    .entry(*group_id)
-                    .or_default()
-                    .insert(*table_id);
-            }
-        }
-
         let commit_sstables = self
-            .correct_commit_ssts(
-                sstables,
-                &table_compaction_group_mapping,
-                &group_members_table_ids,
-            )
+            .correct_commit_ssts(sstables, &table_compaction_group_mapping)
             .await?;
 
         let modified_compaction_groups: Vec<_> = commit_sstables.keys().cloned().collect();
@@ -402,7 +386,6 @@ impl HummockManager {
         &self,
         sstables: Vec<LocalSstableInfo>,
         table_compaction_group_mapping: &HashMap<TableId, CompactionGroupId>,
-        group_members_table_ids: &HashMap<CompactionGroupId, BTreeSet<TableId>>,
     ) -> Result<BTreeMap<CompactionGroupId, Vec<SstableInfo>>> {
         let mut new_sst_id_number = 0;
         let mut sst_to_cg_vec = Vec::with_capacity(sstables.len());
@@ -436,60 +419,18 @@ impl HummockManager {
         let mut new_sst_id = next_sstable_object_id(&self.env, new_sst_id_number).await?;
         let mut commit_sstables: BTreeMap<u64, Vec<SstableInfo>> = BTreeMap::new();
 
-        // for (sst, group_table_ids) in &mut sst_to_cg_vec {
-        //     let sst_size = sst.sst_info.sst_size;
-        //     for (group_id, match_ids) in group_table_ids {
-        //         let split_key = build_split_key_with_table_id(match_ids.last().unwrap() + 1);
-        //         let split_type = group_split::need_to_split(&sst.sst_info, split_key.clone());
-        //         match split_type {
-        //             group_split::SstSplitType::Left => {
-        //                 commit_sstables
-        //                     .entry(*group_id)
-        //                     .or_default()
-        //                     .push(sst.sst_info.clone());
-        //                 break;
-        //             }
-
-        //             group_split::SstSplitType::Right => {
-        //                 // do nothing
-        //                 break;
-        //             }
-
-        //             group_split::SstSplitType::Both => {
-        //                 let mut branch_sst = split_sst(
-        //                     &mut sst.sst_info,
-        //                     &mut new_sst_id,
-        //                     split_key.clone(),
-        //                     sst_size / 2,
-        //                     sst_size / 2,
-        //                 );
-
-        //                 // push the left sst to commit_sstables
-        //                 std::mem::swap(&mut sst.sst_info, &mut branch_sst);
-        //                 commit_sstables
-        //                     .entry(*group_id)
-        //                     .or_default()
-        //                     .push(branch_sst);
-        //             }
-        //         }
-        //     }
-        // }
         for (mut sst, group_table_ids) in sst_to_cg_vec {
-            for (group_id, match_ids) in group_table_ids {
-                let group_members_table_ids = group_members_table_ids.get(&group_id).unwrap();
-                if sst
-                    .sst_info
-                    .table_ids
-                    .iter()
-                    .all(|id| group_members_table_ids.contains(&TableId::new(*id)))
-                {
+            let len = group_table_ids.len();
+            for (index, (group_id, match_ids)) in group_table_ids.into_iter().enumerate() {
+                if sst.sst_info.table_ids == match_ids {
+                    // The SST contains all the tables in the group should be last key
+                    assert!(index == len - 1);
                     commit_sstables
                         .entry(group_id)
                         .or_default()
-                        .push(sst.sst_info.clone());
-                    continue;
+                        .push(sst.sst_info);
+                    break;
                 }
-
                 let origin_sst_size = sst.sst_info.sst_size;
                 let new_sst_size = match_ids
                     .iter()