[enhance](partition_id) check partition id before store meta

be/src/olap/data_dir.cpp

@@ -539,15 +539,23 @@ Status DataDir::load() {
                     StorageEngine::instance()->pending_local_rowsets().add(
                             rowset_meta->rowset_id()),
                     true);
-            if (!commit_txn_status && !commit_txn_status.is<PUSH_TRANSACTION_ALREADY_EXIST>()) {
-                LOG(WARNING) << "failed to add committed rowset: " << rowset_meta->rowset_id()
-                             << " to tablet: " << rowset_meta->tablet_id()
-                             << " for txn: " << rowset_meta->txn_id();
-            } else {
+            if (commit_txn_status || commit_txn_status.is<PUSH_TRANSACTION_ALREADY_EXIST>()) {
                 LOG(INFO) << "successfully to add committed rowset: " << rowset_meta->rowset_id()
                           << " to tablet: " << rowset_meta->tablet_id()
                           << " schema hash: " << rowset_meta->tablet_schema_hash()
                           << " for txn: " << rowset_meta->txn_id();
+
+            } else if (commit_txn_status.is<ErrorCode::INTERNAL_ERROR>()) {
+                LOG(WARNING) << "failed to add committed rowset: " << rowset_meta->rowset_id()
+                             << " to tablet: " << rowset_meta->tablet_id()
+                             << " for txn: " << rowset_meta->txn_id()
+                             << " error: " << commit_txn_status;
+                return commit_txn_status;
+            } else {
+                LOG(WARNING) << "failed to add committed rowset: " << rowset_meta->rowset_id()
+                             << " to tablet: " << rowset_meta->tablet_id()
+                             << " for txn: " << rowset_meta->txn_id()
+                             << " error: " << commit_txn_status;
             }
         } else if (rowset_meta->rowset_state() == RowsetStatePB::VISIBLE &&
                    rowset_meta->tablet_uid() == tablet->tablet_uid()) {

be/src/olap/rowset/rowset_meta_manager.cpp

@@ -88,6 +88,12 @@ Status RowsetMetaManager::get_json_rowset_meta(OlapMeta* meta, TabletUid tablet_
 }
 Status RowsetMetaManager::save(OlapMeta* meta, TabletUid tablet_uid, const RowsetId& rowset_id,
                                const RowsetMetaPB& rowset_meta_pb, bool enable_binlog) {
+    if (rowset_meta_pb.partition_id() <= 0) {
+        LOG(WARNING) << "invalid partition id " << rowset_meta_pb.partition_id() << " tablet "
+                     << rowset_meta_pb.tablet_id();
+        return Status::InternalError("invalid partition id {}, tablet {}",
+                                     rowset_meta_pb.partition_id(), rowset_meta_pb.tablet_id());
+    }
     if (enable_binlog) {
         return _save_with_binlog(meta, tablet_uid, rowset_id, rowset_meta_pb);
     } else {

be/src/olap/tablet_meta_manager.cpp

@@ -93,6 +93,12 @@ Status TabletMetaManager::save(DataDir* store, TTabletId tablet_id, TSchemaHash
     std::string key = fmt::format("{}{}_{}", header_prefix, tablet_id, schema_hash);
     std::string value;
     static_cast<void>(tablet_meta->serialize(&value));
+    if (tablet_meta->partition_id() <= 0) {
+        LOG(WARNING) << "invalid partition id " << tablet_meta->partition_id() << " tablet "
+                     << tablet_meta->tablet_id();
+        return Status::InternalError("invaid partition id {} tablet {}",
+                                     tablet_meta->partition_id(), tablet_meta->tablet_id());
+    }
     OlapMeta* meta = store->get_meta();
     VLOG_NOTICE << "save tablet meta"
                 << ", key:" << key << ", meta length:" << value.length();

be/src/olap/txn_manager.cpp

@@ -295,9 +295,10 @@ Status TxnManager::commit_txn(OlapMeta* meta, TPartitionId partition_id,
                               const RowsetSharedPtr& rowset_ptr, PendingRowsetGuard guard,
                               bool is_recovery) {
     if (partition_id < 1 || transaction_id < 1 || tablet_id < 1) {
-        LOG(FATAL) << "invalid commit req "
-                   << " partition_id=" << partition_id << " transaction_id=" << transaction_id
-                   << " tablet_id=" << tablet_id;
+        LOG(WARNING) << "invalid commit req "
+                     << " partition_id=" << partition_id << " transaction_id=" << transaction_id
+                     << " tablet_id=" << tablet_id;
+        return Status::InternalError("invalid partition id");
     }
 
     pair<int64_t, int64_t> key(partition_id, transaction_id);
@@ -383,10 +384,8 @@ Status TxnManager::commit_txn(OlapMeta* meta, TPartitionId partition_id,
             }
         });
         if (!save_status.ok()) {
-            return Status::Error<ROWSET_SAVE_FAILED>(
-                    "save committed rowset failed. when commit txn rowset_id: {}, tablet id: {}, "
-                    "txn id: {}",
-                    rowset_ptr->rowset_id().to_string(), tablet_id, transaction_id);
+            save_status.append(fmt::format(", txn id: {}", transaction_id));
+            return save_status;
         }
     }
 
@@ -537,10 +536,8 @@ Status TxnManager::publish_txn(OlapMeta* meta, TPartitionId partition_id,
                                           rowset->rowset_meta()->get_rowset_pb(), enable_binlog);
     stats->save_meta_time_us += MonotonicMicros() - t5;
     if (!status.ok()) {
-        return Status::Error<ROWSET_SAVE_FAILED>(
-                "save committed rowset failed. when publish txn rowset_id: {}, tablet id: {}, txn "
-                "id: {}",
-                rowset->rowset_id().to_string(), tablet_id, transaction_id);
+        status.append(fmt::format(", txn id: {}", transaction_id));
+        return status;
     }
 
     // TODO(Drogon): remove these test codes

be/test/olap/delete_handler_test.cpp

@@ -102,6 +102,7 @@ static void tear_down() {
 static void set_default_create_tablet_request(TCreateTabletReq* request) {
     request->tablet_id = 10003;
     request->__set_version(1);
+    request->partition_id = 10004;
     request->tablet_schema.schema_hash = 270068375;
     request->tablet_schema.short_key_column_count = 2;
     request->tablet_schema.keys_type = TKeysType::AGG_KEYS;

be/test/olap/delta_writer_test.cpp

@@ -105,6 +105,7 @@ static void create_tablet_request(int64_t tablet_id, int32_t schema_hash,
                                   TCreateTabletReq* request) {
     request->tablet_id = tablet_id;
     request->__set_version(1);
+    request->partition_id = 10001;
     request->tablet_schema.schema_hash = schema_hash;
     request->tablet_schema.short_key_column_count = 6;
     request->tablet_schema.keys_type = TKeysType::AGG_KEYS;
@@ -268,6 +269,7 @@ static void create_tablet_request_with_sequence_col(int64_t tablet_id, int32_t s
                                                     bool enable_mow = false) {
     request->tablet_id = tablet_id;
     request->__set_version(1);
+    request->partition_id = 30004;
     request->tablet_schema.schema_hash = schema_hash;
     request->tablet_schema.short_key_column_count = 2;
     request->tablet_schema.keys_type = TKeysType::UNIQUE_KEYS;

be/test/olap/engine_storage_migration_task_test.cpp

@@ -102,6 +102,7 @@ static void create_tablet_request_with_sequence_col(int64_t tablet_id, int32_t s
                                                     TCreateTabletReq* request) {
     request->tablet_id = tablet_id;
     request->__set_version(1);
+    request->partition_id = 10001;
     request->tablet_schema.schema_hash = schema_hash;
     request->tablet_schema.short_key_column_count = 2;
     request->tablet_schema.keys_type = TKeysType::UNIQUE_KEYS;

be/test/olap/memtable_memory_limiter_test.cpp

@@ -33,6 +33,7 @@ static void create_tablet_request(int64_t tablet_id, int32_t schema_hash,
                                   TCreateTabletReq* request) {
     request->tablet_id = tablet_id;
     request->__set_version(1);
+    request->partition_id = 30002;
     request->tablet_schema.schema_hash = schema_hash;
     request->tablet_schema.short_key_column_count = 3;
     request->tablet_schema.keys_type = TKeysType::AGG_KEYS;

be/test/olap/ordered_data_compaction_test.cpp

@@ -271,6 +271,7 @@ class OrderedDataCompactionTest : public ::testing::Test {
         std::string json_rowset_meta = R"({
             "rowset_id": 540085,
             "tablet_id": 15674,
+            "partition_id": 10000,
             "txn_id": 4045,
             "tablet_schema_hash": 567997588,
             "rowset_type": "BETA_ROWSET",

be/test/olap/path_gc_test.cpp

@@ -54,6 +54,7 @@ TEST(PathGcTest, GcTabletAndRowset) {
     auto create_tablet = [&](int64_t tablet_id) {
         auto tablet_meta = std::make_shared<TabletMeta>();
         tablet_meta->_tablet_id = tablet_id;
+        (void)tablet_meta->set_partition_id(10000);
         tablet_meta->set_tablet_uid({tablet_id, 0});
         tablet_meta->set_shard_id(tablet_id % 4);
         tablet_meta->_schema_hash = tablet_id;

be/test/olap/remote_rowset_gc_test.cpp

@@ -112,6 +112,7 @@ static void create_tablet_request_with_sequence_col(int64_t tablet_id, int32_t s
                                                     TCreateTabletReq* request) {
     request->tablet_id = tablet_id;
     request->__set_version(1);
+    request->partition_id = 30003;
     request->tablet_schema.schema_hash = schema_hash;
     request->tablet_schema.short_key_column_count = 2;
     request->tablet_schema.keys_type = TKeysType::UNIQUE_KEYS;

be/test/olap/rowid_conversion_test.cpp

@@ -223,6 +223,7 @@ class TestRowIdConversion : public testing::TestWithParam<std::tuple<KeysType, b
         std::string json_rowset_meta = R"({
             "rowset_id": 540081,
             "tablet_id": 15673,
+            "partition_id": 10000,
             "tablet_schema_hash": 567997577,
             "rowset_type": "BETA_ROWSET",
             "rowset_state": "VISIBLE",

be/test/olap/segcompaction_test.cpp

@@ -194,6 +194,7 @@ class SegCompactionTest : public testing::Test {
         std::shared_ptr<DataDir> data_dir = std::make_shared<DataDir>(lTestDir);
         TabletMetaSharedPtr tablet_meta = std::make_shared<TabletMeta>();
         tablet_meta->_tablet_id = 1;
+        tablet_meta->set_partition_id(10000);
         tablet_meta->_schema = tablet_schema;
         auto tablet = std::make_shared<Tablet>(tablet_meta, data_dir.get(), "test_str");
         char* tmp_str = (char*)malloc(20);

be/test/olap/tablet_cooldown_test.cpp

@@ -267,6 +267,7 @@ class TabletCooldownTest : public testing::Test {
 static void create_tablet_request_with_sequence_col(int64_t tablet_id, int32_t schema_hash,
                                                     TCreateTabletReq* request) {
     request->tablet_id = tablet_id;
+    request->partition_id = 30003;
     request->__set_version(1);
     request->tablet_schema.schema_hash = schema_hash;
     request->tablet_schema.short_key_column_count = 2;

be/test/olap/test_data/rowset_meta.json

@@ -1,5 +1,6 @@
 {
     "rowset_id": 10000,
+    "partition_id": 10001,
     "tablet_id": 12046,
     "tablet_schema_hash": 365187263,
     "rowset_type": "BETA_ROWSET",

be/test/olap/test_data/rowset_meta2.json

@@ -1,5 +1,6 @@
 {
     "rowset_id": 10001,
+    "partition_id": 10001,
     "tablet_id": 20487,
     "tablet_schema_hash": 1520686811,
     "rowset_type": "BETA_ROWSET",

be/test/runtime/load_stream_test.cpp

@@ -142,6 +142,7 @@ void construct_schema(OlapTableSchemaParam* schema) {
 static void create_tablet_request(int64_t tablet_id, int32_t schema_hash,
                                   TCreateTabletReq* request) {
     request->tablet_id = tablet_id;
+    request->partition_id = 30001;
     request->__set_version(1);
     request->tablet_schema.schema_hash = schema_hash;
     request->tablet_schema.short_key_column_count = 6;

be/test/vec/olap/vertical_compaction_test.cpp

@@ -273,6 +273,7 @@ class VerticalCompactionTest : public ::testing::Test {
         std::string json_rowset_meta = R"({
             "rowset_id": 540081,
             "tablet_id": 15673,
+            "partition_id": 10000,
             "tablet_schema_hash": 567997577,
             "rowset_type": "BETA_ROWSET",
             "rowset_state": "VISIBLE",