refactor: avoid routing tables returning errors due to incorrect metadata

server/cluster/metadata/cluster_metadata.go

@@ -65,9 +65,9 @@
 }
 
 func (c *ClusterMetadata) GetClusterID() storage.ClusterID {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 
 	return c.clusterID
 }
 
@@ -95,7 +95,7 @@
 		for _, table := range tables {
 			schema, ok := schemaByID[table.SchemaID]
 			if !ok {
 				c.logger.Warn("schema not exits", zap.Uint64("schemaID", uint64(table.SchemaID)))
 			}
 			tableInfos = append(tableInfos, TableInfo{
 				ID:            table.ID,
@@ -176,12 +176,12 @@
 	for _, tableName := range request.TableNames {
 		table, exists, err := c.tableManager.GetTable(request.SchemaName, tableName)
 		if err != nil {
 			c.logger.Error("get table", zap.Error(err), zap.String("schemaName", request.SchemaName), zap.String("tableName", tableName))
 			return err
 		}
 
 		if !exists {
 			c.logger.Error("the table to be closed does not exist", zap.String("schemaName", request.SchemaName), zap.String("tableName", tableName))
 			return errors.WithMessagef(ErrTableNotFound, "table not exists, shcemaName:%s,tableName:%s", request.SchemaName, tableName)
 		}
 
@@ -190,12 +190,12 @@
 	}
 
 	if _, err := c.topologyManager.RemoveTable(ctx, request.OldShardID, tableIDs); err != nil {
 		c.logger.Error("remove table from topology")
 		return err
 	}
 
 	if _, err := c.topologyManager.AddTable(ctx, request.NewShardID, tables); err != nil {
 		c.logger.Error("add table from topology")
 		return err
 	}
 
@@ -240,7 +240,7 @@
 }
 
 func (c *ClusterMetadata) AddTableTopology(ctx context.Context, shardID storage.ShardID, table storage.Table) (CreateTableResult, error) {
 	c.logger.Info("add table topology start", zap.String("cluster", c.Name()), zap.String("tableName", table.Name))
 
 	// Add table to topology manager.
 	result, err := c.topologyManager.AddTable(ctx, shardID, []storage.Table{table})
@@ -252,7 +252,7 @@
 		Table:              table,
 		ShardVersionUpdate: result,
 	}
 	c.logger.Info("add table topology succeed", zap.String("cluster", c.Name()), zap.String("result", fmt.Sprintf("%+v", ret)))
 	return ret, nil
 }
 
@@ -335,7 +335,7 @@
 	// TODO: Consider the design of the entire cluster state, which may require refactoring.
 	if uint32(len(c.registeredNodesCache)) >= c.metaData.MinNodeCount && c.topologyManager.GetClusterState() == storage.ClusterStateEmpty {
 		if err := c.UpdateClusterView(ctx, storage.ClusterStatePrepare, []storage.ShardNode{}); err != nil {
 			c.logger.Error("update cluster view failed", zap.Error(err))
 		}
 	}
 
@@ -456,22 +456,24 @@
 		if err != nil {
 			return RouteTablesResult{}, errors.WithMessage(err, "table manager get table")
 		}
-		if exists {
-			// TODO: Adapt to the current implementation of the partition table, which may need to be reconstructed later.
-			if !table.IsPartitioned() {
-				tables[table.ID] = table
-				tableIDs = append(tableIDs, table.ID)
-			} else {
-				routeEntries[table.Name] = RouteEntry{
-					Table: TableInfo{
-						ID:            table.ID,
-						Name:          table.Name,
-						SchemaID:      table.SchemaID,
-						SchemaName:    schemaName,
-						PartitionInfo: table.PartitionInfo,
-					},
-					NodeShards: nil,
-				}
+		if !exists {
+			continue
+		}
+
+		// TODO: Adapt to the current implementation of the partition table, which may need to be reconstructed later.
+		if !table.IsPartitioned() {
+			tables[table.ID] = table
+			tableIDs = append(tableIDs, table.ID)
+		} else {
+			routeEntries[table.Name] = RouteEntry{
+				Table: TableInfo{
+					ID:            table.ID,
+					Name:          table.Name,
+					SchemaID:      table.SchemaID,
+					SchemaName:    schemaName,
+					PartitionInfo: table.PartitionInfo,
+				},
+				NodeShards: nil,
 			}
 		}
 	}
@@ -495,9 +497,9 @@
 		// If nodeShards length bigger than 1, randomly select a nodeShard.
 		nodeShardsResult := nodeShards
 		if len(nodeShards) > 1 {
-			selectIndex, err := rand.Int(rand.Reader, big.NewInt(int64(len(nodeShards))))
-			if err != nil {
-				return RouteTablesResult{}, errors.WithMessage(err, "generate random node index")
+			selectIndex, err2 := rand.Int(rand.Reader, big.NewInt(int64(len(nodeShards))))
+			if err2 != nil {
+				return RouteTablesResult{}, errors.WithMessage(err2, "generate random node index")
 			}
 			nodeShardsResult = []ShardNodeWithVersion{nodeShards[selectIndex.Uint64()]}
 		}
@@ -583,11 +585,11 @@
 	return c.metaData.ProcedureExecutingBatchSize
 }
 
 func (c *ClusterMetadata) GetCreateTime() uint64 {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 
 	return c.metaData.CreatedAt
 }
 
 func (c *ClusterMetadata) GetClusterState() storage.ClusterState {
@@ -637,11 +639,11 @@
 	}
 }
 
 func (c *ClusterMetadata) GetStorageMetadata() storage.Cluster {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 
 	return c.metaData
 }
 
 // LoadMetadata load cluster metadata from storage.
@@ -653,8 +655,8 @@
 	if err != nil {
 		return errors.WithMessage(err, "get cluster")
 	}
 	c.metaData = metadata
 	return nil
 }
 
 // Initialize the cluster view and shard view of the cluster.

server/cluster/metadata/cluster_metadata_test.go

@@ -116,9 +116,10 @@ func testTableOperation(ctx context.Context, re *require.Assertions, m *metadata
 	re.True(exists)
 	re.Equal(testTableName, t.Name)
 
-	// Route table should return error when table metadata is not exists in any shard.
-	_, err = m.RouteTables(ctx, testSchema, []string{testTableName})
-	re.Error(err)
+	// Route table return empty when table not assign to any node.
+	routeTable, err := m.RouteTables(ctx, testSchema, []string{testTableName})
+	re.NoError(err)
+	re.Equal(0, len(routeTable.RouteEntries[testTableName].NodeShards))
 
 	// Test drop table metadata.
 	dropMetadataResult, err := m.DropTableMetadata(ctx, testSchema, testTableName)

server/cluster/metadata/topology_manager.go

@@ -365,14 +365,18 @@
 	shardViewVersions := make(map[storage.ShardID]uint64, 0)
 	for _, tableID := range tableIDs {
 		shardIDs, ok := m.tableShardMapping[tableID]
+		// If the table is not assigned to any shard, return an empty slice.
 		if !ok {
-			return GetShardNodesByTableIDsResult{}, ErrShardNotFound.WithCausef("table id:%d, not shard is assigned", tableID)
+			tableShardNodes[tableID] = []storage.ShardNode{}
+			continue
 		}
 
 		for _, shardID := range shardIDs {
 			shardNodes, ok := m.shardNodesMapping[shardID]
 			if !ok {
-				return GetShardNodesByTableIDsResult{}, ErrNodeNotFound.WithCausef("shard id:%d, no node is assigned", shardID)
+				// If the shard is not assigned to any node, return an empty slice.
+				tableShardNodes[tableID] = []storage.ShardNode{}
+				continue
 			}
 
 			if _, exists := tableShardNodes[tableID]; !exists {