[core] Add option for catalog options system table

docs/docs/concepts/system-tables.mdx

@@ -609,7 +609,8 @@ SELECT * FROM sys.all_table_options;
 
 ### Catalog Options Table
 
-You can query the catalog's option information through catalog options table. The options not shown will be the default value. You can take reference to [Configuration](../maintenance/configurations#coreoptions).
+You can query the catalog's option information through catalog options table when `catalog-options-table.enabled` is set to `true`.
+The options not shown will be the default value. You can take reference to [Configuration](../maintenance/configurations#coreoptions).
 
 ```sql
 SELECT * FROM sys.catalog_options;

docs/generated/catalog_configuration.html

@@ -92,6 +92,12 @@
             <td>Boolean</td>
             <td>Indicates whether this catalog is case-sensitive.</td>
         </tr>
+        <tr>
+            <td><h5>catalog-options-table.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to support the sys.catalog_options table.</td>
+        </tr>
         <tr>
             <td><h5>client-pool-size</h5></td>
             <td style="word-wrap: break-word;">2</td>

paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java

@@ -183,6 +183,12 @@ public class CatalogOptions {
                                     + "However, during these processes, it does not connect to the metastore; hence, newly added partitions will not be reflected in"
                                     + " the metastore and need to be manually added as separate partition operations.");
 
+    public static final ConfigOption<Boolean> CATALOG_OPTIONS_TABLE_ENABLED =
+            ConfigOptions.key("catalog-options-table.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Whether to support the sys.catalog_options table.");
+
     public static final ConfigOption<Boolean> RESOLVING_FILE_IO_ENABLED =
             ConfigOptions.key("resolving-file-io.enabled")
                     .booleanType()

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -255,7 +255,7 @@ protected abstract void alterDatabaseImpl(String name, List<PropertyChange> chan
     @Override
     public List<String> listTables(String databaseName) throws DatabaseNotExistException {
         if (isSystemDatabase(databaseName)) {
-            return SystemTableLoader.loadGlobalTableNames();
+            return SystemTableLoader.loadGlobalTableNames(context.options());
         }
 
         // check db exists
@@ -289,7 +289,7 @@ public PagedList<Table> listTableDetailsPaged(
         CatalogUtils.validateTableType(this, tableType);
         if (isSystemDatabase(databaseName)) {
             List<Table> systemTables =
-                    SystemTableLoader.loadGlobalTableNames().stream()
+                    SystemTableLoader.loadGlobalTableNames(context.options()).stream()
                             .map(
                                     tableName -> {
                                         try {

paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java

@@ -23,6 +23,7 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.Partition;
 import org.apache.paimon.rest.exceptions.NotImplementedException;
@@ -328,7 +329,12 @@ private static Table createGlobalSystemTable(String tableName, Catalog catalog)
                 return AllPartitionsTable.fromPartitions(
                         toAllPartitions(catalog, listAllTables(catalog)));
             case CATALOG_OPTIONS:
-                return new CatalogOptionsTable(Options.fromMap(catalog.options()));
+                Options catalogOptions = Options.fromMap(catalog.options());
+                if (!catalogOptions.get(CatalogOptions.CATALOG_OPTIONS_TABLE_ENABLED)) {
+                    throw new Catalog.TableNotExistException(
+                            Identifier.create(SYSTEM_DATABASE_NAME, tableName));
+                }
+                return new CatalogOptionsTable(catalogOptions);
             default:
                 throw new Catalog.TableNotExistException(
                         Identifier.create(SYSTEM_DATABASE_NAME, tableName));

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java

@@ -222,7 +222,7 @@ public void alterDatabase(String name, List<PropertyChange> changes, boolean ign
     public List<String> listTables(String databaseName) throws DatabaseNotExistException {
         try {
             if (isSystemDatabase(databaseName)) {
-                return SystemTableLoader.loadGlobalTableNames();
+                return SystemTableLoader.loadGlobalTableNames(context.options());
             }
             return api.listTables(databaseName);
         } catch (NoSuchResourceException e) {
@@ -241,6 +241,11 @@ public PagedList<String> listTablesPaged(
             @Nullable String tableType)
             throws DatabaseNotExistException {
         try {
+            if (isSystemDatabase(databaseName)) {
+                CatalogUtils.validateNamePattern(this, tableNamePattern);
+                CatalogUtils.validateTableType(this, tableType);
+                return listSystemTablesPaged(maxResults, pageToken, tableNamePattern, tableType);
+            }
             return api.listTablesPaged(
                     databaseName, maxResults, pageToken, tableNamePattern, tableType);
         } catch (NoSuchResourceException e) {
@@ -257,6 +262,25 @@ public PagedList<Table> listTableDetailsPaged(
             @Nullable String tableType)
             throws DatabaseNotExistException {
         try {
+            if (isSystemDatabase(db)) {
+                CatalogUtils.validateNamePattern(this, tableNamePattern);
+                CatalogUtils.validateTableType(this, tableType);
+                PagedList<String> pagedSystemTableNames =
+                        listSystemTablesPaged(maxResults, pageToken, tableNamePattern, tableType);
+                List<Table> systemTables =
+                        pagedSystemTableNames.getElements().stream()
+                                .map(
+                                        tableName -> {
+                                            try {
+                                                return getTable(Identifier.create(db, tableName));
+                                            } catch (TableNotExistException ignored) {
+                                                return null;
+                                            }
+                                        })
+                                .filter(Objects::nonNull)
+                                .collect(Collectors.toList());
+                return new PagedList<>(systemTables, pagedSystemTableNames.getNextPageToken());
+            }
             PagedList<GetTableResponse> tables =
                     api.listTableDetailsPaged(
                             db, maxResults, pageToken, tableNamePattern, tableType);
@@ -270,9 +294,33 @@ public PagedList<Table> listTableDetailsPaged(
         }
     }
 
+    private PagedList<String> listSystemTablesPaged(
+            @Nullable Integer maxResults,
+            @Nullable String pageToken,
+            @Nullable String tableNamePattern,
+            @Nullable String tableType) {
+        try {
+            return SystemTableLoader.loadGlobalTableNamesPaged(
+                    context.options(), maxResults, pageToken, tableNamePattern, tableType);
+        } catch (IllegalArgumentException e) {
+            throw new BadRequestException(e.getMessage());
+        }
+    }
+
     @Override
     public List<Table> listTableDetails(String databaseName) throws DatabaseNotExistException {
         try {
+            if (isSystemDatabase(databaseName)) {
+                List<Table> result = new ArrayList<>();
+                for (String tableName : SystemTableLoader.loadGlobalTableNames(context.options())) {
+                    try {
+                        result.add(getTable(Identifier.create(databaseName, tableName)));
+                    } catch (TableNotExistException ignored) {
+                        // ignore
+                    }
+                }
+                return result;
+            }
             List<GetTableResponse> tables = api.listTableDetails(databaseName);
             return tables.stream().map(t -> toTable(databaseName, t)).collect(Collectors.toList());
         } catch (NoSuchResourceException e) {

paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java

@@ -18,8 +18,13 @@
 
 package org.apache.paimon.table.system;
 
+import org.apache.paimon.PagedList;
+import org.apache.paimon.TableType;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
+import org.apache.paimon.utils.StringUtils;
 
 import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
 
@@ -31,6 +36,8 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Function;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import static org.apache.paimon.table.system.AggregationFieldsTable.AGGREGATION_FIELDS;
 import static org.apache.paimon.table.system.AllPartitionsTable.ALL_PARTITIONS;
@@ -92,7 +99,100 @@ public static Table load(String type, FileStoreTable dataTable) {
                 .orElse(null);
     }
 
+    public static List<String> loadGlobalTableNames(Options catalogOptions) {
+        List<String> tableNames = new ArrayList<>(GLOBAL_SYSTEM_TABLES);
+        if (!catalogOptions.get(CatalogOptions.CATALOG_OPTIONS_TABLE_ENABLED)) {
+            tableNames.remove(CATALOG_OPTIONS);
+        }
+        return tableNames;
+    }
+
+    public static PagedList<String> loadGlobalTableNamesPaged(
+            Options catalogOptions,
+            @Nullable Integer maxResults,
+            @Nullable String pageToken,
+            @Nullable String tableNamePattern,
+            @Nullable String tableType) {
+        validatePrefixSqlPattern(tableNamePattern);
+        List<String> tableNames =
+                loadGlobalTableNames(catalogOptions).stream()
+                        .filter(tableName -> matchesNamePattern(tableName, tableNamePattern))
+                        .filter(tableName -> matchesTableType(tableType))
+                        .sorted()
+                        .collect(Collectors.toList());
+
+        Integer pageSize = maxResults != null && maxResults > 0 ? maxResults : null;
+        List<String> pagedTableNames = new ArrayList<>();
+        for (String tableName : tableNames) {
+            if (pageToken != null && tableName.compareTo(pageToken) <= 0) {
+                continue;
+            }
+            if (pageSize != null && pagedTableNames.size() >= pageSize) {
+                break;
+            }
+            pagedTableNames.add(tableName);
+        }
+
+        String nextPageToken =
+                pageSize != null && pagedTableNames.size() == pageSize
+                        ? pagedTableNames.get(pagedTableNames.size() - 1)
+                        : null;
+        return new PagedList<>(pagedTableNames, nextPageToken);
+    }
+
     public static List<String> loadGlobalTableNames() {
-        return GLOBAL_SYSTEM_TABLES;
+        return loadGlobalTableNames(new Options());
+    }
+
+    private static boolean matchesNamePattern(String name, @Nullable String namePattern) {
+        if (StringUtils.isEmpty(namePattern)) {
+            return true;
+        }
+        return Pattern.compile(sqlPatternToRegex(namePattern)).matcher(name).matches();
+    }
+
+    private static boolean matchesTableType(@Nullable String tableType) {
+        return StringUtils.isEmpty(tableType) || TableType.TABLE.toString().equals(tableType);
+    }
+
+    private static void validatePrefixSqlPattern(@Nullable String pattern) {
+        if (StringUtils.isEmpty(pattern)) {
+            return;
+        }
+
+        boolean escaped = false;
+        boolean inWildcardZone = false;
+        for (int i = 0; i < pattern.length(); i++) {
+            char c = pattern.charAt(i);
+            if (escaped) {
+                escaped = false;
+            } else if (c == '\\') {
+                escaped = true;
+            } else if (c == '%') {
+                inWildcardZone = true;
+            } else if (inWildcardZone) {
+                throw new IllegalArgumentException(
+                        "Can only support prefix sql like pattern query now.");
+            }
+        }
+    }
+
+    private static String sqlPatternToRegex(String pattern) {
+        StringBuilder regex = new StringBuilder();
+        boolean escaped = false;
+        for (int i = 0; i < pattern.length(); i++) {
+            char c = pattern.charAt(i);
+            if (escaped) {
+                regex.append(c);
+                escaped = false;
+            } else if (c == '\\') {
+                escaped = true;
+            } else if (c == '%') {
+                regex.append(".*");
+            } else {
+                regex.append(c);
+            }
+        }
+        return "^" + regex + "$";
     }
 }

paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java

@@ -47,6 +47,7 @@
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.table.source.ReadBuilder;
 import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.table.system.SystemTableLoader;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowType;
@@ -85,7 +86,6 @@
 import static org.apache.paimon.catalog.Catalog.SYSTEM_DATABASE_NAME;
 import static org.apache.paimon.table.system.AllTableOptionsTable.ALL_TABLE_OPTIONS;
 import static org.apache.paimon.table.system.CatalogOptionsTable.CATALOG_OPTIONS;
-import static org.apache.paimon.table.system.SystemTableLoader.GLOBAL_SYSTEM_TABLES;
 import static org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatCode;
@@ -1037,15 +1037,19 @@ public void testGetTable() throws Exception {
         Table allTableOptionsTable =
                 catalog.getTable(Identifier.create(SYSTEM_DATABASE_NAME, ALL_TABLE_OPTIONS));
         assertThat(allTableOptionsTable).isNotNull();
-        Table catalogOptionsTable =
-                catalog.getTable(Identifier.create(SYSTEM_DATABASE_NAME, CATALOG_OPTIONS));
-        assertThat(catalogOptionsTable).isNotNull();
+        assertThatExceptionOfType(Catalog.TableNotExistException.class)
+                .isThrownBy(
+                        () ->
+                                catalog.getTable(
+                                        Identifier.create(SYSTEM_DATABASE_NAME, CATALOG_OPTIONS)));
         assertThatExceptionOfType(Catalog.TableNotExistException.class)
                 .isThrownBy(
                         () -> catalog.getTable(Identifier.create(SYSTEM_DATABASE_NAME, "1111")));
 
         List<String> sysTables = catalog.listTables(SYSTEM_DATABASE_NAME);
-        assertThat(sysTables).containsAll(GLOBAL_SYSTEM_TABLES);
+        assertThat(sysTables)
+                .containsExactlyInAnyOrderElementsOf(
+                        SystemTableLoader.loadGlobalTableNames(new Options()));
 
         assertThat(catalog.listViews(SYSTEM_DATABASE_NAME)).isEmpty();
     }