fhan688 · Oct 26, 2026
diff --git a/...-1.20/src/test/java/org/apache/fluss/flink/catalog/Flink120PartitionsPlanCacheITCase.java b/...-1.20/src/test/java/org/apache/fluss/flink/catalog/Flink120PartitionsPlanCacheITCase.java
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.catalog;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.PartitionSpec;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.gateway.api.endpoint.EndpointVersion;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.session.SessionEnvironment;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.service.SqlGatewayServiceImpl;
+import org.apache.flink.table.gateway.service.utils.SqlGatewayServiceExtension;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.flink.table.gateway.api.config.SqlGatewayServiceConfigOptions.SQL_GATEWAY_SESSION_PLAN_CACHE_ENABLED;
+import static org.apache.flink.table.gateway.service.utils.SqlGatewayServiceTestUtil.awaitOperationTermination;
+import static org.apache.flink.table.gateway.service.utils.SqlGatewayServiceTestUtil.fetchAllResults;
+import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests that cached SQL Gateway plans load fresh {@code $partitions} metadata at runtime. */
+public class Flink120PartitionsPlanCacheITCase {
+
+    private static final String CATALOG_NAME = "testcatalog";
+    private static final String DEFAULT_DB = FlinkCatalogOptions.DEFAULT_DATABASE.defaultValue();
+
+    @RegisterExtension
+    public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+            FlussClusterExtension.builder().setNumOfTabletServers(1).build();
+
+    @RegisterExtension
+    @Order(1)
+    static final MiniClusterExtension MINI_CLUSTER =
+            new MiniClusterExtension(
+                    new MiniClusterResourceConfiguration.Builder()
+                            .setNumberTaskManagers(1)
+                            .build());
+
+    @RegisterExtension
+    @Order(2)
+    static final SqlGatewayServiceExtension SQL_GATEWAY_SERVICE_EXTENSION =
+            new SqlGatewayServiceExtension(MINI_CLUSTER::getClientConfiguration);
+
+    private static SqlGatewayServiceImpl service;
+
+    private SessionHandle sessionHandle;
+
+    @BeforeAll
+    static void beforeAll() {
+        service = (SqlGatewayServiceImpl) SQL_GATEWAY_SERVICE_EXTENSION.getService();
+    }
+
+    @BeforeEach
+    void before() {
+        SessionEnvironment sessionEnvironment =
+                SessionEnvironment.newBuilder()
+                        .setSessionEndpointVersion(new EndpointVersion() {})
+                        .addSessionConfig(
+                                Collections.singletonMap(
+                                        SQL_GATEWAY_SESSION_PLAN_CACHE_ENABLED.key(), "true"))
+                        .build();
+        sessionHandle = service.openSession(sessionEnvironment);
+
+        Configuration flussConf = FLUSS_CLUSTER_EXTENSION.getClientConfig();
+        String bootstrapServers = String.join(",", flussConf.get(ConfigOptions.BOOTSTRAP_SERVERS));
+        service.configureSession(
+                sessionHandle,
+                String.format(
+                        "CREATE CATALOG %s WITH ('type' = 'fluss', 'bootstrap.servers' = '%s')",
+                        CATALOG_NAME, bootstrapServers),
+                -1);
+        service.configureSession(sessionHandle, "USE CATALOG " + CATALOG_NAME, -1);
+    }
+
+    @AfterEach
+    void after() {
+        if (sessionHandle != null) {
+            service.closeSession(sessionHandle);
+        }
+    }
+
+    @Test
+    void testCachedPlanLoadsFreshPartitionMetadata() throws Exception {
+        String tableName = "cached_plan_partitions";
+        TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
+        executeUpdate(
+                "CREATE TABLE "
+                        + tableName
+                        + " (id INT, dt STRING) PARTITIONED BY (dt) "
+                        + "WITH ('bucket.num' = '2')");
+
+        String query =
+                "SELECT partition_name, bucket_count FROM "
+                        + tableName
+                        + FlinkCatalog.PARTITIONS_TABLE_SUFFIX;
+
+        try (Connection connection =
+                        ConnectionFactory.createConnection(
+                                FLUSS_CLUSTER_EXTENSION.getClientConfig());
+                Admin admin = connection.getAdmin()) {
+            admin.createPartition(
+                            tablePath,
+                            new PartitionSpec(Collections.singletonMap("dt", "2026-10-25")),
+                            false)
+                    .get();
+            waitForPartitionCount(admin, tablePath, 1);
+
+            List<RowData> initialRows = executeQuery(query);
+            assertThat(initialRows).hasSize(1);
+            assertThat(initialRows.get(0).getString(0).toString()).isEqualTo("2026-10-25");
+            assertThat(initialRows.get(0).getInt(1)).isEqualTo(2);
+
+            admin.createPartition(
+                            tablePath,
+                            new PartitionSpec(Collections.singletonMap("dt", "2026-10-26")),
+                            false)
+                    .get();
+            waitForPartitionCount(admin, tablePath, 2);
+
+            List<RowData> refreshedRows = executeQuery(query);
+            assertThat(refreshedRows).hasSize(2);
+            assertThat(refreshedRows)
+                    .extracting(row -> row.getString(0).toString())
+                    .containsExactlyInAnyOrder("2026-10-25", "2026-10-26");
+            assertThat(refreshedRows).extracting(row -> row.getInt(1)).containsOnly(2);
+        }
+    }
+
+    private void executeUpdate(String statement) throws Exception {
+        OperationHandle operationHandle =
+                service.executeStatement(
+                        sessionHandle,
+                        statement,
+                        -1,
+                        new org.apache.flink.configuration.Configuration());
+        awaitOperationTermination(service, sessionHandle, operationHandle);
+        service.closeOperation(sessionHandle, operationHandle);
+    }
+
+    private List<RowData> executeQuery(String statement) throws Exception {
+        OperationHandle operationHandle =
+                service.executeStatement(
+                        sessionHandle,
+                        statement,
+                        -1,
+                        new org.apache.flink.configuration.Configuration());
+        awaitOperationTermination(service, sessionHandle, operationHandle);
+        List<RowData> results = fetchAllResults(service, sessionHandle, operationHandle);
+        service.closeOperation(sessionHandle, operationHandle);
+        return results;
+    }
+
+    private void waitForPartitionCount(Admin admin, TablePath tablePath, int expectedCount) {
+        retry(
+                Duration.ofMinutes(1),
+                () -> assertThat(admin.listPartitionInfos(tablePath).get()).hasSize(expectedCount));
+    }
+}
diff --git a/...s-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java b/...s-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java
@@ -142,6 +142,7 @@ public class FlinkCatalog extends AbstractCatalog {
     public static final String CHANGELOG_TABLE_SUFFIX = "$changelog";
     public static final String BINLOG_TABLE_SUFFIX = "$binlog";
     public static final String BUCKETS_TABLE_SUFFIX = "$buckets";
+    public static final String PARTITIONS_TABLE_SUFFIX = "$partitions";
 
     protected final ClassLoader classLoader;
 
@@ -446,6 +447,9 @@ public CatalogBaseTable getTable(ObjectPath objectPath)
         } else if (tableName.endsWith(BUCKETS_TABLE_SUFFIX)
                 && !tableName.contains(LAKE_TABLE_SPLITTER)) {
             return getVirtualBucketsTable(objectPath);
+        } else if (tableName.endsWith(PARTITIONS_TABLE_SUFFIX)
+                && !tableName.contains(LAKE_TABLE_SPLITTER)) {
+            return getVirtualPartitionsTable(objectPath);
         }
 
         TablePath tablePath = toTablePath(objectPath);
@@ -979,8 +983,8 @@ protected TablePath toTablePath(ObjectPath objectPath) {
 
     /**
      * Converts an {@link ObjectPath} to a physical {@link TablePath}, stripping any virtual table
-     * suffix ({@code $changelog}, {@code $binlog}, or {@code $buckets}) if present. This is needed
-     * because virtual tables are resolved from their base table.
+     * suffix ({@code $changelog}, {@code $binlog}, {@code $buckets}, or {@code $partitions}) if
+     * present. This is needed because virtual tables are resolved from their base table.
      */
     private TablePath toPhysicalTablePath(ObjectPath objectPath) {
         String tableName = objectPath.getObjectName();
@@ -994,6 +998,9 @@ private TablePath toPhysicalTablePath(ObjectPath objectPath) {
             tableName = tableName.substring(0, tableName.length() - BINLOG_TABLE_SUFFIX.length());
         } else if (tableName.endsWith(BUCKETS_TABLE_SUFFIX)) {
             tableName = tableName.substring(0, tableName.length() - BUCKETS_TABLE_SUFFIX.length());
+        } else if (tableName.endsWith(PARTITIONS_TABLE_SUFFIX)) {
+            tableName =
+                    tableName.substring(0, tableName.length() - PARTITIONS_TABLE_SUFFIX.length());
         }
         return TablePath.of(objectPath.getDatabaseName(), tableName);
     }
@@ -1007,7 +1014,8 @@ static boolean isVirtualTableName(String tableName) {
         return !tableName.contains(LAKE_TABLE_SPLITTER)
                 && (tableName.endsWith(CHANGELOG_TABLE_SUFFIX)
                         || tableName.endsWith(BINLOG_TABLE_SUFFIX)
-                        || tableName.endsWith(BUCKETS_TABLE_SUFFIX));
+                        || tableName.endsWith(BUCKETS_TABLE_SUFFIX)
+                        || tableName.endsWith(PARTITIONS_TABLE_SUFFIX));
     }
 
     private void rejectVirtualTableMutation(ObjectPath objectPath, String operation) {
@@ -1350,4 +1358,54 @@ private CatalogBaseTable getVirtualBucketsTable(ObjectPath objectPath)
                     t);
         }
     }
+
+    /** Creates a read-only virtual table exposing the current partition metadata snapshot. */
+    private CatalogBaseTable getVirtualPartitionsTable(ObjectPath objectPath)
+            throws TableNotExistException, CatalogException {
+        String virtualTableName = objectPath.getObjectName();
+        String baseTableName =
+                virtualTableName.substring(
+                        0, virtualTableName.length() - PARTITIONS_TABLE_SUFFIX.length());
+        ObjectPath baseObjectPath = new ObjectPath(objectPath.getDatabaseName(), baseTableName);
+
+        try {
+            // Only validate immutable table metadata here. Partition rows are loaded by the source
+            // reader at execution time so cached SQL plans observe the latest partitions.
+            TableInfo tableInfo = admin.getTableInfo(toTablePath(baseObjectPath)).get();
+            if (!tableInfo.isPartitioned()) {
+                throw new UnsupportedOperationException(
+                        String.format(
+                                "The $partitions virtual table is only supported for partitioned "
+                                        + "tables, but %s is not partitioned.",
+                                baseObjectPath));
+            }
+
+            Schema partitionsSchema =
+                    Schema.newBuilder()
+                            .column("partition_id", BIGINT().notNull())
+                            .column("partition_name", STRING().notNull())
+                            .column("bucket_count", INT().notNull())
+                            .build();
+            Map<String, String> options = new HashMap<>();
+            options.put(BOOTSTRAP_SERVERS.key(), bootstrapServers);
+            options.putAll(securityConfigs);
+            return CatalogTableAdapter.toCatalogTable(
+                    partitionsSchema,
+                    "Current partition IDs, names, and bucket counts",
+                    Collections.emptyList(),
+                    options);
+        } catch (UnsupportedOperationException e) {
+            throw e;
+        } catch (Exception e) {
+            Throwable t = ExceptionUtils.stripExecutionException(e);
+            if (isTableNotExist(t)) {
+                throw new TableNotExistException(getName(), baseObjectPath);
+            }
+            throw new CatalogException(
+                    String.format(
+                            "Failed to get virtual partitions table %s in %s",
+                            objectPath, getName()),
+                    t);
+        }
+    }
 }
diff --git a/...nk/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java b/...nk/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java
@@ -29,6 +29,7 @@
 import org.apache.fluss.flink.source.BucketsFlinkTableSource;
 import org.apache.fluss.flink.source.ChangelogFlinkTableSource;
 import org.apache.fluss.flink.source.FlinkTableSource;
+import org.apache.fluss.flink.source.PartitionsFlinkTableSource;
 import org.apache.fluss.flink.source.reader.LeaseContext;
 import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils;
 import org.apache.fluss.metadata.MergeEngineType;
@@ -119,6 +120,10 @@ public DynamicTableSource createDynamicTableSource(Context context) {
             return createBucketsTableSource(context, tableIdentifier, tableName);
         }
 
+        if (tableName.endsWith(FlinkCatalog.PARTITIONS_TABLE_SUFFIX)) {
+            return createPartitionsTableSource(context, tableIdentifier, tableName);
+        }
+
         FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
         final ReadableConfig tableOptions = helper.getOptions();
         validateSourceOptions(tableOptions);
@@ -492,4 +497,18 @@ private DynamicTableSource createBucketsTableSource(
                         context.getCatalogTable().getOptions(), context.getConfiguration()),
                 tableOutputType);
     }
+
+    /** Creates a bounded source for a {@code $partitions} virtual table. */
+    private DynamicTableSource createPartitionsTableSource(
+            Context context, ObjectIdentifier tableIdentifier, String tableName) {
+        String baseTableName =
+                tableName.substring(
+                        0, tableName.length() - FlinkCatalog.PARTITIONS_TABLE_SUFFIX.length());
+        RowType tableOutputType = (RowType) context.getPhysicalRowDataType().getLogicalType();
+        return new PartitionsFlinkTableSource(
+                TablePath.of(tableIdentifier.getDatabaseName(), baseTableName),
+                toFlussClientConfig(
+                        context.getCatalogTable().getOptions(), context.getConfiguration()),
+                tableOutputType);
+    }
 }
diff --git a/...nk-common/src/main/java/org/apache/fluss/flink/procedure/ListPartitionInfosProcedure.java b/...nk-common/src/main/java/org/apache/fluss/flink/procedure/ListPartitionInfosProcedure.java
@@ -43,7 +43,11 @@
  * -- List the partition id, partition name and bucket count of each partition
  * CALL sys.list_partition_infos('my_db', 'my_table');
  * </pre>
+ *
+ * @deprecated Use the read-only {@code my_table$partitions} virtual table instead. This procedure
+ *     remains available for backward compatibility.
  */
+@Deprecated
 public class ListPartitionInfosProcedure extends ProcedureBase {
 
     @ProcedureHint(