Re: [PR] [FLUSS][Spark] Add set_cluster_configs and reset_cluster_configs procedures [fluss]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #3204:
URL: https://github.com/apache/fluss/pull/3204#discussion_r3151491314


##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/SetClusterConfigsProcedure.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.spark.procedure
+
+import org.apache.fluss.config.cluster.{AlterConfig, AlterConfigOpType}
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
+import org.apache.spark.unsafe.types.UTF8String
+
+import scala.collection.JavaConverters._
+
+/**
+ * Procedure to set cluster configuration dynamically.
+ *
+ * This procedure allows modifying dynamic cluster configurations. The changes are:
+ *   - Validated by the CoordinatorServer before persistence
+ *   - Persisted in ZooKeeper for durability
+ *   - Applied to all relevant servers (Coordinator and TabletServers)
+ *   - Survive server restarts
+ *
+ * Usage examples:
+ * {{{
+ * -- Set a single configuration
+ * CALL sys.set_cluster_configs(config_pairs => array('key1', 'value1'))
+ *
+ * -- Set multiple configurations at once
+ * CALL sys.set_cluster_configs(config_pairs => array('key1', 'value1', 'key2', 'value2'))
+ * }}}
+ *
+ * Note: Not all configurations support dynamic changes. The server will validate the change and
+ * reject it if the configuration cannot be modified dynamically or if the new value is invalid.
+ */
+class SetClusterConfigsProcedure(tableCatalog: TableCatalog) extends BaseProcedure(tableCatalog) {
+
+  override def parameters(): Array[ProcedureParameter] = {
+    SetClusterConfigsProcedure.PARAMETERS
+  }
+
+  override def outputType(): StructType = {
+    SetClusterConfigsProcedure.OUTPUT_TYPE
+  }
+
+  override def call(args: InternalRow): Array[InternalRow] = {
+    val configPairs = if (args.numFields > 0 && !args.isNullAt(0)) {
+      val pairsArray = args.getArray(0)
+      (0 until pairsArray.numElements())
+        .map(i => pairsArray.getUTF8String(i).toString)

Review Comment:
   `pairsArray.getUTF8String(i)` will throw if any element in `config_pairs` is NULL. Since Spark SQL arrays can contain nulls, this can lead to a confusing runtime failure (Spark internal error) rather than a clear validation error. Consider checking `pairsArray.isNullAt(i)` and throwing an `IllegalArgumentException` with a user-facing message that identifies the null element (and ideally its position/key).
   ```suggestion
           .map { i =>
             if (pairsArray.isNullAt(i)) {
               throw new IllegalArgumentException(
                 s"config_pairs contains a null element at position $i. " +
                   "Please specify valid configuration key/value pairs.")
             }
             pairsArray.getUTF8String(i).toString
           }
   ```



##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/SetClusterConfigsProcedure.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.spark.procedure
+
+import org.apache.fluss.config.cluster.{AlterConfig, AlterConfigOpType}
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
+import org.apache.spark.unsafe.types.UTF8String
+
+import scala.collection.JavaConverters._
+

Review Comment:
   This import is unused in the file (no Java/Scala collection conversions are performed). Removing it avoids warnings and keeps the new file minimal.
   ```suggestion
   
   ```



##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/ResetClusterConfigsProcedure.scala:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.spark.procedure
+
+import org.apache.fluss.config.cluster.{AlterConfig, AlterConfigOpType}
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
+import org.apache.spark.unsafe.types.UTF8String
+
+import scala.collection.JavaConverters._
+
+/**
+ * Procedure to reset cluster configurations to their default values.
+ *
+ * This procedure reverts the configurations to their initial system defaults. The changes are:
+ *   - Validated by the CoordinatorServer before persistence
+ *   - Persisted in ZooKeeper for durability
+ *   - Applied to all relevant servers (Coordinator and TabletServers)
+ *   - Survive server restarts
+ *
+ * Usage examples:
+ * {{{
+ * -- Reset a single configuration
+ * CALL sys.reset_cluster_configs(config_keys => array('key1'))
+ *
+ * -- Reset multiple configurations at once
+ * CALL sys.reset_cluster_configs(config_keys => array('key1', 'key2'))
+ * }}}
+ */
+class ResetClusterConfigsProcedure(tableCatalog: TableCatalog) extends BaseProcedure(tableCatalog) {
+
+  override def parameters(): Array[ProcedureParameter] = {
+    ResetClusterConfigsProcedure.PARAMETERS
+  }
+
+  override def outputType(): StructType = {
+    ResetClusterConfigsProcedure.OUTPUT_TYPE
+  }
+
+  override def call(args: InternalRow): Array[InternalRow] = {
+    val configKeys = if (args.numFields > 0 && !args.isNullAt(0)) {
+      val keysArray = args.getArray(0)
+      (0 until keysArray.numElements())
+        .map(i => keysArray.getUTF8String(i).toString)

Review Comment:
   `keysArray.getUTF8String(i)` will throw if `config_keys` contains NULL. This should be validated explicitly so callers get a deterministic `IllegalArgumentException` (e.g., “config_keys must not contain nulls”) instead of a Spark internal error.
   ```suggestion
           .map { i =>
             if (keysArray.isNullAt(i)) {
               throw new IllegalArgumentException("config_keys must not contain nulls")
             }
             keysArray.getUTF8String(i).toString
           }
   ```



##########
website/docs/engine-spark/procedures.md:
##########
@@ -87,6 +87,69 @@ CALL sys.get_cluster_configs();
 CALL sys.get_cluster_configs(config_keys => ARRAY('kv.rocksdb.shared-rate-limiter.bytes-per-sec', 'datalake.format'));
 ```
 
+### set_cluster_configs
+
+Set cluster configuration values dynamically. The changes are validated, persisted, and applied to all servers without requiring a restart.
+
+**Syntax:**
+
+```sql
+CALL [catalog_name.]sys.set_cluster_configs(config_pairs => ARRAY('key1', 'value1', 'key2', 'value2'))

Review Comment:
   The syntax examples in this doc section omit the trailing semicolon, while the surrounding examples (e.g., `CALL sys.get_cluster_configs();`) include semicolons. For consistency and copy/paste friendliness, consider adding `;` to the end of the syntax lines.



##########
website/docs/engine-spark/procedures.md:
##########
@@ -87,6 +87,69 @@ CALL sys.get_cluster_configs();
 CALL sys.get_cluster_configs(config_keys => ARRAY('kv.rocksdb.shared-rate-limiter.bytes-per-sec', 'datalake.format'));
 ```
 
+### set_cluster_configs
+
+Set cluster configuration values dynamically. The changes are validated, persisted, and applied to all servers without requiring a restart.
+
+**Syntax:**
+
+```sql
+CALL [catalog_name.]sys.set_cluster_configs(config_pairs => ARRAY('key1', 'value1', 'key2', 'value2'))
+```
+
+**Parameters:**
+
+- `config_pairs` (required): Array of configuration key-value pairs. Keys and values must alternate (key1, value1, key2, value2, ...).
+
+**Returns:** A table with columns:
+
+- `config_key`: The configuration key that was set
+- `config_value`: The value that was set
+- `result`: A message indicating the result of the operation
+
+**Example:**
+
+```sql title="Spark SQL"
+-- Set a single configuration
+CALL sys.set_cluster_configs(config_pairs => ARRAY('kv.rocksdb.shared-rate-limiter.bytes-per-sec', '200MB'));
+
+-- Set multiple configurations at once
+CALL sys.set_cluster_configs(config_pairs => ARRAY('kv.rocksdb.shared-rate-limiter.bytes-per-sec', '200MB', 'datalake.format', 'paimon'));
+```
+
+:::note
+Not all configurations support dynamic changes. The server will validate the change and reject it if the configuration cannot be modified dynamically or if the new value is invalid.
+:::
+
+### reset_cluster_configs
+
+Reset cluster configurations to their default values. The changes are validated, persisted, and applied to all servers without requiring a restart.
+
+**Syntax:**
+
+```sql
+CALL [catalog_name.]sys.reset_cluster_configs(config_keys => ARRAY('key1', 'key2'))

Review Comment:
   Same consistency issue as `set_cluster_configs`: consider adding a trailing semicolon in the syntax block to match other procedure examples in this page.
   ```suggestion
   CALL [catalog_name.]sys.reset_cluster_configs(config_keys => ARRAY('key1', 'key2'));
   ```



##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/SetClusterConfigsProcedure.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.spark.procedure
+
+import org.apache.fluss.config.cluster.{AlterConfig, AlterConfigOpType}
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
+import org.apache.spark.unsafe.types.UTF8String
+
+import scala.collection.JavaConverters._
+
+/**
+ * Procedure to set cluster configuration dynamically.
+ *
+ * This procedure allows modifying dynamic cluster configurations. The changes are:
+ *   - Validated by the CoordinatorServer before persistence
+ *   - Persisted in ZooKeeper for durability
+ *   - Applied to all relevant servers (Coordinator and TabletServers)
+ *   - Survive server restarts
+ *
+ * Usage examples:
+ * {{{
+ * -- Set a single configuration
+ * CALL sys.set_cluster_configs(config_pairs => array('key1', 'value1'))
+ *
+ * -- Set multiple configurations at once
+ * CALL sys.set_cluster_configs(config_pairs => array('key1', 'value1', 'key2', 'value2'))
+ * }}}
+ *
+ * Note: Not all configurations support dynamic changes. The server will validate the change and
+ * reject it if the configuration cannot be modified dynamically or if the new value is invalid.
+ */
+class SetClusterConfigsProcedure(tableCatalog: TableCatalog) extends BaseProcedure(tableCatalog) {
+
+  override def parameters(): Array[ProcedureParameter] = {
+    SetClusterConfigsProcedure.PARAMETERS
+  }
+
+  override def outputType(): StructType = {
+    SetClusterConfigsProcedure.OUTPUT_TYPE
+  }
+
+  override def call(args: InternalRow): Array[InternalRow] = {
+    val configPairs = if (args.numFields > 0 && !args.isNullAt(0)) {
+      val pairsArray = args.getArray(0)
+      (0 until pairsArray.numElements())
+        .map(i => pairsArray.getUTF8String(i).toString)
+        .toArray
+    } else {
+      Array.empty[String]
+    }

Review Comment:
   The new parsing logic should be exercised with arrays containing NULL elements (e.g., `array('k1', null)` or `array(null, 'v1')`) to ensure you return a clear validation error rather than a Spark internal exception. Adding one or two negative tests around NULL elements would lock in the intended behavior once validation is added.



##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/procedure/ResetClusterConfigsProcedure.scala:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.spark.procedure
+
+import org.apache.fluss.config.cluster.{AlterConfig, AlterConfigOpType}
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
+import org.apache.spark.unsafe.types.UTF8String
+
+import scala.collection.JavaConverters._
+

Review Comment:
   This import is unused in the file. Please remove it to avoid warnings and keep the new file clean.
   ```suggestion
   
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org