[core] Add validation to upsert key

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java

@@ -97,10 +97,18 @@ public class SchemaValidation {
      * @param schema the schema to be validated
      */
     public static void validateTableSchema(TableSchema schema) {
+        CoreOptions options = new CoreOptions(schema.options());
+
         validateOnlyContainPrimitiveType(schema.fields(), schema.primaryKeys(), "primary key");
         validateOnlyContainPrimitiveType(schema.fields(), schema.partitionKeys(), "partition");
+        validateOnlyContainPrimitiveType(schema.fields(), options.upsertKey(), "upsert key");
 
-        CoreOptions options = new CoreOptions(schema.options());
+        if (!options.upsertKey().isEmpty() && !schema.primaryKeys().isEmpty()) {
+            throw new RuntimeException(
+                    String.format(
+                            "Cannot define 'upsert-key' %s with 'primary-key' %s.",
+                            options.upsertKey(), schema.primaryKeys()));
+        }
 
         validateBucket(schema, options);
 
@@ -251,6 +259,9 @@ private static void validateOnlyContainPrimitiveType(
             }
             for (String fieldName : fieldNames) {
                 DataField rowField = rowFields.get(fieldName);
+                if (rowField == null) {
+                    throw new IllegalArgumentException("Cannot find field: " + fieldName);
+                }
                 DataType dataType = rowField.type();
                 if (PRIMARY_KEY_UNSUPPORTED_LOGICAL_TYPES.stream()
                         .anyMatch(c -> c.isInstance(dataType))) {

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTestBase.scala

@@ -24,6 +24,15 @@ import org.apache.spark.sql.Row
 
 abstract class RewriteUpsertTableTestBase extends PaimonSparkTestBase {
 
+  test("Rewrite Upsert Table: cannot define with primary key") {
+    assert(intercept[Exception] {
+      sql("""
+            |CREATE TABLE T (k INT, a INT, b STRING)
+            |TBLPROPERTIES ('upsert-key' = 'k', 'primary-key' = 'k')
+            |""".stripMargin)
+    }.getMessage.contains("Cannot define 'upsert-key' [k] with 'primary-key' [k]."))
+  }
+
   test("Rewrite Upsert Table: rewrite insert without sequence field") {
     sql("""
           |CREATE TABLE T (k INT, a INT, b STRING)