feat: manage all catalogs of the iceberg table in a transactional manner (#23597)

Author: yezizp2012

e2e_test/iceberg/test_case/pure_slt/iceberg_engine.slt

@@ -90,34 +90,6 @@ DROP TABLE full_type_t;
 
 # test connector
 
-statement error
-create table datagen_t (
-  id BIGINT,
-  item_name VARCHAR,
-  description VARCHAR,
-  initial_bid BIGINT,
-  reserve BIGINT,
-  date_time TIMESTAMP,
-  expires TIMESTAMP,
-  seller BIGINT,
-  category BIGINT,
-  extra VARCHAR)
-with (
-  connector = 'datagen',
-  fields.id.kind = 'sequence',
-  fields.id.start = '1',
-  commit_checkpoint_interval = 1,
-  source_rate_limit = 1000
-) engine = iceberg;
-----
-db error: ERROR: Failed to run the query
-
-Caused by:
-  Not supported: source_rate_limit for iceberg table engine during table creation
-HINT: Please remove source_rate_limit from WITH options.
-
-
-
 statement ok
 create table datagen_t (
   id BIGINT,

e2e_test/iceberg/test_case/pure_slt/iceberg_engine_append_only.slt

@@ -92,35 +92,6 @@ DROP TABLE full_type_t;
 
 # test connector
 
-statement error
-create table datagen_t (
-  id BIGINT,
-  item_name VARCHAR,
-  description VARCHAR,
-  initial_bid BIGINT,
-  reserve BIGINT,
-  date_time TIMESTAMP,
-  expires TIMESTAMP,
-  seller BIGINT,
-  category BIGINT,
-  extra VARCHAR)
-APPEND ONLY
-with (
-  connector = 'datagen',
-  fields.id.kind = 'sequence',
-  fields.id.start = '1',
-  commit_checkpoint_interval = 1,
-  source_rate_limit = 1000
-) engine = iceberg;
-----
-db error: ERROR: Failed to run the query
-
-Caused by:
-  Not supported: source_rate_limit for iceberg table engine during table creation
-HINT: Please remove source_rate_limit from WITH options.
-
-
-
 statement ok
 create table datagen_t (
   id BIGINT,

src/common/src/catalog/mod.rs

@@ -137,6 +137,9 @@ pub const CDC_OFFSET_COLUMN_NAME: &str = "_rw_offset";
 pub const CDC_SOURCE_COLUMN_NUM: u32 = 3;
 pub const CDC_TABLE_NAME_COLUMN_NAME: &str = "_rw_table_name";
 
+pub const ICEBERG_SOURCE_PREFIX: &str = "__iceberg_source_";
+pub const ICEBERG_SINK_PREFIX: &str = "__iceberg_sink_";
+
 /// The local system catalog reader in the frontend node.
 pub trait SysCatalogReader: Sync + Send + 'static {
     /// Reads the data of the system catalog table.

src/frontend/src/catalog/table_catalog.rs

@@ -18,8 +18,8 @@ use std::collections::{HashMap, HashSet};
 use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use risingwave_common::catalog::{
-    ColumnCatalog, ColumnDesc, ConflictBehavior, CreateType, Engine, Field, Schema,
-    StreamJobStatus, TableDesc, TableId, TableVersionId,
+    ColumnCatalog, ColumnDesc, ConflictBehavior, CreateType, Engine, Field, ICEBERG_SINK_PREFIX,
+    ICEBERG_SOURCE_PREFIX, Schema, StreamJobStatus, TableDesc, TableId, TableVersionId,
 };
 use risingwave_common::hash::{VnodeCount, VnodeCountCompat};
 use risingwave_common::id::{JobId, SourceId};
@@ -211,9 +211,6 @@ pub struct TableCatalog {
     pub cdc_table_type: Option<ExternalCdcTableType>,
 }
 
-pub const ICEBERG_SOURCE_PREFIX: &str = "__iceberg_source_";
-pub const ICEBERG_SINK_PREFIX: &str = "__iceberg_sink_";
-
 #[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
 #[cfg_attr(test, derive(Default))]
 pub enum TableType {

src/frontend/src/handler/create_sink.rs

@@ -24,7 +24,9 @@ use pgwire::pg_response::{PgResponse, StatementType};
 use risingwave_common::array::arrow::IcebergArrowConvert;
 use risingwave_common::array::arrow::arrow_schema_iceberg::DataType as ArrowDataType;
 use risingwave_common::bail;
-use risingwave_common::catalog::{ColumnCatalog, ConnectionId, ObjectId, Schema, UserId};
+use risingwave_common::catalog::{
+    ColumnCatalog, ConnectionId, ICEBERG_SINK_PREFIX, ObjectId, Schema, UserId,
+};
 use risingwave_common::license::Feature;
 use risingwave_common::secret::LocalSecretManager;
 use risingwave_common::system_param::reader::SystemParamsRead;
@@ -577,6 +579,13 @@ pub async fn handle_create_sink(
         return Ok(resp);
     }
 
+    if stmt.sink_name.base_name().starts_with(ICEBERG_SINK_PREFIX) {
+        return Err(RwError::from(ErrorCode::InvalidInputSyntax(format!(
+            "Sink name cannot start with reserved prefix '{}'",
+            ICEBERG_SINK_PREFIX
+        ))));
+    }
+
     let (mut sink, graph, target_table_catalog, dependencies) = {
         let SinkPlanContext {
             query,

src/frontend/src/handler/create_source.rs

@@ -116,6 +116,7 @@ use validate::{SOURCE_ALLOWED_CONNECTION_CONNECTOR, SOURCE_ALLOWED_CONNECTION_SC
 mod additional_column;
 use additional_column::check_and_add_timestamp_column;
 pub use additional_column::handle_addition_columns;
+use risingwave_common::catalog::ICEBERG_SOURCE_PREFIX;
 use risingwave_common::id::SourceId;
 
 use crate::stream_fragmenter::GraphJobType;
@@ -1073,6 +1074,17 @@ pub async fn handle_create_source(
         return Ok(resp);
     }
 
+    if stmt
+        .source_name
+        .base_name()
+        .starts_with(ICEBERG_SOURCE_PREFIX)
+    {
+        return Err(RwError::from(InvalidInputSyntax(format!(
+            "Source name cannot start with reserved prefix '{}'",
+            ICEBERG_SOURCE_PREFIX
+        ))));
+    }
+
     if handler_args.with_options.is_empty() {
         return Err(RwError::from(InvalidInputSyntax(
             "missing WITH clause".to_owned(),

src/frontend/src/handler/create_table.rs

@@ -26,7 +26,8 @@ use pgwire::pg_response::{PgResponse, StatementType};
 use prost::Message as _;
 use risingwave_common::catalog::{
     CdcTableDesc, ColumnCatalog, ColumnDesc, ConflictBehavior, DEFAULT_SCHEMA_NAME, Engine,
-    ObjectId, RISINGWAVE_ICEBERG_ROW_ID, ROW_ID_COLUMN_NAME, TableId,
+    ICEBERG_SINK_PREFIX, ICEBERG_SOURCE_PREFIX, ObjectId, RISINGWAVE_ICEBERG_ROW_ID,
+    ROW_ID_COLUMN_NAME, TableId,
 };
 use risingwave_common::config::MetaBackend;
 use risingwave_common::global_jvm::Jvm;
@@ -61,12 +62,12 @@ use risingwave_sqlparser::ast::{
 use risingwave_sqlparser::parser::{IncludeOption, Parser};
 use thiserror_ext::AsReport;
 
+use super::RwPgResponse;
 use super::create_source::{CreateSourceType, SqlColumnStrategy, bind_columns_from_source};
-use super::{RwPgResponse, alter_streaming_rate_limit};
 use crate::binder::{Clause, SecureCompareContext, bind_data_type};
 use crate::catalog::root_catalog::SchemaPath;
 use crate::catalog::source_catalog::SourceCatalog;
-use crate::catalog::table_catalog::{ICEBERG_SINK_PREFIX, ICEBERG_SOURCE_PREFIX, TableVersion};
+use crate::catalog::table_catalog::TableVersion;
 use crate::catalog::{ColumnId, DatabaseId, SchemaId, SourceId, check_column_name_not_reserved};
 use crate::error::{ErrorCode, Result, RwError, bail_bind_error};
 use crate::expr::{Expr, ExprImpl, ExprRewriter};
@@ -99,7 +100,6 @@ use risingwave_connector::sink::iceberg::{
     SNAPSHOT_EXPIRATION_RETAIN_LAST, WRITE_MODE, parse_partition_by_exprs,
 };
 use risingwave_pb::ddl_service::create_iceberg_table_request::{PbSinkJobInfo, PbTableJobInfo};
-use risingwave_pb::meta::PbThrottleTarget;
 
 use crate::handler::create_sink::{SinkPlanContext, gen_sink_plan};
 
@@ -1394,7 +1394,7 @@ fn bind_cdc_table_schema(
 
 #[allow(clippy::too_many_arguments)]
 pub async fn handle_create_table(
-    mut handler_args: HandlerArgs,
+    handler_args: HandlerArgs,
     table_name: ObjectName,
     column_defs: Vec<ColumnDef>,
     wildcard_idx: Option<usize>,
@@ -1422,21 +1422,6 @@ pub async fn handle_create_table(
         risingwave_sqlparser::ast::Engine::Hummock => Engine::Hummock,
         risingwave_sqlparser::ast::Engine::Iceberg => Engine::Iceberg,
     };
-    if engine == Engine::Iceberg && handler_args.with_options.get_connector().is_some() {
-        // HACK: since we don't have atomic DDL, table with connector may lose data.
-        // FIXME: remove this after https://github.com/risingwavelabs/risingwave/issues/21863
-        if let Some(_rate_limit) = handler_args.with_options.insert(
-            OverwriteOptions::SOURCE_RATE_LIMIT_KEY.to_owned(),
-            "0".to_owned(),
-        ) {
-            // prevent user specified rate limit
-            return Err(ErrorCode::NotSupported(
-                "source_rate_limit for iceberg table engine during table creation".to_owned(),
-                "Please remove source_rate_limit from WITH options.".to_owned(),
-            )
-            .into());
-        }
-    }
 
     if let Either::Right(resp) = session.check_relation_name_duplicated(
         table_name.clone(),
@@ -2174,8 +2159,6 @@ pub async fn create_iceberg_engine_table(
     )
     .await?;
 
-    let has_connector = source.is_some();
-
     // before we create the table, ensure the JVM is initialized as we use jdbc catalog right now.
     // If JVM isn't initialized successfully, current not atomic ddl will result in a partially created iceberg engine table.
     let _ = Jvm::get_or_init()?;
@@ -2211,17 +2194,6 @@ pub async fn create_iceberg_engine_table(
         res?
     }
 
-    // TODO: remove it together with rate limit rewrite after we support atomic DDL in meta side.
-    if has_connector {
-        alter_streaming_rate_limit::handle_alter_streaming_rate_limit(
-            handler_args,
-            PbThrottleTarget::TableWithSource,
-            table_name,
-            -1,
-        )
-        .await?;
-    }
-
     Ok(())
 }
 

src/frontend/src/handler/drop_sink.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use pgwire::pg_response::{PgResponse, StatementType};
+use risingwave_common::catalog::ICEBERG_SINK_PREFIX;
 use risingwave_sqlparser::ast::ObjectName;
 
 use super::RwPgResponse;
@@ -56,6 +57,14 @@ pub async fn handle_drop_sink(
         sink
     };
 
+    if sink_name.starts_with(ICEBERG_SINK_PREFIX) {
+        return Err(crate::error::ErrorCode::NotSupported(
+            "Dropping Iceberg sinks is not supported".to_owned(),
+            "Please use DROP TABLE command.".to_owned(),
+        )
+        .into());
+    }
+
     let sink_id = sink.id;
 
     let catalog_writer = session.catalog_writer()?;

src/frontend/src/handler/drop_source.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use pgwire::pg_response::{PgResponse, StatementType};
+use risingwave_common::catalog::ICEBERG_SOURCE_PREFIX;
 use risingwave_sqlparser::ast::ObjectName;
 
 use super::RwPgResponse;
@@ -68,6 +69,14 @@ pub async fn handle_drop_source(
         }
     };
 
+    if source_name.starts_with(ICEBERG_SOURCE_PREFIX) {
+        return Err(crate::error::ErrorCode::NotSupported(
+            "Dropping Iceberg sources is not supported".to_owned(),
+            "Please use DROP TABLE command.".to_owned(),
+        )
+        .into());
+    }
+
     session.check_privilege_for_drop_alter(schema_name, &*source)?;
 
     let catalog_writer = session.catalog_writer()?;