cherry-pick 18941

.licenserc.yaml

@@ -18,7 +18,7 @@ header:
     - "src/sqlparser/**/*.rs"
     - "java/connector-node/risingwave-source-cdc/src/main/java/com/risingwave/connector/cdc/debezium/internal/*.java"
     - "java/connector-node/risingwave-sink-iceberg/src/main/java/org/apache/iceberg/**/*.java"
-    - "src/meta/model_v2/migration/**/*.rs"
+    - "src/meta/model/migration/**/*.rs"
     - "lints/ui/**"
 
   comment: on-failure

Cargo.toml

@@ -29,8 +29,8 @@ members = [
     "src/license",
     "src/meta",
     "src/meta/dashboard",
-    "src/meta/model_v2",
-    "src/meta/model_v2/migration",
+    "src/meta/model",
+    "src/meta/model/migration",
     "src/meta/node",
     "src/meta/service",
     "src/object_store",
@@ -228,8 +228,8 @@ risingwave_mem_table_spill_test = { path = "./src/stream/spill_test" }
 risingwave_meta = { path = "./src/meta" }
 risingwave_meta_dashboard = { path = "./src/meta/dashboard" }
 risingwave_meta_service = { path = "./src/meta/service" }
-risingwave_meta_model_migration = { path = "src/meta/model_v2/migration" }
-risingwave_meta_model_v2 = { path = "./src/meta/model_v2" }
+risingwave_meta_model = { path = "src/meta/model" }
+risingwave_meta_model_migration = { path = "src/meta/model/migration" }
 risingwave_meta_node = { path = "./src/meta/node" }
 risingwave_object_store = { path = "./src/object_store" }
 risingwave_pb = { path = "./src/prost" }

src/ctl/Cargo.toml

@@ -32,8 +32,8 @@ risingwave_connector = { workspace = true }
 risingwave_frontend = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 risingwave_meta = { workspace = true }
+risingwave_meta_model = { workspace = true }
 risingwave_meta_model_migration = { workspace = true }
-risingwave_meta_model_v2 = { workspace = true }
 risingwave_object_store = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }

src/ctl/src/cmd_impl/meta/reschedule.rs

@@ -19,7 +19,7 @@ use anyhow::{anyhow, Result};
 use inquire::Confirm;
 use itertools::Itertools;
 use regex::Regex;
-use risingwave_meta_model_v2::WorkerId;
+use risingwave_meta_model::WorkerId;
 use risingwave_pb::common::WorkerNode;
 use risingwave_pb::meta::{GetClusterInfoResponse, PbWorkerReschedule};
 use serde::{Deserialize, Serialize};

src/meta/Cargo.toml

@@ -58,8 +58,8 @@ risingwave_connector = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 risingwave_license = { workspace = true }
 risingwave_meta_dashboard = { workspace = true }
+risingwave_meta_model = { workspace = true }
 risingwave_meta_model_migration = { workspace = true }
-risingwave_meta_model_v2 = { workspace = true }
 risingwave_object_store = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }

src/meta/model_v2/Cargo.toml → src/meta/model/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "risingwave_meta_model_v2"
+name = "risingwave_meta_model"
 version = { workspace = true }
 edition = { workspace = true }
 homepage = { workspace = true }

src/meta/model_v2/src/README.md → src/meta/model/src/README.md

@@ -1,15 +1,18 @@
 # How to define changes between versions and generate migration and model files
 
-- Generate a new migration file and apply it to the database, check [migration](../migration/README.md) for more details. Let's take a local PG database as an example(`postgres://postgres:@localhost:5432/postgres`):
+- Generate a new migration file and apply it to the database, check [migration](../migration/README.md) for more
+  details. Let's take a local PG database as an example(`postgres://postgres:@localhost:5432/postgres`):
     ```sh
     export DATABASE_URL=postgres://postgres:@localhost:5432/postgres;
     cargo run -- generate MIGRATION_NAME
     cargo run -- up
     ```
-  - Define tables, indexes, foreign keys in the file. The new generated file will include a sample migration script,
-  you can replace it with your own migration scripts, like defining or changing tables, indexes, foreign keys and other
-  dml operation to do data correctness etc. Check [writing-migration](https://www.sea-ql.org/SeaORM/docs/migration/writing-migration/)
-  for more details.
+    - Define tables, indexes, foreign keys in the file. The new generated file will include a sample migration script,
+      you can replace it with your own migration scripts, like defining or changing tables, indexes, foreign keys and
+      other
+      dml operation to do data correctness etc.
+      Check [writing-migration](https://www.sea-ql.org/SeaORM/docs/migration/writing-migration/)
+      for more details.
   ```rust
     #[async_trait::async_trait]
     impl MigrationTrait for Migration {
@@ -24,11 +27,12 @@
       }
     }
     ```
-- Apply migration, and generate model files for new tables and indexes from the database, so you don't need to write them manually,
+- Apply migration, and generate model files for new tables and indexes from the database, so you don't need to write
+  them manually,
     ```sh
     cargo run -- up
     sea-orm-cli generate entity -u postgres://postgres:@localhost:5432/postgres -s public -o {target_dir}
-    cp {target_dir}/xxx.rs src/meta/src/model_v2/
+    cp {target_dir}/xxx.rs src/meta/src/model/
     ```
 - Defines enum and array types in the model files, since they're basically only supported in PG, and we need to
   define them in the model files manually. For example:

src/meta/service/Cargo.toml

@@ -26,7 +26,7 @@ risingwave_common = { workspace = true }
 risingwave_connector = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 risingwave_meta = { workspace = true }
-risingwave_meta_model_v2 = { workspace = true }
+risingwave_meta_model = { workspace = true }
 risingwave_pb = { workspace = true }
 sea-orm = { workspace = true }
 serde_json = "1"

src/meta/service/src/cloud_service.rs

@@ -24,7 +24,7 @@ use risingwave_connector::source::{
 };
 use risingwave_connector::{dispatch_source_prop, WithOptionsSecResolved};
 use risingwave_meta::manager::MetadataManager;
-use risingwave_meta_model_v2::ConnectionId;
+use risingwave_meta_model::ConnectionId;
 use risingwave_pb::catalog::connection::Info::PrivateLinkService;
 use risingwave_pb::cloud_service::cloud_service_server::CloudService;
 use risingwave_pb::cloud_service::rw_cloud_validate_source_response::{Error, ErrorType};

src/meta/service/src/cluster_limit_service.rs

@@ -19,7 +19,7 @@ use risingwave_common::util::cluster_limit::{
 };
 use risingwave_meta::manager::{MetaSrvEnv, MetadataManager};
 use risingwave_meta::MetaResult;
-use risingwave_meta_model_v2::WorkerId;
+use risingwave_meta_model::WorkerId;
 use risingwave_pb::common::worker_node::State;
 use risingwave_pb::common::WorkerType;
 use risingwave_pb::meta::cluster_limit_service_server::ClusterLimitService;

src/meta/service/src/cluster_service.rs

@@ -14,7 +14,7 @@
 
 use risingwave_meta::barrier::BarrierManagerRef;
 use risingwave_meta::manager::MetadataManager;
-use risingwave_meta_model_v2::WorkerId;
+use risingwave_meta_model::WorkerId;
 use risingwave_pb::common::worker_node::State;
 use risingwave_pb::common::HostAddress;
 use risingwave_pb::meta::cluster_service_server::ClusterService;

src/meta/service/src/scale_service.rs

@@ -16,7 +16,7 @@ use risingwave_common::catalog::TableId;
 use risingwave_meta::manager::MetadataManager;
 use risingwave_meta::model::TableParallelism;
 use risingwave_meta::stream::{RescheduleOptions, ScaleControllerRef, WorkerReschedule};
-use risingwave_meta_model_v2::FragmentId;
+use risingwave_meta_model::FragmentId;
 use risingwave_pb::common::WorkerType;
 use risingwave_pb::meta::scale_service_server::ScaleService;
 use risingwave_pb::meta::{

src/meta/service/src/stream_service.rs

@@ -21,7 +21,7 @@ use risingwave_meta::manager::{LocalNotification, MetadataManager};
 use risingwave_meta::model;
 use risingwave_meta::model::ActorId;
 use risingwave_meta::stream::{SourceManagerRunningInfo, ThrottleConfig};
-use risingwave_meta_model_v2::{SourceId, StreamingParallelism};
+use risingwave_meta_model::{SourceId, StreamingParallelism};
 use risingwave_pb::meta::cancel_creating_jobs_request::Jobs;
 use risingwave_pb::meta::list_actor_splits_response::FragmentType;
 use risingwave_pb::meta::list_table_fragments_response::{

src/meta/service/src/telemetry_service.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use risingwave_meta::controller::SqlMetaStore;
-use risingwave_meta_model_v2::prelude::Cluster;
+use risingwave_meta_model::prelude::Cluster;
 use risingwave_pb::meta::telemetry_info_service_server::TelemetryInfoService;
 use risingwave_pb::meta::{GetTelemetryInfoRequest, TelemetryInfoResponse};
 use sea_orm::EntityTrait;

src/meta/service/src/user_service.rs

@@ -14,7 +14,7 @@
 
 use itertools::Itertools;
 use risingwave_meta::manager::MetadataManager;
-use risingwave_meta_model_v2::UserId;
+use risingwave_meta_model::UserId;
 use risingwave_pb::user::grant_privilege::Object;
 use risingwave_pb::user::update_user_request::UpdateField;
 use risingwave_pb::user::user_service_server::UserService;

src/meta/src/backup_restore/backup_manager.rs

@@ -31,7 +31,7 @@ use risingwave_pb::meta::subscribe_response::{Info, Operation};
 use thiserror_ext::AsReport;
 use tokio::task::JoinHandle;
 
-use crate::backup_restore::meta_snapshot_builder_v2;
+use crate::backup_restore::meta_snapshot_builder;
 use crate::backup_restore::metrics::BackupManagerMetrics;
 use crate::hummock::sequence::next_meta_backup_id;
 use crate::hummock::{HummockManagerRef, HummockVersionSafePoint};
@@ -353,7 +353,7 @@ impl BackupWorker {
             };
             let meta_store = backup_manager_clone.env.meta_store();
             let mut snapshot_builder =
-                meta_snapshot_builder_v2::MetaSnapshotV2Builder::new(meta_store);
+                meta_snapshot_builder::MetaSnapshotV2Builder::new(meta_store);
             // Reuse job id as snapshot id.
             snapshot_builder
                 .build(job_id, hummock_version_builder)

src/meta/src/backup_restore/meta_snapshot_builder_v2.rs → src/meta/src/backup_restore/meta_snapshot_builder.rs

@@ -19,7 +19,7 @@ use risingwave_backup::error::{BackupError, BackupResult};
 use risingwave_backup::meta_snapshot_v2::{MetaSnapshotV2, MetadataV2};
 use risingwave_backup::MetaSnapshotId;
 use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
-use risingwave_meta_model_v2 as model_v2;
+use risingwave_meta_model as model;
 use risingwave_pb::hummock::PbHummockVersionDelta;
 use sea_orm::{DbErr, EntityTrait, QueryOrder, TransactionTrait};
 
@@ -83,8 +83,8 @@ impl MetaSnapshotV2Builder {
             )
             .await
             .map_err(map_db_err)?;
-        let version_deltas = model_v2::prelude::HummockVersionDelta::find()
-            .order_by_asc(model_v2::hummock_version_delta::Column::Id)
+        let version_deltas = model::prelude::HummockVersionDelta::find()
+            .order_by_asc(model::hummock_version_delta::Column::Id)
             .all(&txn)
             .await
             .map_err(map_db_err)?

src/meta/src/backup_restore/mod.rs

@@ -15,7 +15,7 @@
 mod backup_manager;
 pub use backup_manager::*;
 mod error;
-mod meta_snapshot_builder_v2;
+mod meta_snapshot_builder;
 mod metrics;
 mod restore;
 mod restore_impl;

src/meta/src/barrier/command.rs

@@ -22,7 +22,7 @@ use risingwave_common::hash::ActorMapping;
 use risingwave_common::types::Timestamptz;
 use risingwave_common::util::epoch::Epoch;
 use risingwave_connector::source::SplitImpl;
-use risingwave_meta_model_v2::{ObjectId, WorkerId};
+use risingwave_meta_model::{ObjectId, WorkerId};
 use risingwave_pb::catalog::{CreateType, Table};
 use risingwave_pb::common::PbWorkerNode;
 use risingwave_pb::meta::table_fragments::PbActorStatus;
@@ -1116,7 +1116,7 @@ impl CommandContext {
                 // Apply the split changes in source manager.
                 self.barrier_manager_context
                     .source_manager
-                    .drop_source_fragments(std::slice::from_ref(old_table_fragments))
+                    .drop_source_fragments_vec(std::slice::from_ref(old_table_fragments))
                     .await;
                 let source_fragments = new_table_fragments.stream_source_fragments();
                 // XXX: is it possible to have backfill fragments here?

src/meta/src/barrier/creating_job/barrier_control.rs

@@ -21,7 +21,7 @@ use std::time::Instant;
 use prometheus::HistogramTimer;
 use risingwave_common::catalog::TableId;
 use risingwave_common::metrics::{LabelGuardedHistogram, LabelGuardedIntGauge};
-use risingwave_meta_model_v2::WorkerId;
+use risingwave_meta_model::WorkerId;
 use risingwave_pb::stream_service::BarrierCompleteResponse;
 use tracing::debug;
 

src/meta/src/barrier/creating_job/mod.rs

@@ -22,7 +22,7 @@ use std::sync::Arc;
 
 use risingwave_common::catalog::TableId;
 use risingwave_common::metrics::LabelGuardedIntGauge;
-use risingwave_meta_model_v2::WorkerId;
+use risingwave_meta_model::WorkerId;
 use risingwave_pb::common::WorkerNode;
 use risingwave_pb::ddl_service::DdlProgress;
 use risingwave_pb::hummock::HummockVersionStats;

src/meta/src/barrier/creating_job/status.rs

@@ -19,7 +19,7 @@ use std::sync::Arc;
 
 use risingwave_common::hash::ActorId;
 use risingwave_common::util::epoch::Epoch;
-use risingwave_meta_model_v2::WorkerId;
+use risingwave_meta_model::WorkerId;
 use risingwave_pb::hummock::HummockVersionStats;
 use risingwave_pb::stream_plan::barrier_mutation::Mutation;
 use risingwave_pb::stream_plan::StreamActor;

src/meta/src/barrier/info.rs

@@ -15,7 +15,7 @@
 use std::collections::{HashMap, HashSet};
 
 use risingwave_common::catalog::TableId;
-use risingwave_meta_model_v2::WorkerId;
+use risingwave_meta_model::WorkerId;
 use risingwave_pb::common::WorkerNode;
 use tracing::warn;
 

src/meta/src/barrier/mod.rs

@@ -36,7 +36,7 @@ use risingwave_hummock_sdk::table_watermark::{
     merge_multiple_new_table_watermarks, TableWatermarks,
 };
 use risingwave_hummock_sdk::{HummockSstableObjectId, LocalSstableInfo};
-use risingwave_meta_model_v2::WorkerId;
+use risingwave_meta_model::WorkerId;
 use risingwave_pb::ddl_service::DdlProgress;
 use risingwave_pb::hummock::HummockVersionStats;
 use risingwave_pb::meta::{PausedReason, PbRecoveryStatus};

src/meta/src/barrier/progress.rs

@@ -18,7 +18,7 @@ use std::mem::take;
 
 use risingwave_common::catalog::TableId;
 use risingwave_common::util::epoch::Epoch;
-use risingwave_meta_model_v2::ObjectId;
+use risingwave_meta_model::ObjectId;
 use risingwave_pb::catalog::CreateType;
 use risingwave_pb::ddl_service::DdlProgress;
 use risingwave_pb::hummock::HummockVersionStats;

src/meta/src/barrier/recovery.rs

@@ -21,7 +21,7 @@ use risingwave_common::catalog::TableId;
 use risingwave_common::config::DefaultParallelism;
 use risingwave_common::hash::WorkerSlotId;
 use risingwave_common::util::epoch::Epoch;
-use risingwave_meta_model_v2::{StreamingParallelism, WorkerId};
+use risingwave_meta_model::{StreamingParallelism, WorkerId};
 use risingwave_pb::meta::subscribe_response::{Info, Operation};
 use risingwave_pb::meta::{PausedReason, Recovery};
 use risingwave_pb::stream_plan::barrier_mutation::Mutation;

src/meta/src/barrier/rpc.rs

@@ -26,7 +26,7 @@ use itertools::Itertools;
 use risingwave_common::catalog::TableId;
 use risingwave_common::util::tracing::TracingContext;
 use risingwave_hummock_sdk::HummockVersionId;
-use risingwave_meta_model_v2::WorkerId;
+use risingwave_meta_model::WorkerId;
 use risingwave_pb::common::{ActorInfo, WorkerNode};
 use risingwave_pb::stream_plan::barrier_mutation::Mutation;
 use risingwave_pb::stream_plan::{Barrier, BarrierMutation, StreamActor, SubscriptionUpstreamInfo};