[server] Forbid any table/partition creation that begins with __

Author: caozhen1937

fluss-common/src/main/java/org/apache/fluss/metadata/TablePath.java

@@ -54,6 +54,8 @@ public class TablePath implements Serializable {
     // RecordAccumulator.ready)
     private Integer hash;
 
+    private static final String INTERNAL_NAME_PREFIX = "__";
+
     public TablePath(String databaseName, String tableName) {
         this.databaseName = databaseName;
         this.tableName = tableName;
@@ -127,20 +129,35 @@ public String toString() {
 
     public static void validateDatabaseName(String databaseName) throws InvalidDatabaseException {
         String dbError = detectInvalidName(databaseName);
-        if (dbError != null) {
+        String dbInternalNameError = validatePrefix(databaseName);
+        if (dbError != null || dbInternalNameError != null) {
             throw new InvalidDatabaseException(
-                    "Database name " + databaseName + " is invalid: " + dbError);
+                    "Database name "
+                            + databaseName
+                            + " is invalid: "
+                            + (dbError != null ? dbError : dbInternalNameError));
         }
     }
 
     public static void validateTableName(String tableName) throws InvalidTableException {
         String tableError = detectInvalidName(tableName);
-        if (tableError != null) {
+        String tableInternalNameError = validatePrefix(tableName);
+        if (tableError != null || tableInternalNameError != null) {
             throw new InvalidTableException(
-                    "Table name " + tableName + " is invalid: " + tableError);
+                    "Table name "
+                            + tableName
+                            + " is invalid: "
+                            + (tableError != null ? tableError : tableInternalNameError));
         }
     }
 
+    public static String validatePrefix(String identifier) throws InvalidTableException {
+        if (identifier != null && identifier.startsWith(INTERNAL_NAME_PREFIX)) {
+            return "'" + INTERNAL_NAME_PREFIX + "' is not allowed as prefix";
+        }
+        return null;
+    }
+
     public static String detectInvalidName(String identifier) {
         if (identifier == null) {
             return "null string is not allowed";

fluss-common/src/main/java/org/apache/fluss/utils/PartitionUtils.java

@@ -37,6 +37,7 @@
 import java.util.Map;
 
 import static org.apache.fluss.metadata.TablePath.detectInvalidName;
+import static org.apache.fluss.metadata.TablePath.validatePrefix;
 
 /** Utils for partition. */
 public class PartitionUtils {
@@ -66,7 +67,10 @@ public class PartitionUtils {
     private static final String HOUR_FORMAT = "yyyyMMddHH";
 
     public static void validatePartitionSpec(
-            TablePath tablePath, List<String> partitionKeys, PartitionSpec partitionSpec) {
+            TablePath tablePath,
+            List<String> partitionKeys,
+            PartitionSpec partitionSpec,
+            boolean isCreate) {
         Map<String, String> partitionSpecMap = partitionSpec.getSpecMap();
         if (partitionKeys.size() != partitionSpecMap.size()) {
             throw new InvalidPartitionException(
@@ -87,16 +91,21 @@ public static void validatePartitionSpec(
             }
         }
 
-        validatePartitionValues(reOrderedPartitionValues);
+        validatePartitionValues(reOrderedPartitionValues, isCreate);
     }
 
     @VisibleForTesting
-    static void validatePartitionValues(List<String> partitionValues) {
+    static void validatePartitionValues(List<String> partitionValues, boolean isCreate) {
         for (String value : partitionValues) {
-            String invalidName = detectInvalidName(value);
-            if (invalidName != null) {
+            String invalidNameError = detectInvalidName(value);
+            if (invalidNameError != null || (isCreate && validatePrefix(value) != null)) {
                 throw new InvalidPartitionException(
-                        "The partition value " + value + " is invalid: " + invalidName);
+                        "The partition value "
+                                + value
+                                + " is invalid: "
+                                + (invalidNameError != null
+                                        ? invalidNameError
+                                        : validatePrefix(value)));
             }
         }
     }

fluss-common/src/test/java/org/apache/fluss/metadata/TablePathTest.java

@@ -37,6 +37,13 @@ void testValidate() {
         assertThat(path.isValid()).isTrue();
         assertThat(path.toString()).isEqualTo("db_2-abc3.table-1_abc_2");
 
+        // assert invalid name prefix
+        TablePath invalidPath = TablePath.of("db_2", "__table-1");
+        assertThatThrownBy(invalidPath::validate)
+                .isInstanceOf(InvalidTableException.class)
+                .hasMessageContaining(
+                        "Table name __table-1 is invalid: '__' is not allowed as prefix");
+
         // check max length
         String longName = StringUtils.repeat("a", 200);
         assertThat(TablePath.of(longName, longName).isValid()).isTrue();

fluss-common/src/test/java/org/apache/fluss/utils/PartitionUtilsTest.java

@@ -52,18 +52,23 @@ class PartitionUtilsTest {
 
     @Test
     void testValidatePartitionValues() {
-        assertThatThrownBy(() -> validatePartitionValues(Arrays.asList("$1", "2")))
+        assertThatThrownBy(() -> validatePartitionValues(Arrays.asList("$1", "2"), true))
                 .isInstanceOf(InvalidPartitionException.class)
                 .hasMessageContaining(
                         "The partition value $1 is invalid: '$1' contains one "
                                 + "or more characters other than ASCII alphanumerics, '_' and '-'");
 
-        assertThatThrownBy(() -> validatePartitionValues(Arrays.asList("?1", "2")))
+        assertThatThrownBy(() -> validatePartitionValues(Arrays.asList("?1", "2"), true))
                 .isInstanceOf(InvalidPartitionException.class)
                 .hasMessageContaining(
                         "The partition value ?1 is invalid: '?1' contains one or more "
                                 + "characters other than ASCII alphanumerics, '_' and '-'");
 
+        assertThatThrownBy(() -> validatePartitionValues(Arrays.asList("__p1", "2"), true))
+                .isInstanceOf(InvalidPartitionException.class)
+                .hasMessageContaining(
+                        "The partition value __p1 is invalid: '__' is not allowed as prefix");
+
         TableDescriptor descriptor =
                 TableDescriptor.builder()
                         .schema(DATA1_SCHEMA)
@@ -80,7 +85,8 @@ void testValidatePartitionValues() {
                                 validatePartitionSpec(
                                         tableInfo.getTablePath(),
                                         tableInfo.getPartitionKeys(),
-                                        new PartitionSpec(Collections.emptyMap())))
+                                        new PartitionSpec(Collections.emptyMap()),
+                                        true))
                 .isInstanceOf(InvalidPartitionException.class)
                 .hasMessageContaining(
                         "PartitionSpec size is not equal to partition keys size for "

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java

@@ -376,7 +376,7 @@ public CompletableFuture<CreatePartitionResponse> createPartition(
 
         // first, validate the partition spec, and get resolved partition spec.
         PartitionSpec partitionSpec = getPartitionSpec(request.getPartitionSpec());
-        validatePartitionSpec(tablePath, table.partitionKeys, partitionSpec);
+        validatePartitionSpec(tablePath, table.partitionKeys, partitionSpec, true);
         ResolvedPartitionSpec partitionToCreate =
                 ResolvedPartitionSpec.fromPartitionSpec(table.partitionKeys, partitionSpec);
 
@@ -417,7 +417,7 @@ public CompletableFuture<DropPartitionResponse> dropPartition(DropPartitionReque
 
         // first, validate the partition spec.
         PartitionSpec partitionSpec = getPartitionSpec(request.getPartitionSpec());
-        validatePartitionSpec(tablePath, table.partitionKeys, partitionSpec);
+        validatePartitionSpec(tablePath, table.partitionKeys, partitionSpec, false);
         ResolvedPartitionSpec partitionToDrop =
                 ResolvedPartitionSpec.fromPartitionSpec(table.partitionKeys, partitionSpec);