add create or replace view

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonViewExec.scala

@@ -48,13 +48,24 @@ case class CreatePaimonViewExec(
   override def output: Seq[Attribute] = Nil
 
   override protected def run(): Seq[InternalRow] = {
+    if (columnAliases.nonEmpty || columnComments.nonEmpty || queryColumnNames.nonEmpty) {
+      throw new UnsupportedOperationException(
+        "columnAliases, columnComments and queryColumnNames are not supported now")
+    }
+
+    // Note: for replace just drop then create ,this operation is non-atomic.
+    if (replace) {
+      catalog.dropView(ident, true)
+    }
+
     catalog.createView(
       ident,
       viewSchema,
       queryText,
       comment.orNull,
       properties.asJava,
       allowExisting)
+
     Nil
   }
 

paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonViewTestBase.scala

@@ -44,11 +44,19 @@ abstract class PaimonViewTestBase extends PaimonHiveTestBase {
                   sql("SELECT * FROM v1 WHERE id >= (SELECT max(id) FROM v1)"),
                   Seq(Row(2)))
 
+                // test drop view
                 sql("DROP VIEW IF EXISTS v1")
                 checkAnswer(sql("SHOW VIEWS"), Seq())
                 sql("CREATE VIEW v1 AS SELECT * FROM t WHERE id > 1")
                 checkAnswer(sql("SHOW VIEWS"), Seq(Row("test_db", "v1", false)))
                 checkAnswer(sql("SELECT * FROM v1"), Seq(Row(2)))
+
+                // test create or replace view
+                intercept[Exception] {
+                  sql("CREATE VIEW v1 AS SELECT * FROM t WHERE id < 2")
+                }
+                sql("CREATE OR REPLACE VIEW v1 AS SELECT * FROM t WHERE id < 2")
+                checkAnswer(sql("SELECT * FROM v1"), Seq(Row(1)))
               }
             }
           }