Merge branch 'main' into bz/embedded-jaeger-ui

risingwavelabs · Jan 2, 2024 · 0625ba7 · 0625ba7
2 parents 8a079a0 + 1f4b882
commit 0625ba7
Show file tree

Hide file tree

Showing 51 changed files with 2,873 additions and 1,371 deletions.
diff --git a/java/common-utils/src/main/java/com/risingwave/java/utils/MetaClient.java b/java/common-utils/src/main/java/com/risingwave/java/utils/MetaClient.java
@@ -105,7 +105,7 @@ public MetaClient(String metaAddr, ScheduledExecutorService scheduler) {
                         .build();
         AddWorkerNodeResponse resp = clusterStub.addWorkerNode(req);
 
-        this.workerId = resp.getNode().getId();
+        this.workerId = resp.getNodeId();
     }
 
     public HummockVersion pinVersion() {

diff --git a/proto/ddl_service.proto b/proto/ddl_service.proto
@@ -16,8 +16,7 @@ message CreateDatabaseRequest {
 
 message CreateDatabaseResponse {
   common.Status status = 1;
-  uint32 database_id = 2;
-  uint64 version = 3;
+  uint64 version = 2;
 }
 
 message DropDatabaseRequest {
@@ -35,8 +34,7 @@ message CreateSchemaRequest {
 
 message CreateSchemaResponse {
   common.Status status = 1;
-  uint32 schema_id = 2;
-  uint64 version = 3;
+  uint64 version = 2;
 }
 
 message DropSchemaRequest {
@@ -55,8 +53,7 @@ message CreateSourceRequest {
 
 message CreateSourceResponse {
   common.Status status = 1;
-  uint32 source_id = 2;
-  uint64 version = 3;
+  uint64 version = 2;
 }
 
 message DropSourceRequest {
@@ -87,8 +84,7 @@ message CreateSinkRequest {
 
 message CreateSinkResponse {
   common.Status status = 1;
-  uint32 sink_id = 2;
-  uint64 version = 3;
+  uint64 version = 2;
 }
 
 message DropSinkRequest {
@@ -109,8 +105,7 @@ message CreateMaterializedViewRequest {
 
 message CreateMaterializedViewResponse {
   common.Status status = 1;
-  uint32 table_id = 2;
-  uint64 version = 3;
+  uint64 version = 2;
 }
 
 message DropMaterializedViewRequest {
@@ -129,8 +124,7 @@ message CreateViewRequest {
 
 message CreateViewResponse {
   common.Status status = 1;
-  uint32 view_id = 2;
-  uint64 version = 3;
+  uint64 version = 2;
 }
 
 message DropViewRequest {
@@ -166,8 +160,7 @@ message CreateTableRequest {
 
 message CreateTableResponse {
   common.Status status = 1;
-  uint32 table_id = 2;
-  uint64 version = 3;
+  uint64 version = 2;
 }
 
 message AlterNameRequest {
@@ -235,8 +228,7 @@ message CreateFunctionRequest {
 
 message CreateFunctionResponse {
   common.Status status = 1;
-  uint32 function_id = 2;
-  uint64 version = 3;
+  uint64 version = 2;
 }
 
 message DropFunctionRequest {
@@ -277,8 +269,7 @@ message CreateIndexRequest {
 
 message CreateIndexResponse {
   common.Status status = 1;
-  uint32 index_id = 2;
-  uint64 version = 4;
+  uint64 version = 2;
 }
 
 message DropIndexRequest {
@@ -352,9 +343,8 @@ message CreateConnectionRequest {
 }
 
 message CreateConnectionResponse {
-  uint32 connection_id = 1;
   // global catalog version
-  uint64 version = 2;
+  uint64 version = 1;
 }
 
 message ListConnectionsRequest {}

diff --git a/proto/meta.proto b/proto/meta.proto
@@ -288,11 +288,12 @@ message AddWorkerNodeResponse {
   reserved 3;
   reserved "system_params";
   common.Status status = 1;
-  common.WorkerNode node = 2;
+  optional uint32 node_id = 2;
 }
 
 message ActivateWorkerNodeRequest {
   common.HostAddress host = 1;
+  uint32 node_id = 2;
 }
 
 message ActivateWorkerNodeResponse {

diff --git a/src/frontend/src/catalog/catalog_service.rs b/src/frontend/src/catalog/catalog_service.rs
@@ -195,7 +195,7 @@ pub struct CatalogWriterImpl {
 #[async_trait::async_trait]
 impl CatalogWriter for CatalogWriterImpl {
     async fn create_database(&self, db_name: &str, owner: UserId) -> Result<()> {
-        let (_, version) = self
+        let version = self
             .meta_client
             .create_database(PbDatabase {
                 name: db_name.to_string(),
@@ -212,7 +212,7 @@ impl CatalogWriter for CatalogWriterImpl {
         schema_name: &str,
         owner: UserId,
     ) -> Result<()> {
-        let (_, version) = self
+        let version = self
             .meta_client
             .create_schema(PbSchema {
                 id: 0,
@@ -231,7 +231,7 @@ impl CatalogWriter for CatalogWriterImpl {
         graph: StreamFragmentGraph,
     ) -> Result<()> {
         let create_type = table.get_create_type().unwrap_or(PbCreateType::Foreground);
-        let (_, version) = self
+        let version = self
             .meta_client
             .create_materialized_view(table, graph)
             .await?;
@@ -242,7 +242,7 @@ impl CatalogWriter for CatalogWriterImpl {
     }
 
     async fn create_view(&self, view: PbView) -> Result<()> {
-        let (_, version) = self.meta_client.create_view(view).await?;
+        let version = self.meta_client.create_view(view).await?;
         self.wait_version(version).await
     }
 
@@ -252,7 +252,7 @@ impl CatalogWriter for CatalogWriterImpl {
         table: PbTable,
         graph: StreamFragmentGraph,
     ) -> Result<()> {
-        let (_, version) = self.meta_client.create_index(index, table, graph).await?;
+        let version = self.meta_client.create_index(index, table, graph).await?;
         self.wait_version(version).await
     }
 
@@ -263,7 +263,7 @@ impl CatalogWriter for CatalogWriterImpl {
         graph: StreamFragmentGraph,
         job_type: PbTableJobType,
     ) -> Result<()> {
-        let (_, version) = self
+        let version = self
             .meta_client
             .create_table(source, table, graph, job_type)
             .await?;
@@ -290,7 +290,7 @@ impl CatalogWriter for CatalogWriterImpl {
     }
 
     async fn create_source(&self, source: PbSource) -> Result<()> {
-        let (_id, version) = self.meta_client.create_source(source).await?;
+        let version = self.meta_client.create_source(source).await?;
         self.wait_version(version).await
     }
 
@@ -299,7 +299,7 @@ impl CatalogWriter for CatalogWriterImpl {
         source: PbSource,
         graph: StreamFragmentGraph,
     ) -> Result<()> {
-        let (_id, version) = self
+        let version = self
             .meta_client
             .create_source_with_graph(source, graph)
             .await?;
@@ -312,15 +312,15 @@ impl CatalogWriter for CatalogWriterImpl {
         graph: StreamFragmentGraph,
         affected_table_change: Option<ReplaceTablePlan>,
     ) -> Result<()> {
-        let (_id, version) = self
+        let version = self
             .meta_client
             .create_sink(sink, graph, affected_table_change)
             .await?;
         self.wait_version(version).await
     }
 
     async fn create_function(&self, function: PbFunction) -> Result<()> {
-        let (_, version) = self.meta_client.create_function(function).await?;
+        let version = self.meta_client.create_function(function).await?;
         self.wait_version(version).await
     }
 
@@ -332,7 +332,7 @@ impl CatalogWriter for CatalogWriterImpl {
         owner_id: u32,
         connection: create_connection_request::Payload,
     ) -> Result<()> {
-        let (_, version) = self
+        let version = self
             .meta_client
             .create_connection(
                 connection_name,