minor

proto/common.proto

@@ -54,9 +54,8 @@ message WorkerNode {
     bool is_streaming = 1;
     bool is_serving = 2;
     bool is_unschedulable = 3;
-    // Secondary host address for the worker node.
     // This is used for frontend node to register its rpc address
-    string secondary_host = 4;
+    string internal_rpc_host_addr = 4;
   }
   message Resource {
     string rw_version = 1;

proto/meta.proto

@@ -312,9 +312,8 @@ message AddWorkerNodeRequest {
     bool is_streaming = 2;
     bool is_serving = 3;
     bool is_unschedulable = 4;
-    // Secondary host address for the worker node.
     // This is used for frontend node to register its rpc address
-    string secondary_host = 5;
+    string internal_rpc_host_addr = 5;
   }
   common.WorkerType worker_type = 1;
   common.HostAddress host = 2;

src/batch/src/worker_manager/worker_node_manager.rs

@@ -430,7 +430,7 @@ mod tests {
                     is_unschedulable: false,
                     is_serving: true,
                     is_streaming: true,
-                    secondary_host: "".to_string(),
+                    internal_rpc_host_addr: "".to_string(),
                 }),
                 transactional_id: Some(1),
                 ..Default::default()
@@ -445,7 +445,7 @@ mod tests {
                     is_unschedulable: false,
                     is_serving: true,
                     is_streaming: false,
-                    secondary_host: "".to_string(),
+                    internal_rpc_host_addr: "".to_string(),
                 }),
                 transactional_id: Some(2),
                 ..Default::default()

src/common/src/vnode_mapping/vnode_placement.rs

@@ -213,7 +213,7 @@ mod tests {
             is_unschedulable: false,
             is_serving: true,
             is_streaming: false,
-            secondary_host: "".to_string(),
+            internal_rpc_host_addr: "".to_string(),
         };
 
         let count_same_vnode_mapping = |wm1: &WorkerSlotMapping, wm2: &WorkerSlotMapping| {

src/compute/src/server.rs

@@ -127,7 +127,7 @@ pub async fn compute_node_serve(
             is_streaming: opts.role.for_streaming(),
             is_serving: opts.role.for_serving(),
             is_unschedulable: false,
-            secondary_host: "".to_string(),
+            internal_rpc_host_addr: "".to_string(),
         },
         &config.meta,
     )

src/ctl/src/cmd_impl/meta/migration.rs

@@ -162,7 +162,7 @@ pub async fn migrate(from: EtcdBackend, target: String, force_clean: bool) -> an
                 is_serving: Set(pb_property.is_serving),
                 is_unschedulable: Set(pb_property.is_unschedulable),
                 parallelism: Set(worker.worker_node.parallelism() as _),
-                secondary_host: Set(pb_property.secondary_host.clone()),
+                internal_rpc_host_addr: Set(pb_property.internal_rpc_host_addr.clone()),
             };
             WorkerProperty::insert(property)
                 .exec(&meta_store_sql.conn)

src/frontend/src/catalog/system_catalog/rw_catalog/rw_worker_nodes.rs

@@ -33,7 +33,7 @@ struct RwWorkerNode {
     is_streaming: Option<bool>,
     is_serving: Option<bool>,
     is_unschedulable: Option<bool>,
-    secondary_host: Option<String>,
+    internal_rpc_host_addr: Option<String>,
     rw_version: Option<String>,
     system_total_memory_bytes: Option<i64>,
     system_total_cpu_cores: Option<i64>,
@@ -74,7 +74,7 @@ async fn read_rw_worker_nodes_info(reader: &SysCatalogReaderImpl) -> Result<Vec<
                 } else {
                     None
                 },
-                secondary_host: property.map(|p| p.secondary_host.clone()),
+                internal_rpc_host_addr: property.map(|p| p.internal_rpc_host_addr.clone()),
                 rw_version: resource.map(|r| r.rw_version.to_owned()),
                 system_total_memory_bytes: resource.map(|r| r.total_memory_bytes as _),
                 system_total_cpu_cores: resource.map(|r| r.total_cpu_cores as _),

src/frontend/src/scheduler/distributed/query.rs

@@ -681,7 +681,7 @@ pub(crate) mod tests {
                 is_unschedulable: false,
                 is_serving: true,
                 is_streaming: true,
-                secondary_host: "".to_string(),
+                internal_rpc_host_addr: "".to_string(),
             }),
             transactional_id: Some(0),
             ..Default::default()
@@ -699,7 +699,7 @@ pub(crate) mod tests {
                 is_unschedulable: false,
                 is_serving: true,
                 is_streaming: true,
-                secondary_host: "".to_string(),
+                internal_rpc_host_addr: "".to_string(),
             }),
             transactional_id: Some(1),
             ..Default::default()
@@ -717,7 +717,7 @@ pub(crate) mod tests {
                 is_unschedulable: false,
                 is_serving: true,
                 is_streaming: true,
-                secondary_host: "".to_string(),
+                internal_rpc_host_addr: "".to_string(),
             }),
             transactional_id: Some(2),
             ..Default::default()

src/frontend/src/session.rs

@@ -274,7 +274,7 @@ impl FrontendEnv {
             WorkerType::Frontend,
             &frontend_address,
             AddWorkerNodeProperty {
-                secondary_host: frontend_rpc_addr.to_string(),
+                internal_rpc_host_addr: frontend_rpc_addr.to_string(),
                 ..Default::default()
             },
             &config.meta,

src/meta/model_v2/migration/src/m20240726_063833_auto_schema_change.rs

@@ -19,7 +19,7 @@ impl MigrationTrait for Migration {
             .alter_table(
                 MigrationTable::alter()
                     .table(WorkerProperty::Table)
-                    .add_column(ColumnDef::new(WorkerProperty::SecondaryHost).string())
+                    .add_column(ColumnDef::new(WorkerProperty::InternalRpcHostAddr).string())
                     .to_owned(),
             )
             .await
@@ -39,7 +39,7 @@ impl MigrationTrait for Migration {
             .alter_table(
                 MigrationTable::alter()
                     .table(WorkerProperty::Table)
-                    .drop_column(WorkerProperty::SecondaryHost)
+                    .drop_column(WorkerProperty::InternalRpcHostAddr)
                     .to_owned(),
             )
             .await
@@ -55,5 +55,5 @@ enum Table {
 #[derive(DeriveIden)]
 enum WorkerProperty {
     Table,
-    SecondaryHost,
+    InternalRpcHostAddr,
 }

src/meta/model_v2/src/worker_property.rs

@@ -26,7 +26,7 @@ pub struct Model {
     pub is_streaming: bool,
     pub is_serving: bool,
     pub is_unschedulable: bool,
-    pub secondary_host: String,
+    pub internal_rpc_host_addr: String,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]

src/meta/src/controller/catalog.rs

@@ -2879,12 +2879,12 @@ impl CatalogController {
 
     pub async fn get_table_by_cdc_table_id(
         &self,
-        cdc_table_name: String,
+        cdc_table_id: String,
     ) -> MetaResult<Vec<PbTable>> {
         let inner = self.inner.read().await;
         let table_objs = Table::find()
             .find_also_related(Object)
-            .filter(table::Column::CdcTableId.eq(cdc_table_name))
+            .filter(table::Column::CdcTableId.eq(cdc_table_id))
             .all(&inner.db)
             .await?;
         Ok(table_objs

src/meta/src/controller/cluster.rs

@@ -85,7 +85,7 @@ impl From<WorkerInfo> for PbWorkerNode {
                 is_streaming: p.is_streaming,
                 is_serving: p.is_serving,
                 is_unschedulable: p.is_unschedulable,
-                secondary_host: p.secondary_host.clone(),
+                internal_rpc_host_addr: p.internal_rpc_host_addr.clone(),
             }),
             transactional_id: info.0.transaction_id.map(|id| id as _),
             resource: info.2.resource,
@@ -724,7 +724,7 @@ impl ClusterControllerInner {
                 is_streaming: Set(add_property.is_streaming),
                 is_serving: Set(add_property.is_serving),
                 is_unschedulable: Set(add_property.is_unschedulable),
-                secondary_host: Set(add_property.secondary_host),
+                internal_rpc_host_addr: Set(add_property.internal_rpc_host_addr),
             };
             WorkerProperty::insert(property).exec(&txn).await?;
         }
@@ -971,7 +971,7 @@ mod tests {
             is_streaming: true,
             is_serving: true,
             is_unschedulable: false,
-            secondary_host: "".to_string(),
+            internal_rpc_host_addr: "".to_string(),
         };
         let hosts = mock_worker_hosts_for_test(worker_count);
         let mut worker_ids = vec![];
@@ -1062,7 +1062,7 @@ mod tests {
             is_streaming: true,
             is_serving: true,
             is_unschedulable: false,
-            secondary_host: "".to_string(),
+            internal_rpc_host_addr: "".to_string(),
         };
         let worker_id = cluster_ctl
             .add_worker(

src/meta/src/hummock/manager/tests.rs

@@ -405,7 +405,7 @@ async fn test_release_context_resource() {
                 is_streaming: true,
                 is_serving: true,
                 is_unschedulable: false,
-                secondary_host: "".to_string(),
+                internal_rpc_host_addr: "".to_string(),
             },
             Default::default(),
         )
@@ -488,7 +488,7 @@ async fn test_hummock_manager_basic() {
                 is_streaming: true,
                 is_serving: true,
                 is_unschedulable: false,
-                secondary_host: "".to_string(),
+                internal_rpc_host_addr: "".to_string(),
             },
             Default::default(),
         )

src/meta/src/hummock/test_utils.rs

@@ -352,7 +352,7 @@ pub async fn setup_compute_env_with_metric(
                 is_streaming: true,
                 is_serving: true,
                 is_unschedulable: false,
-                secondary_host: "".to_string(),
+                internal_rpc_host_addr: "".to_string(),
             },
             Default::default(),
         )

src/meta/src/manager/cluster.rs

@@ -504,7 +504,7 @@ impl ClusterManager {
                 is_streaming: worker_property.is_streaming,
                 is_serving: worker_property.is_serving,
                 is_unschedulable: worker_property.is_unschedulable,
-                secondary_host: worker_property.secondary_host,
+                internal_rpc_host_addr: worker_property.internal_rpc_host_addr,
             })
         } else {
             None
@@ -839,7 +839,7 @@ mod tests {
                         is_streaming: true,
                         is_serving: true,
                         is_unschedulable: false,
-                        secondary_host: "".to_string(),
+                        internal_rpc_host_addr: "".to_string(),
                     },
                     Default::default(),
                 )
@@ -881,7 +881,7 @@ mod tests {
                     is_streaming: true,
                     is_serving: true,
                     is_unschedulable: false,
-                    secondary_host: "".to_string(),
+                    internal_rpc_host_addr: "".to_string(),
                 },
                 Default::default(),
             )
@@ -904,7 +904,7 @@ mod tests {
                     is_streaming: true,
                     is_serving: true,
                     is_unschedulable: false,
-                    secondary_host: "".to_string(),
+                    internal_rpc_host_addr: "".to_string(),
                 },
                 Default::default(),
             )
@@ -954,7 +954,7 @@ mod tests {
                     is_streaming: true,
                     is_serving: true,
                     is_unschedulable: false,
-                    secondary_host: "".to_string(),
+                    internal_rpc_host_addr: "".to_string(),
                 },
                 Default::default(),
             )
@@ -1013,7 +1013,7 @@ mod tests {
                     is_streaming: true,
                     is_serving: true,
                     is_unschedulable: false,
-                    secondary_host: "".to_string(),
+                    internal_rpc_host_addr: "".to_string(),
                 },
                 Default::default(),
             )

src/meta/src/stream/stream_manager.rs

@@ -1038,7 +1038,7 @@ mod tests {
                         is_streaming: true,
                         is_serving: true,
                         is_unschedulable: false,
-                        secondary_host: "".to_string(),
+                        internal_rpc_host_addr: "".to_string(),
                     },
                     Default::default(),
                 )

src/rpc_client/src/lib.rs

@@ -137,7 +137,7 @@ where
                 .property
                 .as_ref()
                 .expect("frontend node property is missing");
-            HostAddr::from_str(prop.secondary_host.as_str())?
+            HostAddr::from_str(prop.internal_rpc_host_addr.as_str())?
         } else {
             node.get_host().unwrap().into()
         };

src/stream/src/task/stream_manager.rs

@@ -577,7 +577,7 @@ impl StreamActorManager {
             } = actor;
             let actor = actor.unwrap();
             let actor_id = actor.actor_id;
-            let streaming_config = Arc::new(shared_context.config.clone());
+            let streaming_config = self.env.config().clone();
             let actor_context = ActorContext::create(
                 &actor,
                 self.env.total_mem_usage(),