update docs and tolerate placeholder for incomplete catalogs

Signed-off-by: Bugen Zhao <[email protected]>

proto/catalog.proto

@@ -419,13 +419,13 @@ message Table {
 
   // Total vnode count of the table.
   //
-  // Can be unset if...
-  // - The catalog is generated by the frontend and not yet persisted, this is
-  //   because the vnode count of each fragment (then internal tables) is determined
-  //   by the meta service.
-  // - The table is created in older versions where variable vnode count is not
+  // Use `VnodeCountCompat::vnode_count` to access it.
+  //
+  // - Can be unset if the table is created in older versions where variable vnode count is not
   //   supported, in which case a default value of 256 should be used.
-  //   Use `VnodeCountCompat::vnode_count` to access it.
+  // - Can be placeholder value `Some(0)` if the catalog is generated by the frontend and the
+  //   corresponding job is still in `Creating` status, in which case calling `vnode_count`
+  //   will panic.
   //
   // Please note that this field is not intended to describe the expected vnode count
   // for a streaming job. Instead, refer to `stream_plan.StreamFragmentGraph.max_parallelism`.

src/frontend/src/catalog/table_catalog.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::assert_matches::assert_matches;
 use std::collections::{HashMap, HashSet};
 
 use fixedbitset::FixedBitSet;
@@ -174,14 +175,9 @@ pub struct TableCatalog {
 
     /// Total vnode count of the table.
     ///
-    /// Can be placeholder if the catalog is generated by the frontend and not yet persisted. This is
-    /// because the vnode count of each fragment (then internal tables) is determined by the
-    /// meta service. In this case, calling [`TableCatalog::vnode_count()`] will panic.
-    /// See also [`StreamMaterialize::derive_table_catalog`] and [`TableCatalogBuilder::build`].
-    ///
-    /// On the contrary, if this comes from a [`PbTable`], the field must have value no matter
-    /// whether the table is created before or after the version we introduced variable vnode
-    /// count support. We will handle backward compatibility when obtaining the value.
+    /// Can be [`VnodeCount::Placeholder`] if the catalog is generated by the frontend and the
+    /// corresponding job is still in `Creating` status, in which case calling [`Self::vnode_count`]
+    /// will panic.
     ///
     /// [`StreamMaterialize::derive_table_catalog`]: crate::optimizer::plan_node::StreamMaterialize::derive_table_catalog
     /// [`TableCatalogBuilder::build`]: crate::optimizer::plan_node::utils::TableCatalogBuilder::build
@@ -561,7 +557,13 @@ impl From<PbTable> for TableCatalog {
             OptionalAssociatedSourceId::AssociatedSourceId(id) => id,
         });
         let name = tb.name.clone();
-        let vnode_count = tb.vnode_count();
+
+        let vnode_count = tb.vnode_count_inner();
+        if let VnodeCount::Placeholder = vnode_count {
+            // Only allow placeholder vnode count for creating tables.
+            // After the table is created, an `Update` notification will be used to update the vnode count field.
+            assert_matches!(stream_job_status, PbStreamJobStatus::Creating);
+        }
 
         let mut col_names = HashSet::new();
         let mut col_index: HashMap<i32, usize> = HashMap::new();
@@ -632,7 +634,7 @@ impl From<PbTable> for TableCatalog {
                 .map(TableId::from)
                 .collect_vec(),
             cdc_table_id: tb.cdc_table_id,
-            vnode_count: VnodeCount::set(vnode_count), /* from existing (persisted) tables, vnode_count must be set */
+            vnode_count,
         }
     }
 }