Repository: risingwavelabs/risingwave (language: python)

Issue:
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:

Fix patch:
diff --git a/src/frontend/src/handler/create_sink.rs b/src/frontend/src/handler/create_sink.rs
index 5a45d39d132a..03e58ab4de3c 100644
--- a/src/frontend/src/handler/create_sink.rs
+++ b/src/frontend/src/handler/create_sink.rs
@@ -369,77 +369,85 @@ pub async fn get_partition_compute_info(
 }
 
 async fn get_partition_compute_info_for_iceberg(
-    iceberg_config: &IcebergConfig,
+    _iceberg_config: &IcebergConfig,
 ) -> Result<Option<PartitionComputeInfo>> {
-    // TODO: check table if exists
-    if iceberg_config.create_table_if_not_exists {
-        return Ok(None);
-    }
-    let table = iceberg_config.load_table().await?;
-    let Some(partition_spec) = table.current_table_metadata().current_partition_spec().ok() else {
-        return Ok(None);
-    };
+    // TODO: enable partition compute for iceberg after fixing the issue of sink decoupling.
+    return Ok(None);
 
-    if partition_spec.is_unpartitioned() {
-        return Ok(None);
-    }
+    #[allow(unreachable_code)]
+    {
+        // TODO: check table if exists
+        if _iceberg_config.create_table_if_not_exists {
+            return Ok(None);
+        }
+        let table = _iceberg_config.load_table().await?;
+        let Some(partition_spec) = table.current_table_metadata().current_partition_spec().ok()
+        else {
+            return Ok(None);
+        };
 
-    // Separate the partition spec into two parts: sparse partition and range partition.
-    // Sparse partition means that the data distribution is more sparse at a given time.
-    // Range partition means that the data distribution is likely same at a given time.
-    // Only compute the partition and shuffle by them for the sparse partition.
-    let has_sparse_partition = partition_spec.fields.iter().any(|f| match f.transform {
-        // Sparse partition
-        icelake::types::Transform::Identity
-        | icelake::types::Transform::Truncate(_)
-        | icelake::types::Transform::Bucket(_) => true,
-        // Range partition
-        icelake::types::Transform::Year
-        | icelake::types::Transform::Month
-        | icelake::types::Transform::Day
-        | icelake::types::Transform::Hour
-        | icelake::types::Transform::Void => false,
-    });
+        if partition_spec.is_unpartitioned() {
+            return Ok(None);
+        }
 
-    if !has_sparse_partition {
-        return Ok(None);
-    }
+        // Separate the partition spec into two parts: sparse partition and range partition.
+        // Sparse partition means that the data distribution is more sparse at a given time.
+        // Range partition means that the data distribution is likely same at a given time.
+        // Only compute the partition and shuffle by them for the sparse partition.
+        let has_sparse_partition = partition_spec.fields.iter().any(|f| match f.transform {
+            // Sparse partition
+            icelake::types::Transform::Identity
+            | icelake::types::Transform::Truncate(_)
+            | icelake::types::Transform::Bucket(_) => true,
+            // Range partition
+            icelake::types::Transform::Year
+            | icelake::types::Transform::Month
+            | icelake::types::Transform::Day
+            | icelake::types::Transform::Hour
+            | icelake::types::Transform::Void => false,
+        });
 
-    let arrow_type: ArrowDataType = table
-        .current_partition_type()
-        .map_err(|err| RwError::from(ErrorCode::SinkError(err.into())))?
-        .try_into()
-        .map_err(|_| {
-            RwError::from(ErrorCode::SinkError(
-                "Fail to convert iceberg partition type to arrow type".into(),
-            ))
-        })?;
-    let Some(schema) = table.current_table_metadata().current_schema().ok() else {
-        return Ok(None);
-    };
-    let partition_fields = partition_spec
-        .fields
-        .iter()
-        .map(|f| {
-            let source_f = schema
-                .look_up_field_by_id(f.source_column_id)
-                .ok_or(RwError::from(ErrorCode::SinkError(
-                    "Fail to look up iceberg partition field".into(),
-                )))?;
-            Ok((source_f.name.clone(), f.transform))
-        })
-        .collect::<Result<Vec<_>>>()?;
-
-    let ArrowDataType::Struct(partition_type) = arrow_type else {
-        return Err(RwError::from(ErrorCode::SinkError(
-            "Partition type of iceberg should be a struct type".into(),
-        )));
-    };
+        if !has_sparse_partition {
+            return Ok(None);
+        }
 
-    Ok(Some(PartitionComputeInfo::Iceberg(IcebergPartitionInfo {
-        partition_type: IcebergArrowConvert.struct_from_fields(&partition_type)?,
-        partition_fields,
-    })))
+        let arrow_type: ArrowDataType = table
+            .current_partition_type()
+            .map_err(|err| RwError::from(ErrorCode::SinkError(err.into())))?
+            .try_into()
+            .map_err(|_| {
+                RwError::from(ErrorCode::SinkError(
+                    "Fail to convert iceberg partition type to arrow type".into(),
+                ))
+            })?;
+        let Some(schema) = table.current_table_metadata().current_schema().ok() else {
+            return Ok(None);
+        };
+        let partition_fields = partition_spec
+            .fields
+            .iter()
+            .map(|f| {
+                let source_f =
+                    schema
+                        .look_up_field_by_id(f.source_column_id)
+                        .ok_or(RwError::from(ErrorCode::SinkError(
+                            "Fail to look up iceberg partition field".into(),
+                        )))?;
+                Ok((source_f.name.clone(), f.transform))
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        let ArrowDataType::Struct(partition_type) = arrow_type else {
+            return Err(RwError::from(ErrorCode::SinkError(
+                "Partition type of iceberg should be a struct type".into(),
+            )));
+        };
+
+        Ok(Some(PartitionComputeInfo::Iceberg(IcebergPartitionInfo {
+            partition_type: IcebergArrowConvert.struct_from_fields(&partition_type)?,
+            partition_fields,
+        })))
+    }
 }
 
 pub async fn handle_create_sink(


Test patch:
diff --git a/src/frontend/planner_test/tests/testdata/output/sink.yaml b/src/frontend/planner_test/tests/testdata/output/sink.yaml
index db5d533223cb..0bc336bd21a2 100644
--- a/src/frontend/planner_test/tests/testdata/output/sink.yaml
+++ b/src/frontend/planner_test/tests/testdata/output/sink.yaml
@@ -197,9 +197,7 @@
     );
   explain_output: |
     StreamSink { type: append-only, columns: [v1, v2, v3, v4, t1._row_id(hidden)] }
-    └─StreamExchange { dist: HashShard($expr1) }
-      └─StreamProject { exprs: [t1.v1, t1.v2, t1.v3, t1.v4, t1._row_id, Row(t1.v1, IcebergTransform('bucket[1]':Varchar, t1.v2), IcebergTransform('truncate[1]':Varchar, t1.v3), null:Int32) as $expr1] }
-        └─StreamTableScan { table: t1, columns: [v1, v2, v3, v4, _row_id] }
+    └─StreamTableScan { table: t1, columns: [v1, v2, v3, v4, _row_id] }
 - id: create_mock_iceberg_sink_append_only_with_range_partition
   sql: |
     create table t1 (v1 date, v2 timestamp, v3 timestamp with time zone, v4 timestamp);
@@ -239,9 +237,7 @@
     );
   explain_output: |
     StreamSink { type: upsert, columns: [v1, v2, v3, v4, t1._row_id(hidden)], downstream_pk: [t1.v1] }
-    └─StreamExchange { dist: HashShard($expr1) }
-      └─StreamProject { exprs: [t1.v1, t1.v2, t1.v3, t1.v4, t1._row_id, Row(t1.v1, IcebergTransform('bucket[1]':Varchar, t1.v2), IcebergTransform('truncate[1]':Varchar, t1.v3), null:Int32) as $expr1] }
-        └─StreamTableScan { table: t1, columns: [v1, v2, v3, v4, _row_id] }
+    └─StreamTableScan { table: t1, columns: [v1, v2, v3, v4, _row_id] }
 - id: create_mock_iceberg_sink_upsert_with_range_partition
   sql: |
     create table t1 (v1 date, v2 timestamp, v3 timestamp with time zone, v4 timestamp);