[core] Should prevent users from specifying partition columns as blob field (#6753)

Author: steFaiz

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

@@ -78,7 +78,6 @@
 import static org.apache.paimon.table.SpecialFields.KEY_FIELD_PREFIX;
 import static org.apache.paimon.table.SpecialFields.SYSTEM_FIELD_NAMES;
 import static org.apache.paimon.types.DataTypeRoot.ARRAY;
-import static org.apache.paimon.types.DataTypeRoot.BLOB;
 import static org.apache.paimon.types.DataTypeRoot.MAP;
 import static org.apache.paimon.types.DataTypeRoot.MULTISET;
 import static org.apache.paimon.types.DataTypeRoot.ROW;
@@ -650,16 +649,21 @@ private static void validateRowTracking(TableSchema schema, CoreOptions options)
                     "Data evolution config must disabled with deletion-vectors.enabled");
         }
 
-        if (schema.fields().stream().map(DataField::type).anyMatch(t -> t.is(BLOB))) {
+        List<String> blobNames =
+                BlobType.splitBlob(schema.logicalRowType()).getRight().getFieldNames();
+        if (!blobNames.isEmpty()) {
             checkArgument(
                     options.dataEvolutionEnabled(),
                     "Data evolution config must enabled for table with BLOB type column.");
             checkArgument(
-                    BlobType.splitBlob(schema.logicalRowType()).getRight().getFieldCount() == 1,
+                    blobNames.size() == 1,
                     "Table with BLOB type column only support one BLOB column.");
             checkArgument(
                     schema.fields().size() > 1,
                     "Table with BLOB type column must have other normal columns.");
+            checkArgument(
+                    !schema.partitionKeys().contains(blobNames.get(0)),
+                    "The BLOB type column can not be part of partition keys.");
         }
     }
 

paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java

@@ -56,6 +56,18 @@ private void validateTableSchemaExec(Map<String, String> options) {
                 new TableSchema(1, fields, 10, partitionKeys, primaryKeys, options, ""));
     }
 
+    private void validateBlobSchema(Map<String, String> options, List<String> partitions) {
+        List<DataField> fields =
+                Arrays.asList(
+                        new DataField(0, "f0", DataTypes.INT()),
+                        new DataField(1, "f1", DataTypes.INT()),
+                        new DataField(2, "f2", DataTypes.BLOB()),
+                        new DataField(3, "f3", DataTypes.STRING()));
+        options.put(BUCKET.key(), String.valueOf(-1));
+        validateTableSchema(
+                new TableSchema(1, fields, 10, partitions, Collections.emptyList(), options, ""));
+    }
+
     @Test
     public void testOnlyTimestampMillis() {
         Map<String, String> options = new HashMap<>();
@@ -122,4 +134,21 @@ public void testRecordLevelTimeField() {
                 .hasMessageContaining(
                         "The record level time field type should be one of INT, BIGINT, or TIMESTAMP, but field type is STRING.");
     }
+
+    @Test
+    public void testBlobTableSchema() {
+        Map<String, String> options = new HashMap<>();
+
+        // 1. must set row-tracking = true and data-evolution = true
+        options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+        assertThatThrownBy(() -> validateBlobSchema(options, Collections.emptyList()))
+                .hasMessage("Data evolution config must enabled for table with BLOB type column.");
+
+        // 2. blob column cannot be part of partition keys
+        options.clear();
+        options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+        options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+        assertThatThrownBy(() -> validateBlobSchema(options, Collections.singletonList("f2")))
+                .hasMessage("The BLOB type column can not be part of partition keys.");
+    }
 }