diff --git a/fluss-gateway/src/backend/fake.rs b/fluss-gateway/src/backend/fake.rs
index 589c8af8362..189265b1396 100644
--- a/fluss-gateway/src/backend/fake.rs
+++ b/fluss-gateway/src/backend/fake.rs
@@ -216,7 +216,9 @@ fn write_table_info(
         schema = schema.column(name, data_type);
     }
     if let Some(keys) = primary_key {
-        schema = schema.primary_key(keys.iter().copied());
+        schema = schema
+            .primary_key(keys.iter().copied())
+            .expect("valid fixture primary key");
     }
     let descriptor = TableDescriptor::builder()
         .schema(schema.build().expect("valid fixture schema"))
diff --git a/fluss-gateway/src/protocol/rest/ddl.rs b/fluss-gateway/src/protocol/rest/ddl.rs
index 582323c1d9f..48207924154 100644
--- a/fluss-gateway/src/protocol/rest/ddl.rs
+++ b/fluss-gateway/src/protocol/rest/ddl.rs
@@ -520,7 +520,9 @@ fn table_descriptor(database: &str, body: CreateTableBody) -> GatewayResult<Tabl
         }
     }
     if let Some(primary_key) = body.primary_key {
-        schema = schema.primary_key(primary_key);
+        schema = schema.primary_key(primary_key).map_err(|error| {
+            GatewayError::invalid_argument(format!("invalid table schema: {error}"))
+        })?;
     }
     let schema = schema.build().map_err(|error| {
         GatewayError::invalid_argument(format!("invalid table schema: {error}"))
@@ -782,6 +784,7 @@ mod tests {
             )
             .with_comment("the order total")
             .primary_key(["id", "dt"])
+            .unwrap()
             .build()
             .expect("a fixture schema");
         let (status, location, body) = post(
diff --git a/fluss-gateway/src/protocol/rest/metadata.rs b/fluss-gateway/src/protocol/rest/metadata.rs
index c52f586f768..38fc2786d0c 100644
--- a/fluss-gateway/src/protocol/rest/metadata.rs
+++ b/fluss-gateway/src/protocol/rest/metadata.rs
@@ -708,6 +708,7 @@ mod tests {
             .with_comment("the order total")
             .column("dt", DataType::String(StringType::with_nullable(false)))
             .primary_key(["id", "dt"])
+            .unwrap()
             .build()
             .expect("the described schema is valid");
         let descriptor = TableDescriptor::builder()
diff --git a/fluss-gateway/src/protocol/rest/records.rs b/fluss-gateway/src/protocol/rest/records.rs
index d69be343be6..624b97ce4c2 100644
--- a/fluss-gateway/src/protocol/rest/records.rs
+++ b/fluss-gateway/src/protocol/rest/records.rs
@@ -755,6 +755,7 @@ mod tests {
                 DataType::String(fluss::metadata::StringType::with_nullable(false)),
             )
             .primary_key(["id"])
+            .unwrap()
             .build()
             .unwrap();
         let descriptor = TableDescriptor::builder()
@@ -787,6 +788,7 @@ mod tests {
                 DataType::BigInt(fluss::metadata::BigIntType::with_nullable(true)),
             )
             .primary_key(["uid"])
+            .unwrap()
             .enable_auto_increment("uid_int")
             .unwrap()
             .build()
diff --git a/fluss-gateway/tests/e2e_cluster.rs b/fluss-gateway/tests/e2e_cluster.rs
index ac974c399d3..f8a08b2dd17 100644
--- a/fluss-gateway/tests/e2e_cluster.rs
+++ b/fluss-gateway/tests/e2e_cluster.rs
@@ -515,6 +515,7 @@ async fn assert_schema_recreation(api: &Api, connection: &FlussConnection) {
                     .column(columns[0], DataTypes::string())
                     .column(columns[1], DataTypes::string())
                     .primary_key(["id"])
+                    .unwrap()
                     .build()
                     .unwrap(),
             )
@@ -726,6 +727,7 @@ async fn metadata_apis_support_plaintext_and_sasl_fluss_clusters() {
                 .column("name", DataTypes::string())
                 .column("note", DataTypes::string())
                 .primary_key(["id"])
+                .unwrap()
                 .build()
                 .expect("build the KV schema"),
         )
diff --git a/fluss-rust/README.md b/fluss-rust/README.md
index 9d9d9c8522d..4edf086774c 100644
--- a/fluss-rust/README.md
+++ b/fluss-rust/README.md
@@ -198,7 +198,7 @@ async fn main() -> Result<()> {
         .column("id", DataTypes::int())
         .column("name", DataTypes::string())
         .column("score", DataTypes::bigint())
-        .primary_key(vec!["id"])
+        .primary_key(vec!["id"])?
         .build()?;
     let descriptor = TableDescriptor::builder().schema(schema).build()?;
     admin.create_table(&table_path, &descriptor, true).await?;
diff --git a/fluss-rust/bindings/cpp/src/types.rs b/fluss-rust/bindings/cpp/src/types.rs
index a19949c2d11..84b88b98406 100644
--- a/fluss-rust/bindings/cpp/src/types.rs
+++ b/fluss-rust/bindings/cpp/src/types.rs
@@ -244,7 +244,7 @@ pub fn ffi_descriptor_to_core(
     }
 
     if !descriptor.schema.primary_keys.is_empty() {
-        schema_builder = schema_builder.primary_key(descriptor.schema.primary_keys.clone());
+        schema_builder = schema_builder.primary_key(descriptor.schema.primary_keys.clone())?;
     }
 
     for auto_increment_column in &descriptor.schema.auto_increment_columns {
diff --git a/fluss-rust/bindings/elixir/native/fluss_nif/src/schema.rs b/fluss-rust/bindings/elixir/native/fluss_nif/src/schema.rs
index 375fee881ee..bc53526f771 100644
--- a/fluss-rust/bindings/elixir/native/fluss_nif/src/schema.rs
+++ b/fluss-rust/bindings/elixir/native/fluss_nif/src/schema.rs
@@ -178,7 +178,9 @@ fn table_descriptor_new(
         schema_builder = schema_builder.column(name, to_fluss_type(dt));
     }
     if !schema.primary_key.is_empty() {
-        schema_builder = schema_builder.primary_key(schema.primary_key);
+        schema_builder = schema_builder
+            .primary_key(schema.primary_key)
+            .map_err(to_nif_err)?;
     }
 
     let built_schema = schema_builder.build().map_err(to_nif_err)?;
diff --git a/fluss-rust/bindings/python/src/metadata.rs b/fluss-rust/bindings/python/src/metadata.rs
index af1cede8ffe..01db2f5940e 100644
--- a/fluss-rust/bindings/python/src/metadata.rs
+++ b/fluss-rust/bindings/python/src/metadata.rs
@@ -176,7 +176,9 @@ impl Schema {
 
         if let Some(pk_columns) = primary_keys {
             if !pk_columns.is_empty() {
-                builder = builder.primary_key(pk_columns);
+                builder = builder
+                    .primary_key(pk_columns)
+                    .map_err(|e| FlussError::new_err(format!("Failed to build schema: {e}")))?;
             }
         }
 
diff --git a/fluss-rust/crates/examples/src/example_auto_increment_table.rs b/fluss-rust/crates/examples/src/example_auto_increment_table.rs
index f3beae775a5..d9fbf34aa8a 100644
--- a/fluss-rust/crates/examples/src/example_auto_increment_table.rs
+++ b/fluss-rust/crates/examples/src/example_auto_increment_table.rs
@@ -39,7 +39,7 @@ pub async fn main() -> Result<()> {
                 .column("uid", DataTypes::string())
                 .column("region", DataTypes::string())
                 .column("uid_int", DataTypes::bigint())
-                .primary_key(vec!["uid"])
+                .primary_key(vec!["uid"])?
                 .enable_auto_increment("uid_int")?
                 .build()?,
         )
diff --git a/fluss-rust/crates/examples/src/example_kv_changelog.rs b/fluss-rust/crates/examples/src/example_kv_changelog.rs
index 1554e66a06e..3d07528454f 100644
--- a/fluss-rust/crates/examples/src/example_kv_changelog.rs
+++ b/fluss-rust/crates/examples/src/example_kv_changelog.rs
@@ -39,7 +39,7 @@ pub async fn main() -> Result<()> {
             Schema::builder()
                 .column("id", DataTypes::int())
                 .column("name", DataTypes::string())
-                .primary_key(vec!["id"])
+                .primary_key(vec!["id"])?
                 .build()?,
         )
         .distributed_by(Some(1), vec!["id".to_string()])
diff --git a/fluss-rust/crates/examples/src/example_kv_table.rs b/fluss-rust/crates/examples/src/example_kv_table.rs
index d3367490958..b44498feb9a 100644
--- a/fluss-rust/crates/examples/src/example_kv_table.rs
+++ b/fluss-rust/crates/examples/src/example_kv_table.rs
@@ -36,7 +36,7 @@ pub async fn main() -> Result<()> {
                 .column("id", DataTypes::int())
                 .column("name", DataTypes::string())
                 .column("age", DataTypes::bigint())
-                .primary_key(vec!["id"])
+                .primary_key(vec!["id"])?
                 .build()?,
         )
         .build()?;
diff --git a/fluss-rust/crates/examples/src/example_partitioned_kv_table.rs b/fluss-rust/crates/examples/src/example_partitioned_kv_table.rs
index c9231120dff..d2d44bbb5f6 100644
--- a/fluss-rust/crates/examples/src/example_partitioned_kv_table.rs
+++ b/fluss-rust/crates/examples/src/example_partitioned_kv_table.rs
@@ -38,7 +38,7 @@ pub async fn main() -> Result<()> {
                 .column("region", DataTypes::string())
                 .column("zone", DataTypes::bigint())
                 .column("score", DataTypes::bigint())
-                .primary_key(vec!["id", "region", "zone"])
+                .primary_key(vec!["id", "region", "zone"])?
                 .build()?,
         )
         .partitioned_by(vec!["region", "zone"])
diff --git a/fluss-rust/crates/examples/src/example_partitioned_prefix_lookup.rs b/fluss-rust/crates/examples/src/example_partitioned_prefix_lookup.rs
index 009afbc41b8..ea9993810a4 100644
--- a/fluss-rust/crates/examples/src/example_partitioned_prefix_lookup.rs
+++ b/fluss-rust/crates/examples/src/example_partitioned_prefix_lookup.rs
@@ -45,7 +45,7 @@ pub async fn main() -> Result<()> {
                 .column("session_id", DataTypes::string())
                 .column("event_seq", DataTypes::bigint())
                 .column("event_data", DataTypes::string())
-                .primary_key(vec!["region", "user_id", "session_id", "event_seq"])
+                .primary_key(vec!["region", "user_id", "session_id", "event_seq"])?
                 .build()?,
         )
         .partitioned_by(vec!["region"])
diff --git a/fluss-rust/crates/examples/src/example_prefix_lookup.rs b/fluss-rust/crates/examples/src/example_prefix_lookup.rs
index 2bd583a1003..2171fc25894 100644
--- a/fluss-rust/crates/examples/src/example_prefix_lookup.rs
+++ b/fluss-rust/crates/examples/src/example_prefix_lookup.rs
@@ -40,7 +40,7 @@ pub async fn main() -> Result<()> {
                 .column("session_id", DataTypes::string())
                 .column("event_seq", DataTypes::bigint())
                 .column("event_data", DataTypes::string())
-                .primary_key(vec!["user_id", "session_id", "event_seq"])
+                .primary_key(vec!["user_id", "session_id", "event_seq"])?
                 .build()?,
         )
         .distributed_by(
diff --git a/fluss-rust/crates/fluss/README.md b/fluss-rust/crates/fluss/README.md
index 8a07cb5bc81..92becc594b4 100644
--- a/fluss-rust/crates/fluss/README.md
+++ b/fluss-rust/crates/fluss/README.md
@@ -27,11 +27,11 @@ async fn main() -> Result<()> {
 
     // ---- Primary key (KV) table: upsert and lookup ----
     let kv_path = TablePath::new("fluss", "users");
-    let mut kv_schema = Schema::builder()
+    let kv_schema = Schema::builder()
         .column("id", DataTypes::int())
         .column("name", DataTypes::string())
         .column("age", DataTypes::bigint())
-        .primary_key(vec!["id"]);
+        .primary_key(vec!["id"])?;
     let kv_descriptor = TableDescriptor::builder()
         .schema(kv_schema.build()?)
         .build()?;
diff --git a/fluss-rust/crates/fluss/src/client/table/scanner.rs b/fluss-rust/crates/fluss/src/client/table/scanner.rs
index b68968ac278..6aba4c306fa 100644
--- a/fluss-rust/crates/fluss/src/client/table/scanner.rs
+++ b/fluss-rust/crates/fluss/src/client/table/scanner.rs
@@ -3140,7 +3140,7 @@ mod tests {
             .column("name", DataTypes::string());
 
         if has_primary_key {
-            schema_builder = schema_builder.primary_key(vec!["id"]);
+            schema_builder = schema_builder.primary_key(vec!["id"]).unwrap();
         }
 
         let schema = schema_builder.build().unwrap();
diff --git a/fluss-rust/crates/fluss/src/lib.rs b/fluss-rust/crates/fluss/src/lib.rs
index e9b0995af4c..3066dd853ce 100644
--- a/fluss-rust/crates/fluss/src/lib.rs
+++ b/fluss-rust/crates/fluss/src/lib.rs
@@ -44,11 +44,11 @@
 //!
 //!     // ---- Primary key (KV) table: upsert and lookup ----
 //!     let kv_path = TablePath::new("fluss", "users");
-//!     let mut kv_schema = Schema::builder()
+//!     let kv_schema = Schema::builder()
 //!         .column("id", DataTypes::int())
 //!         .column("name", DataTypes::string())
 //!         .column("age", DataTypes::bigint())
-//!         .primary_key(vec!["id"]);
+//!         .primary_key(vec!["id"])?;
 //!     let kv_descriptor = TableDescriptor::builder()
 //!         .schema(kv_schema.build()?)
 //!         .build()?;
diff --git a/fluss-rust/crates/fluss/src/metadata/json_serde.rs b/fluss-rust/crates/fluss/src/metadata/json_serde.rs
index c504660e92b..7e79bb65b62 100644
--- a/fluss-rust/crates/fluss/src/metadata/json_serde.rs
+++ b/fluss-rust/crates/fluss/src/metadata/json_serde.rs
@@ -527,7 +527,7 @@ impl JsonSerde for Schema {
                 );
             }
 
-            schema_builder = schema_builder.primary_key(primary_keys);
+            schema_builder = schema_builder.primary_key(primary_keys)?;
         }
 
         if let Some(auto_increment_node) = node.get(Self::AUTO_INCREMENT_COLUMN_NAME) {
@@ -763,6 +763,7 @@ mod tests {
             .column("id", DataTypes::int())
             .column("seq", DataTypes::bigint())
             .primary_key(["id"])
+            .unwrap()
             .enable_auto_increment("seq")
             .unwrap()
             .build()
@@ -789,6 +790,7 @@ mod tests {
         let schema = Schema::builder()
             .column("id", DataTypes::int())
             .primary_key(["id"])
+            .unwrap()
             .build()
             .unwrap();
 
diff --git a/fluss-rust/crates/fluss/src/metadata/schema_util.rs b/fluss-rust/crates/fluss/src/metadata/schema_util.rs
index 498a526eaef..3952e6e1261 100644
--- a/fluss-rust/crates/fluss/src/metadata/schema_util.rs
+++ b/fluss-rust/crates/fluss/src/metadata/schema_util.rs
@@ -196,6 +196,7 @@ mod tests {
         let origin = Schema::builder()
             .column("a", DataTypes::int())
             .primary_key(["a"])
+            .unwrap()
             .build()
             .unwrap();
         let expected = schema_with_ids(&[(0, "a", DataTypes::int())]);
diff --git a/fluss-rust/crates/fluss/src/metadata/table.rs b/fluss-rust/crates/fluss/src/metadata/table.rs
index c03a60065b2..d016dfb7821 100644
--- a/fluss-rust/crates/fluss/src/metadata/table.rs
+++ b/fluss-rust/crates/fluss/src/metadata/table.rs
@@ -291,7 +291,7 @@ impl SchemaBuilder {
         self
     }
 
-    pub fn primary_key<I, S>(self, column_names: I) -> Self
+    pub fn primary_key<I, S>(self, column_names: I) -> Result<Self>
     where
         I: IntoIterator<Item = S>,
         S: Into<String>,
@@ -307,12 +307,18 @@ impl SchemaBuilder {
         mut self,
         constraint_name: N,
         column_names: Vec<P>,
-    ) -> Self {
+    ) -> Result<Self> {
+        if self.primary_key.is_some() {
+            return Err(IllegalArgument {
+                message: "Multiple primary keys are not supported.".to_string(),
+            });
+        }
+
         self.primary_key = Some(PrimaryKey::new(
             constraint_name.into(),
             column_names.into_iter().map(|s| s.into()).collect(),
         ));
-        self
+        Ok(self)
     }
 
     /// Declares a column to be auto-incremented. With an auto-increment column in the table,
@@ -531,6 +537,19 @@ impl SchemaBuilder {
             return Ok(columns.to_vec());
         };
 
+        if pk.column_names.is_empty() {
+            return Err(Error::invalid_table(
+                "Primary key constraint must be defined for at least a single column.",
+            ));
+        }
+
+        let primary_key_names: Vec<_> = pk.column_names.iter().collect();
+        if let Some(duplicates) = Self::find_duplicates(&primary_key_names) {
+            return Err(Error::invalid_table(format!(
+                "Primary key constraint must not contain duplicate columns. Found: {duplicates:?}"
+            )));
+        }
+
         let pk_set: HashSet<_> = pk.column_names.iter().collect();
         let all_columns: HashSet<_> = columns.iter().map(|c| &c.name).collect();
         if !pk_set.is_subset(&all_columns) {
@@ -1652,6 +1671,48 @@ mod tests {
     use super::*;
     use crate::metadata::DataTypes;
 
+    #[test]
+    fn invalid_primary_keys_are_rejected() {
+        for (primary_keys, expected_message) in [
+            (
+                Vec::<&str>::new(),
+                "Primary key constraint must be defined for at least a single column.",
+            ),
+            (
+                vec!["id", "id"],
+                "Primary key constraint must not contain duplicate columns.",
+            ),
+        ] {
+            let err = Schema::builder()
+                .column("id", DataTypes::int())
+                .primary_key(primary_keys)
+                .unwrap()
+                .build()
+                .unwrap_err();
+
+            assert!(
+                err.to_string().contains(expected_message),
+                "unexpected error: {err}"
+            );
+        }
+    }
+
+    #[test]
+    fn multiple_primary_keys_are_rejected() {
+        let err = Schema::builder()
+            .column("id", DataTypes::int())
+            .primary_key(["id"])
+            .unwrap()
+            .primary_key_named("another_pk", vec!["id"])
+            .unwrap_err();
+
+        assert!(
+            err.to_string()
+                .contains("Multiple primary keys are not supported."),
+            "unexpected error: {err}"
+        );
+    }
+
     #[test]
     fn auto_increment_column_requires_a_primary_key_table() {
         let err = Schema::builder()
@@ -1674,6 +1735,7 @@ mod tests {
             .column("id", DataTypes::bigint())
             .column("name", DataTypes::string())
             .primary_key(["id"])
+            .unwrap()
             .enable_auto_increment("id")
             .unwrap()
             .build()
@@ -1691,6 +1753,7 @@ mod tests {
             .column("id", DataTypes::int())
             .column("seq", DataTypes::string())
             .primary_key(["id"])
+            .unwrap()
             .enable_auto_increment("seq")
             .unwrap()
             .build()
@@ -1706,6 +1769,7 @@ mod tests {
                 .column("id", DataTypes::int())
                 .column("seq", accepted)
                 .primary_key(["id"])
+                .unwrap()
                 .enable_auto_increment("seq")
                 .unwrap()
                 .build()
@@ -1771,6 +1835,7 @@ mod tests {
             .column("id", DataTypes::int())
             .column("name", DataTypes::string())
             .primary_key(vec!["id".to_string()])
+            .unwrap()
             .build()
             .unwrap();
 
diff --git a/fluss-rust/crates/fluss/tests/integration/admin.rs b/fluss-rust/crates/fluss/tests/integration/admin.rs
index 13e763a36f8..64218abdf3d 100644
--- a/fluss-rust/crates/fluss/tests/integration/admin.rs
+++ b/fluss-rust/crates/fluss/tests/integration/admin.rs
@@ -99,6 +99,7 @@ mod admin_test {
             .with_comment("User's age (optional)")
             .column("email", DataTypes::string())
             .primary_key(vec!["id".to_string()])
+            .unwrap()
             .build()
             .expect("Failed to build table schema");
 
@@ -217,6 +218,7 @@ mod admin_test {
                     .column("name", DataTypes::string())
                     .column("seq", DataTypes::bigint())
                     .primary_key(vec!["id".to_string()])
+                    .unwrap()
                     .enable_auto_increment("seq")
                     .expect("Failed to enable auto increment")
                     .build()
@@ -292,6 +294,7 @@ mod admin_test {
             .column("dt", DataTypes::string())
             .column("region", DataTypes::string())
             .primary_key(vec!["id", "dt", "region"])
+            .unwrap()
             .build()
             .expect("Failed to build table schema");
 
diff --git a/fluss-rust/crates/fluss/tests/integration/batch_scanner.rs b/fluss-rust/crates/fluss/tests/integration/batch_scanner.rs
index 5d10b59103a..57002cde88d 100644
--- a/fluss-rust/crates/fluss/tests/integration/batch_scanner.rs
+++ b/fluss-rust/crates/fluss/tests/integration/batch_scanner.rs
@@ -363,6 +363,7 @@ mod batch_scanner_test {
                     .column("id", DataTypes::int())
                     .column("name", DataTypes::string())
                     .primary_key(vec!["id"])
+                    .unwrap()
                     .build()
                     .expect("schema"),
             )
diff --git a/fluss-rust/crates/fluss/tests/integration/kv_changelog.rs b/fluss-rust/crates/fluss/tests/integration/kv_changelog.rs
index 30a85fd30e0..d1ba06d9224 100644
--- a/fluss-rust/crates/fluss/tests/integration/kv_changelog.rs
+++ b/fluss-rust/crates/fluss/tests/integration/kv_changelog.rs
@@ -46,6 +46,7 @@ mod kv_changelog_test {
                     .column("id", DataTypes::int())
                     .column("name", DataTypes::string())
                     .primary_key(vec!["id"])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
diff --git a/fluss-rust/crates/fluss/tests/integration/kv_table.rs b/fluss-rust/crates/fluss/tests/integration/kv_table.rs
index 965f06bbbf6..0aa1ec44634 100644
--- a/fluss-rust/crates/fluss/tests/integration/kv_table.rs
+++ b/fluss-rust/crates/fluss/tests/integration/kv_table.rs
@@ -59,6 +59,7 @@ mod kv_table_test {
                     .column("name", DataTypes::string())
                     .column("age", DataTypes::bigint())
                     .primary_key(vec!["id"])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
@@ -198,6 +199,7 @@ mod kv_table_test {
                     .column("user_id", DataTypes::int())
                     .column("score", DataTypes::bigint())
                     .primary_key(vec!["region", "user_id"])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
@@ -317,6 +319,7 @@ mod kv_table_test {
                     )
                     .column("tags", DataTypes::array(DataTypes::string()))
                     .primary_key(vec!["id"])
+                    .unwrap()
                     .build()
                     .expect("schema"),
             )
@@ -502,6 +505,7 @@ mod kv_table_test {
                     )
                     .column("tags", DataTypes::array(DataTypes::string()))
                     .primary_key(vec!["region", "user_id"])
+                    .unwrap()
                     .build()
                     .expect("schema"),
             )
@@ -685,6 +689,7 @@ mod kv_table_test {
                     .column("id", DataTypes::int())
                     .column("name", DataTypes::string())
                     .primary_key(vec!["region", "id"])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
@@ -799,6 +804,7 @@ mod kv_table_test {
                     .column("c", DataTypes::bigint())
                     .column("d", DataTypes::string())
                     .primary_key(vec!["a", "b", "c"])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
@@ -905,6 +911,7 @@ mod kv_table_test {
                     .column("c", DataTypes::bigint())
                     .column("d", DataTypes::string())
                     .primary_key(vec!["region", "a", "b", "c"])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
@@ -1015,6 +1022,7 @@ mod kv_table_test {
                     .column("b", DataTypes::string())
                     .column("c", DataTypes::bigint())
                     .primary_key(vec!["a", "b", "c"])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
@@ -1070,6 +1078,7 @@ mod kv_table_test {
                     .column("name", DataTypes::string())
                     .column("value", DataTypes::bigint())
                     .primary_key(vec!["id".to_string()])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
@@ -1202,6 +1211,7 @@ mod kv_table_test {
                     .column("b", DataTypes::string())
                     .column("c", DataTypes::string())
                     .primary_key(vec!["a", "b"])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
@@ -1273,6 +1283,7 @@ mod kv_table_test {
                     .column("id", DataTypes::int())
                     .column("name", DataTypes::string())
                     .primary_key(vec!["id"])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
@@ -1380,6 +1391,7 @@ mod kv_table_test {
                     .column("col_bytes", DataTypes::bytes())
                     .column("col_binary", DataTypes::binary(4))
                     .primary_key(vec!["id"])
+                    .unwrap()
                     .build()
                     .expect("Failed to build schema"),
             )
diff --git a/fluss-rust/crates/fluss/tests/integration/utils.rs b/fluss-rust/crates/fluss/tests/integration/utils.rs
index 2ad0e6c368a..2bf5dea10ce 100644
--- a/fluss-rust/crates/fluss/tests/integration/utils.rs
+++ b/fluss-rust/crates/fluss/tests/integration/utils.rs
@@ -500,7 +500,7 @@ impl ColumnPlan {
             sb = sb.column(*n, dt.clone());
         }
         if let Some(keys) = pk {
-            sb = sb.primary_key(keys.iter().copied());
+            sb = sb.primary_key(keys.iter().copied()).expect("primary key");
         }
         sb.build().expect("schema build")
     }
diff --git a/fluss-rust/website/docs/user-guide/rust/example/partitioned-tables.md b/fluss-rust/website/docs/user-guide/rust/example/partitioned-tables.md
index e583e06ead7..56e4f22b21e 100644
--- a/fluss-rust/website/docs/user-guide/rust/example/partitioned-tables.md
+++ b/fluss-rust/website/docs/user-guide/rust/example/partitioned-tables.md
@@ -147,7 +147,7 @@ let table_descriptor = TableDescriptor::builder()
             .column("region", DataTypes::string())
             .column("zone", DataTypes::bigint())
             .column("score", DataTypes::bigint())
-            .primary_key(vec!["user_id", "region", "zone"])
+            .primary_key(vec!["user_id", "region", "zone"])?
             .build()?,
     )
     .partitioned_by(vec!["region", "zone"])
diff --git a/fluss-rust/website/docs/user-guide/rust/example/prefix-lookup.md b/fluss-rust/website/docs/user-guide/rust/example/prefix-lookup.md
index 619ba8341d3..b8cbf388cb2 100644
--- a/fluss-rust/website/docs/user-guide/rust/example/prefix-lookup.md
+++ b/fluss-rust/website/docs/user-guide/rust/example/prefix-lookup.md
@@ -28,7 +28,7 @@ let table_descriptor = TableDescriptor::builder()
             .column("session_id", DataTypes::string())
             .column("event_seq", DataTypes::bigint())
             .column("event_data", DataTypes::string())
-            .primary_key(vec!["user_id", "session_id", "event_seq"])
+            .primary_key(vec!["user_id", "session_id", "event_seq"])?
             .build()?,
     )
     // Bucket key (user_id, session_id) is a prefix of the primary key.
@@ -75,7 +75,7 @@ let table_descriptor = TableDescriptor::builder()
             .column("session_id", DataTypes::string())
             .column("event_seq", DataTypes::bigint())
             .column("event_data", DataTypes::string())
-            .primary_key(vec!["region", "user_id", "session_id", "event_seq"])
+            .primary_key(vec!["region", "user_id", "session_id", "event_seq"])?
             .build()?,
     )
     .partitioned_by(vec!["region"])
diff --git a/fluss-rust/website/docs/user-guide/rust/example/primary-key-tables.md b/fluss-rust/website/docs/user-guide/rust/example/primary-key-tables.md
index b3d03823e86..5645ff8f123 100644
--- a/fluss-rust/website/docs/user-guide/rust/example/primary-key-tables.md
+++ b/fluss-rust/website/docs/user-guide/rust/example/primary-key-tables.md
@@ -16,7 +16,7 @@ let table_descriptor = TableDescriptor::builder()
             .column("id", DataTypes::int())
             .column("name", DataTypes::string())
             .column("age", DataTypes::bigint())
-            .primary_key(vec!["id"])
+            .primary_key(vec!["id"])?
             .build()?,
     )
     .build()?;
diff --git a/website/docs/apis/rust/example/partitioned-tables.md b/website/docs/apis/rust/example/partitioned-tables.md
index e583e06ead7..56e4f22b21e 100644
--- a/website/docs/apis/rust/example/partitioned-tables.md
+++ b/website/docs/apis/rust/example/partitioned-tables.md
@@ -147,7 +147,7 @@ let table_descriptor = TableDescriptor::builder()
             .column("region", DataTypes::string())
             .column("zone", DataTypes::bigint())
             .column("score", DataTypes::bigint())
-            .primary_key(vec!["user_id", "region", "zone"])
+            .primary_key(vec!["user_id", "region", "zone"])?
             .build()?,
     )
     .partitioned_by(vec!["region", "zone"])
diff --git a/website/docs/apis/rust/example/prefix-lookup.md b/website/docs/apis/rust/example/prefix-lookup.md
index 619ba8341d3..b8cbf388cb2 100644
--- a/website/docs/apis/rust/example/prefix-lookup.md
+++ b/website/docs/apis/rust/example/prefix-lookup.md
@@ -28,7 +28,7 @@ let table_descriptor = TableDescriptor::builder()
             .column("session_id", DataTypes::string())
             .column("event_seq", DataTypes::bigint())
             .column("event_data", DataTypes::string())
-            .primary_key(vec!["user_id", "session_id", "event_seq"])
+            .primary_key(vec!["user_id", "session_id", "event_seq"])?
             .build()?,
     )
     // Bucket key (user_id, session_id) is a prefix of the primary key.
@@ -75,7 +75,7 @@ let table_descriptor = TableDescriptor::builder()
             .column("session_id", DataTypes::string())
             .column("event_seq", DataTypes::bigint())
             .column("event_data", DataTypes::string())
-            .primary_key(vec!["region", "user_id", "session_id", "event_seq"])
+            .primary_key(vec!["region", "user_id", "session_id", "event_seq"])?
             .build()?,
     )
     .partitioned_by(vec!["region"])
diff --git a/website/docs/apis/rust/example/primary-key-tables.md b/website/docs/apis/rust/example/primary-key-tables.md
index b3d03823e86..5645ff8f123 100644
--- a/website/docs/apis/rust/example/primary-key-tables.md
+++ b/website/docs/apis/rust/example/primary-key-tables.md
@@ -16,7 +16,7 @@ let table_descriptor = TableDescriptor::builder()
             .column("id", DataTypes::int())
             .column("name", DataTypes::string())
             .column("age", DataTypes::bigint())
-            .primary_key(vec!["id"])
+            .primary_key(vec!["id"])?
             .build()?,
     )
     .build()?;