diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java
index ed3814ab7824..f3a396c48189 100644
--- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java
+++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java
@@ -114,9 +114,9 @@ public void testChainTable(@TempDir java.nio.file.Path tempDir) throws IOExcepti
                         + "    `t1` BIGINT COMMENT 't1',\n"
                         + "    `t2` BIGINT COMMENT 't2',\n"
                         + "    `t3` STRING COMMENT 't3'\n"
-                        + "  ) PARTITIONED BY (`dt` STRING COMMENT 'dt') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
-                        + "WITH\n"
-                        + "  SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+                        + "  ) USING paimon\n"
+                        + "PARTITIONED BY (`dt` STRING COMMENT 'dt')\n"
+                        + "TBLPROPERTIES (\n"
                         + "    'bucket-key' = 't1',\n"
                         + "    'primary-key' = 'dt,t1',\n"
                         + "    'partition.timestamp-pattern' = '$dt',\n"
@@ -349,9 +349,9 @@ public void testHourlyChainTable(@TempDir java.nio.file.Path tempDir) throws IOE
                         + "    `t1` BIGINT COMMENT 't1',\n"
                         + "    `t2` BIGINT COMMENT 't2',\n"
                         + "    `t3` STRING COMMENT 't3'\n"
-                        + "  ) PARTITIONED BY (`dt` STRING COMMENT 'dt', `hour` STRING COMMENT 'hour') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
-                        + "WITH\n"
-                        + "  SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+                        + "  ) USING paimon\n"
+                        + "PARTITIONED BY (`dt` STRING COMMENT 'dt', `hour` STRING COMMENT 'hour')\n"
+                        + "TBLPROPERTIES (\n"
                         + "    'bucket-key' = 't1',\n"
                         + "    'primary-key' = 'dt,hour,t1',\n"
                         + "    'partition.timestamp-pattern' = '$dt $hour:00:00',\n"
@@ -592,9 +592,9 @@ public void testChainTableWithPartialUpdate(@TempDir java.nio.file.Path tempDir)
                         + "    `seq` BIGINT COMMENT 'seq',\n"
                         + "    `v1` STRING COMMENT 'v1',\n"
                         + "    `v2` STRING COMMENT 'v2'\n"
-                        + "  ) PARTITIONED BY (`dt` STRING COMMENT 'dt') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
-                        + "WITH\n"
-                        + "  SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+                        + "  ) USING paimon\n"
+                        + "PARTITIONED BY (`dt` STRING COMMENT 'dt')\n"
+                        + "TBLPROPERTIES (\n"
                         + "    'bucket-key' = 'key',\n"
                         + "    'primary-key' = 'dt,key',\n"
                         + "    'partition.timestamp-pattern' = '$dt',\n"
@@ -748,9 +748,9 @@ public void testDropSnapshotPartition(@TempDir java.nio.file.Path tempDir) throw
                         + "    `t1` BIGINT COMMENT 't1',\n"
                         + "    `t2` BIGINT COMMENT 't2',\n"
                         + "    `t3` STRING COMMENT 't3'\n"
-                        + "  ) PARTITIONED BY (`dt` STRING COMMENT 'dt') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
-                        + "WITH\n"
-                        + "  SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+                        + "  ) USING paimon\n"
+                        + "PARTITIONED BY (`dt` STRING COMMENT 'dt')\n"
+                        + "TBLPROPERTIES (\n"
                         + "    'bucket-key' = 't1',\n"
                         + "    'primary-key' = 'dt,t1',\n"
                         + "    'partition.timestamp-pattern' = '$dt',\n"
@@ -822,7 +822,7 @@ public void testChainTableCacheInvalidation(@TempDir java.nio.file.Path tempDir)
                         + "    `t1` string ,"
                         + "    `t2` string ,"
                         + "    `t3` string"
-                        + ") PARTITIONED BY (`date` string)"
+                        + ") USING paimon PARTITIONED BY (`date` string) "
                         + "TBLPROPERTIES ("
                         + "   'chain-table.enabled' = 'true'"
                         + "  ,'primary-key' = 'date,t1'"
@@ -878,9 +878,9 @@ public void testChainTableWithGroupPartition(@TempDir java.nio.file.Path tempDir
                         + "    `t1` BIGINT COMMENT 't1',\n"
                         + "    `t2` BIGINT COMMENT 't2',\n"
                         + "    `t3` STRING COMMENT 't3'\n"
-                        + "  ) PARTITIONED BY (`region` STRING, `dt` STRING COMMENT 'dt') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
-                        + "WITH\n"
-                        + "  SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+                        + "  ) USING paimon\n"
+                        + "PARTITIONED BY (`region` STRING, `dt` STRING COMMENT 'dt')\n"
+                        + "TBLPROPERTIES (\n"
                         + "    'bucket-key' = 't1',\n"
                         + "    'primary-key' = 'region,dt,t1',\n"
                         + "    'partition.timestamp-pattern' = '$dt',\n"
@@ -1310,9 +1310,9 @@ public void testHourlyChainTableWithGroupPartition(@TempDir java.nio.file.Path t
                         + "    `t1` BIGINT COMMENT 't1',\n"
                         + "    `t2` BIGINT COMMENT 't2',\n"
                         + "    `t3` STRING COMMENT 't3'\n"
-                        + "  ) PARTITIONED BY (`region` STRING, `dt` STRING COMMENT 'dt', `hour` STRING COMMENT 'hour') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
-                        + "WITH\n"
-                        + "  SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+                        + "  ) USING paimon\n"
+                        + "PARTITIONED BY (`region` STRING, `dt` STRING COMMENT 'dt', `hour` STRING COMMENT 'hour')\n"
+                        + "TBLPROPERTIES (\n"
                         + "    'bucket-key' = 't1',\n"
                         + "    'primary-key' = 'region,dt,hour,t1',\n"
                         + "    'partition.timestamp-pattern' = '$dt $hour:00:00',\n"
@@ -1767,9 +1767,9 @@ public void testChainTableWithMultiGroupPartition(@TempDir java.nio.file.Path te
                         + "    `t1` BIGINT COMMENT 't1',\n"
                         + "    `t2` BIGINT COMMENT 't2',\n"
                         + "    `t3` STRING COMMENT 't3'\n"
-                        + "  ) PARTITIONED BY (`region` STRING, `biz_type` STRING COMMENT 'biz_type', `dt` STRING COMMENT 'dt') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
-                        + "WITH\n"
-                        + "  SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+                        + "  ) USING paimon\n"
+                        + "PARTITIONED BY (`region` STRING, `biz_type` STRING COMMENT 'biz_type', `dt` STRING COMMENT 'dt')\n"
+                        + "TBLPROPERTIES (\n"
                         + "    'bucket-key' = 't1',\n"
                         + "    'primary-key' = 'region,biz_type,dt,t1',\n"
                         + "    'partition.timestamp-pattern' = '$dt',\n"
@@ -2314,7 +2314,7 @@ public void testChainTableWithBranchOption(@TempDir java.nio.file.Path tempDir)
                         + "  `t1` BIGINT,\n"
                         + "  `t2` BIGINT,\n"
                         + "  `t3` STRING\n"
-                        + ") PARTITIONED BY (`dt` STRING)\n"
+                        + ") USING paimon PARTITIONED BY (`dt` STRING)\n"
                         + "TBLPROPERTIES (\n"
                         + "  'bucket-key' = 't1',\n"
                         + "  'primary-key' = 'dt,t1',\n"
@@ -2378,7 +2378,7 @@ public void testChainTableWithMultiChainKeys(@TempDir java.nio.file.Path tempDir
                         + "  `t1` BIGINT COMMENT 't1',\n"
                         + "  `t2` BIGINT COMMENT 't2',\n"
                         + "  `t3` STRING COMMENT 't3'\n"
-                        + ") PARTITIONED BY (`dt` STRING, `hr` STRING)\n"
+                        + ") USING paimon PARTITIONED BY (`dt` STRING, `hr` STRING)\n"
                         + "TBLPROPERTIES (\n"
                         + "  'bucket-key' = 't1',\n"
                         + "  'primary-key' = 'dt,hr,t1',\n"
@@ -2436,7 +2436,8 @@ public void testChainTableWithDeletionVectors(@TempDir java.nio.file.Path tempDi
                         + "  `t1` BIGINT COMMENT 't1',\n"
                         + "  `t2` BIGINT COMMENT 't2',\n"
                         + "  `t3` STRING COMMENT 't3'\n"
-                        + ") PARTITIONED BY (`region` STRING COMMENT 'region', `date` STRING COMMENT 'date')\n"
+                        + ") USING paimon\n"
+                        + "PARTITIONED BY (`region` STRING COMMENT 'region', `date` STRING COMMENT 'date')\n"
                         + "TBLPROPERTIES (\n"
                         + "  'chain-table.enabled' = 'true',\n"
                         + "  'deletion-vectors.enabled' = 'true',\n"
diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkDataEvolutionITCase.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkDataEvolutionITCase.java
index d9a8c6c71164..f6b48d9e69ff 100644
--- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkDataEvolutionITCase.java
+++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkDataEvolutionITCase.java
@@ -105,9 +105,9 @@ public void testDataEvolution(@TempDir java.nio.file.Path tempDir) throws IOExce
                         + "    `g_1_2` BIGINT COMMENT 'g_1_2',\n"
                         + "    `g_2_1` BIGINT COMMENT 'g_2_1',\n"
                         + "    `g_2_2` BIGINT COMMENT 'g_2_2'\n"
-                        + "  ) PARTITIONED BY (`dt` STRING COMMENT 'dt') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
-                        + "WITH\n"
-                        + "  SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+                        + "  ) USING paimon\n"
+                        + "PARTITIONED BY (`dt` STRING COMMENT 'dt')\n"
+                        + "TBLPROPERTIES (\n"
                         + "    'file.compression' = 'snappy',\n"
                         + "    'manifest.compression' = 'snappy',\n"
                         + "    'row-tracking.enabled' = 'true',\n"
@@ -124,9 +124,9 @@ public void testDataEvolution(@TempDir java.nio.file.Path tempDir) throws IOExce
                         + "    `id` BIGINT COMMENT 'id',\n"
                         + "    `g_2_1` BIGINT COMMENT 'g_2_1',\n"
                         + "    `g_2_2` BIGINT COMMENT 'g_2_2'\n"
-                        + "  ) PARTITIONED BY (`dt` STRING COMMENT 'dt') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
-                        + "WITH\n"
-                        + "  SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+                        + "  ) USING paimon\n"
+                        + "PARTITIONED BY (`dt` STRING COMMENT 'dt')\n"
+                        + "TBLPROPERTIES (\n"
                         + "    'file.compression' = 'snappy',\n"
                         + "    'manifest.compression' = 'snappy',\n"
                         + "    'partition.timestamp-pattern' = '$dt',\n"
diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkMultimodalITCase.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkMultimodalITCase.java
index 6166d257fc12..c75662d1d12a 100644
--- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkMultimodalITCase.java
+++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkMultimodalITCase.java
@@ -85,9 +85,9 @@ public void testVector(@TempDir java.nio.file.Path tempDir) throws IOException {
         spark.sql(
                 "\n"
                         + "CREATE TABLE my_db1.vector_test (gid BIGINT, sid STRING, embs ARRAY<FLOAT>)"
-                        + " PARTITIONED BY (`date` STRING COMMENT 'date') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
-                        + "WITH\n"
-                        + "  SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+                        + " USING paimon\n"
+                        + "PARTITIONED BY (`date` STRING COMMENT 'date')\n"
+                        + "TBLPROPERTIES (\n"
                         + "    'vector.file.format'='lance',\n"
                         + "    'vector-field'='embs',\n"
                         + "    'field.embs.vector-dim'='4',\n"