Repository: risingwavelabs/risingwave
File: src/frontend/src/handler/create_table.rs (Rust)

Code before:
    // 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()?;

    let catalog_writer = session.catalog_writer()?;
    let action = match job_type {
        TableJobType::SharedCdcSource => LongRunningNotificationAction::MonitorBackfillJob,
        _ => LongRunningNotificationAction::DiagnoseBarrierLatency,
    };
    let res = execute_with_long_running_notification(
        catalog_writer.create_iceberg_table(
            PbTableJobInfo {
                source,
                table: Some(table.to_prost()),
                fragment_graph: Some(graph),
                job_type: job_type as _,
            },
            PbSinkJobInfo {
                sink: Some(sink_catalog.to_proto()),
                fragment_graph: Some(sink_graph),
            },
            iceberg_source_catalog.to_prost(),
            if_not_exists,
        ),
        &session,
        "CREATE ICEBERG TABLE",
        action,
    )
    .await;

    if res.is_err() {
        let _ = iceberg_catalog
            .drop_table(&table_identifier)
            .await
            .inspect_err(|err| {
                tracing::error!(
                    "failed to drop iceberg table {} after create iceberg engine table failed: {}",
                    table_identifier,
                    err.as_report()
                );
            });
        res?
    }

    Ok(())
}

pub fn check_create_table_with_source(
    with_options: &WithOptions,
    format_encode: Option<FormatEncodeOptions>,
    include_column_options: &IncludeOption,

Review comment (suggestion):
The operation name should be "CREATE TABLE" instead of "CREATE ICEBERG TABLE". Users execute `CREATE TABLE` statements regardless of the engine type. Exposing internal engine implementation details (Iceberg engine) in user-facing notifications is confusing and inconsistent with how the operation is presented to users.
```suggestion
        "CREATE TABLE",
```

Code after:
    // 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()?;

    let catalog_writer = session.catalog_writer()?;
    let action = match job_type {
        TableJobType::SharedCdcSource => LongRunningNotificationAction::MonitorBackfillJob,
        _ => LongRunningNotificationAction::DiagnoseBarrierLatency,
    };
    let res = execute_with_long_running_notification(
        catalog_writer.create_iceberg_table(
            PbTableJobInfo {
                source,
                table: Some(table.to_prost()),
                fragment_graph: Some(graph),
                job_type: job_type as _,
            },
            PbSinkJobInfo {
                sink: Some(sink_catalog.to_proto()),
                fragment_graph: Some(sink_graph),
            },
            iceberg_source_catalog.to_prost(),
            if_not_exists,
        ),
        &session,
        "CREATE TABLE",
        action,
    )
    .await;

    if res.is_err() {
        let _ = iceberg_catalog
            .drop_table(&table_identifier)
            .await
            .inspect_err(|err| {
                tracing::error!(
                    "failed to drop iceberg table {} after create iceberg engine table failed: {}",
                    table_identifier,
                    err.as_report()
                );
            });
        res?
    }

    Ok(())
}

pub fn check_create_table_with_source(
    with_options: &WithOptions,
    format_encode: Option<FormatEncodeOptions>,
    include_column_options: &IncludeOption,