diff --git a/src/frontend/src/catalog/purify.rs b/src/frontend/src/catalog/purify.rs
index ab61ba0b021ee..e221e97c39f6b 100644
--- a/src/frontend/src/catalog/purify.rs
+++ b/src/frontend/src/catalog/purify.rs
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use anyhow::Context;
+use itertools::Itertools;
 use prost::Message as _;
 use risingwave_common::bail;
 use risingwave_common::catalog::{ColumnCatalog, ColumnId};
@@ -103,11 +105,20 @@ pub fn try_purify_table_source_create_sql_ast(
 
             let default_value_option = column_def
                 .options
-                .extract_if(|o| matches!(o.option, ColumnOption::DefaultValue { .. }))
-                .next();
-
-            let expr = default_value_option.map(|o| match o.option {
-                ColumnOption::DefaultValue(expr) => expr,
+                .extract_if(|o| {
+                    matches!(
+                        o.option,
+                        ColumnOption::DefaultValue { .. }
+                            | ColumnOption::DefaultValuePersisted { .. }
+                    )
+                })
+                .at_most_one()
+                .ok()
+                .context("multiple default value options found")?;
+
+            let expr = default_value_option.and_then(|o| match o.option {
+                ColumnOption::DefaultValue(expr) => Some(expr),
+                ColumnOption::DefaultValuePersisted { expr, .. } => expr,
                 _ => unreachable!(),
             });
 
diff --git a/src/frontend/src/handler/create_table.rs b/src/frontend/src/handler/create_table.rs
index 6c1cfa69220fa..3d87b00046590 100644
--- a/src/frontend/src/handler/create_table.rs
+++ b/src/frontend/src/handler/create_table.rs
@@ -70,7 +70,7 @@ use crate::catalog::root_catalog::SchemaPath;
 use crate::catalog::source_catalog::SourceCatalog;
 use crate::catalog::table_catalog::{TableVersion, ICEBERG_SINK_PREFIX, ICEBERG_SOURCE_PREFIX};
 use crate::catalog::{check_valid_column_name, ColumnId, DatabaseId, SchemaId};
-use crate::error::{ErrorCode, Result, RwError};
+use crate::error::{bail_bind_error, ErrorCode, Result, RwError};
 use crate::expr::{Expr, ExprImpl, ExprRewriter};
 use crate::handler::create_source::{
     bind_connector_props, bind_create_source_or_table_with_connector, bind_source_watermark,
@@ -391,6 +391,17 @@ pub fn bind_sql_column_constraints(
                     }
                 }
                 ColumnOption::DefaultValuePersisted { persisted, expr: _ } => {
+                    // When a `DEFAULT INTERNAL` is used internally for schema change, the persisted value
+                    // should already be set during purifcation. So if we encounter an empty value here, it
+                    // means the user has specified it explicitly in the SQL statement, typically by
+                    // directly copying the result of `SHOW CREATE TABLE` and executing it.
+                    if persisted.is_empty() {
+                        bail_bind_error!(
+                            "DEFAULT INTERNAL is only used for internal purposes, \
+                             please specify a concrete default value"
+                        );
+                    }
+
                     let desc = DefaultColumnDesc::decode(&*persisted)
                         .expect("failed to decode persisted `DefaultColumnDesc`");
 
diff --git a/src/sqlparser/src/ast/ddl.rs b/src/sqlparser/src/ast/ddl.rs
index 84f643d7898d2..6c7e8689eb6f7 100644
--- a/src/sqlparser/src/ast/ddl.rs
+++ b/src/sqlparser/src/ast/ddl.rs
@@ -782,7 +782,7 @@ impl fmt::Display for ColumnOption {
                 if let Some(expr) = expr {
                     write!(f, "DEFAULT {}", expr)
                 } else {
-                    write!(f, "DEFAULT ...")
+                    write!(f, "DEFAULT INTERNAL")
                 }
             }
             Unique { is_primary } => {
diff --git a/src/sqlparser/src/parser.rs b/src/sqlparser/src/parser.rs
index 38e5514ecc585..27a627375204c 100644
--- a/src/sqlparser/src/parser.rs
+++ b/src/sqlparser/src/parser.rs
@@ -2798,7 +2798,15 @@ impl Parser<'_> {
         } else if self.parse_keyword(Keyword::NULL) {
             Ok(Some(ColumnOption::Null))
         } else if self.parse_keyword(Keyword::DEFAULT) {
-            Ok(Some(ColumnOption::DefaultValue(self.parse_expr()?)))
+            if self.parse_keyword(Keyword::INTERNAL) {
+                Ok(Some(ColumnOption::DefaultValuePersisted {
+                    // Placeholder. Will fill during definition purification for schema change.
+                    persisted: Default::default(),
+                    expr: None,
+                }))
+            } else {
+                Ok(Some(ColumnOption::DefaultValue(self.parse_expr()?)))
+            }
         } else if self.parse_keywords(&[Keyword::PRIMARY, Keyword::KEY]) {
             Ok(Some(ColumnOption::Unique { is_primary: true }))
         } else if self.parse_keyword(Keyword::UNIQUE) {