[server] Forbid any table/partition creation that begins with __ (#1706)

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

* addressed yunhong's comments

* cover test in FlussAdminITCase

Author: caozhen1937

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java

@@ -281,7 +281,7 @@ void testAlterTable() throws Exception {
     }
 
     @Test
-    void testCreateInvalidDatabaseAndTable() {
+    void testCreateInvalidDatabaseAndTable() throws Exception {
         assertThatThrownBy(
                         () ->
                                 admin.createDatabase(
@@ -290,6 +290,15 @@ void testCreateInvalidDatabaseAndTable() {
                 .isInstanceOf(InvalidDatabaseException.class)
                 .hasMessageContaining(
                         "Database name *invalid_db* is invalid: '*invalid_db*' contains one or more characters other than");
+        //  test internal database with '__' prefix is not allowed
+        assertThatThrownBy(
+                        () ->
+                                admin.createDatabase(
+                                                "__internal_db", DatabaseDescriptor.EMPTY, false)
+                                        .get())
+                .isInstanceOf(InvalidDatabaseException.class)
+                .hasMessageContaining(
+                        "Database name __internal_db is invalid: '__' is not allowed as prefix, since it is reserved for internal databases/internal tables/internal partitions in Fluss server");
         assertThatThrownBy(
                         () ->
                                 admin.createTable(
@@ -309,6 +318,17 @@ void testCreateInvalidDatabaseAndTable() {
                                         .get())
                 .isInstanceOf(InvalidDatabaseException.class)
                 .hasMessageContaining("Database name null is invalid: null string is not allowed");
+        //  test internal table with '__' prefix is not allowed
+        assertThatThrownBy(
+                        () ->
+                                admin.createTable(
+                                                TablePath.of("db", "__internal_table"),
+                                                DEFAULT_TABLE_DESCRIPTOR,
+                                                false)
+                                        .get())
+                .isInstanceOf(InvalidTableException.class)
+                .hasMessageContaining(
+                        "Table name __internal_table is invalid: '__' is not allowed as prefix, since it is reserved for internal databases/internal tables/internal partitions in Fluss server");
     }
 
     @Test
@@ -748,6 +768,21 @@ void testAddAndDropPartitions() throws Exception {
         admin.createTable(tablePath, partitionedTable, true).get();
         assertPartitionInfo(admin.listPartitionInfos(tablePath).get(), Collections.emptyList());
 
+        // test internal partition with '__' prefix is not allowed
+        assertThatThrownBy(
+                        () ->
+                                admin.createPartition(
+                                                tablePath,
+                                                newPartitionSpec(
+                                                        Arrays.asList("age"),
+                                                        Arrays.asList("__18")),
+                                                false)
+                                        .get())
+                .cause()
+                .isInstanceOf(InvalidPartitionException.class)
+                .hasMessageContaining(
+                        "The partition value __18 is invalid: '__' is not allowed as prefix, since it is reserved for internal databases/internal tables/internal partitions in Fluss server");
+
         // add two partitions.
         admin.createPartition(tablePath, newPartitionSpec("age", "10"), false).get();
         admin.createPartition(tablePath, newPartitionSpec("age", "11"), false).get();

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

@@ -54,6 +54,12 @@ public class TablePath implements Serializable {
     // RecordAccumulator.ready)
     private Integer hash;
 
+    // Prefix reserved for internal system names. User-defined database and table and others names
+    // are not
+    // allowed to start with this prefix to prevent conflicts with system-generated identifiers.
+    // This convention aligns with and maintains compatibility with Apache Kafka's naming standards.
+    private static final String INTERNAL_NAME_PREFIX = "__";
+
     public TablePath(String databaseName, String tableName) {
         this.databaseName = databaseName;
         this.tableName = tableName;
@@ -127,18 +133,36 @@ 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, since it is reserved"
+                    + " for internal databases/internal tables/internal partitions in Fluss server";
         }
+        return null;
     }
 
     public static String detectInvalidName(String identifier) {

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

@@ -36,6 +36,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 {
@@ -65,7 +66,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(
@@ -86,16 +90,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

@@ -45,25 +45,35 @@
 import static org.apache.fluss.utils.PartitionUtils.validatePartitionSpec;
 import static org.apache.fluss.utils.PartitionUtils.validatePartitionValues;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link PartitionUtils}. */
 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"), false))
                 .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, "
+                                + "since it is reserved for internal databases/internal tables/internal partitions in Fluss server");
+
+        assertThatNoException()
+                .isThrownBy(() -> validatePartitionValues(Arrays.asList("__p1", "2"), false));
+
         TableDescriptor descriptor =
                 TableDescriptor.builder()
                         .schema(DATA1_SCHEMA)
@@ -80,7 +90,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

@@ -448,7 +448,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);
 
@@ -489,7 +489,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);