This is an automated email from the ASF dual-hosted git repository. gyfora pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 51074d3e860d81720c7c69e49faacb4111f0f253 Author: Gabor Somogyi <[email protected]> AuthorDate: Mon Jul 20 12:04:57 2026 +0200 [FLINK-40176][state-processor-api] Introduce basic StateCatalog functionality Introduces StateCatalog, a read-only Flink SQL catalog that discovers checkpoints/savepoints from configured directories and exposes their metadata as queryable databases and views. This first slice covers catalog registration (StateCatalogFactory/StateCatalogOptions), directory scanning and database naming (SnapshotDiscovery), and the per-snapshot "metadata" view backed by the savepoint_metadata table function. Keyed/non-keyed state table support is added in later commits. --- .../flink/state/catalog/SnapshotDiscovery.java | 337 ++++++++++++++ .../apache/flink/state/catalog/StateCatalog.java | 493 +++++++++++++++++++++ .../flink/state/catalog/StateCatalogFactory.java | 86 ++++ .../flink/state/catalog/StateCatalogOptions.java | 59 +++ .../org.apache.flink.table.factories.Factory | 1 + .../flink/state/catalog/SnapshotDiscoveryTest.java | 248 +++++++++++ .../state/catalog/StateCatalogDiscoveryITCase.java | 208 +++++++++ .../flink/state/catalog/StateCatalogTest.java | 143 ++++++ 8 files changed, 1575 insertions(+) diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/SnapshotDiscovery.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/SnapshotDiscovery.java new file mode 100644 index 00000000000..cc73ce36c3e --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/SnapshotDiscovery.java @@ -0,0 +1,337 @@ +/* + * 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.flink.state.catalog; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.core.fs.FileStatus; +import org.apache.flink.core.fs.Path; +import org.apache.flink.util.StringUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.time.Instant; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Discovers Flink checkpoints and savepoints within a set of labelled directories. + * + * <p>Each configured directory is associated with a user-chosen label. By default, database names + * are derived as {@code label/creationTs/relative-path}, where {@code creationTs} is the + * modification time of the snapshot's {@code _metadata} file formatted as {@code + * yyyy-MM-dd'T'HH:mm:ssX} (e.g. {@code 2026-07-22T10:30:45Z}) and {@code relative-path} is the + * verbatim path from the configured directory to the snapshot directory (e.g. {@code + * my-app/2026-07-22T10:30:45Z/savepoint-acce1cedsad} or {@code + * my-app/2026-07-22T10:30:45Z/a1b2c3d4.../chk-3}). The {@code creationTs} segment can be disabled + * via {@code dbNameIncludeTs}, in which case names fall back to {@code label/relative-path}. + */ +@Internal +class SnapshotDiscovery { + + private static final Logger LOG = LoggerFactory.getLogger(SnapshotDiscovery.class); + + private static final String METADATA_FILE_NAME = "_metadata"; + + private static final DateTimeFormatter CREATION_TS_FORMATTER = + DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssX").withZone(ZoneOffset.UTC); + + private final Map<String, Path> labelToDir; + private final int listingParallelism; + private final boolean dbNameIncludeTs; + + private ExecutorService listingExecutor; + + SnapshotDiscovery( + Map<String, String> labelToDirPath, int listingParallelism, boolean dbNameIncludeTs) { + this.labelToDir = validateAndConvert(labelToDirPath); + this.listingParallelism = listingParallelism; + this.dbNameIncludeTs = dbNameIncludeTs; + } + + void start() { + listingExecutor = createListingExecutor(listingParallelism); + } + + void stop() { + if (listingExecutor != null) { + listingExecutor.shutdownNow(); + listingExecutor = null; + } + } + + // ------------------------------------------------------------------------- + // Public API + // ------------------------------------------------------------------------- + + /** + * Full BFS scan of all configured directories, listing directories at the same depth + * concurrently (up to {@code listingParallelism} at a time, with automatic backpressure via + * {@link ThreadPoolExecutor.CallerRunsPolicy}). Returns one database name per discovered {@code + * _metadata} file. + * + * @throws IOException if every configured directory fails to scan + */ + List<String> list() throws IOException { + List<String> result = new ArrayList<>(); + Set<String> seen = new LinkedHashSet<>(); + IOException err = null; + boolean allFailed = true; + + for (Map.Entry<String, Path> entry : labelToDir.entrySet()) { + String label = entry.getKey(); + Path dir = entry.getValue(); + try { + for (FileStatus metadataFileStatus : findMetadataFileStatuses(dir)) { + String dbName = buildDatabaseName(label, dir, metadataFileStatus); + if (!seen.add(dbName)) { + LOG.warn("Duplicate database name '{}'. Skipping.", dbName); + continue; + } + result.add(dbName); + } + allFailed = false; + } catch (IOException e) { + LOG.warn("Failed to scan {}: {}", dir, e.getMessage(), e); + err = e; + } + } + + if (allFailed) { + throw new IOException("All configured directories failed to scan", err); + } + return result; + } + + /** + * Reverses {@link #buildDatabaseName} to recover the label and the verbatim relative path from + * {@code dbName}, then verifies the snapshot's {@code _metadata} file with a single {@code + * getFileStatus} call. Returns the snapshot directory path if it exists. + */ + Optional<String> find(String dbName) { + if (StringUtils.isNullOrWhitespaceOnly(dbName)) { + return Optional.empty(); + } + int labelSlash = dbName.indexOf('/'); + if (labelSlash == 0) { + return Optional.empty(); + } + + String label; + String relativePath; + if (dbNameIncludeTs) { + // A name with no '/' has no room for the mandatory creationTs segment. + if (labelSlash < 0) { + return Optional.empty(); + } + label = dbName.substring(0, labelSlash); + String afterLabel = dbName.substring(labelSlash + 1); + int tsSlash = afterLabel.indexOf('/'); + relativePath = tsSlash < 0 ? "" : afterLabel.substring(tsSlash + 1); + } else { + // A name with no '/' means the configured directory itself is the snapshot. + label = labelSlash < 0 ? dbName : dbName.substring(0, labelSlash); + relativePath = labelSlash < 0 ? "" : dbName.substring(labelSlash + 1); + } + + Path dir = labelToDir.get(label); + if (dir == null) { + return Optional.empty(); + } + + Path metadataFile = + relativePath.isEmpty() + ? new Path(dir, METADATA_FILE_NAME) + : new Path(dir, relativePath + "/" + METADATA_FILE_NAME); + try { + dir.getFileSystem().getFileStatus(metadataFile); + return Optional.of(metadataFile.getParent().toString()); + } catch (IOException e) { + return Optional.empty(); + } + } + + // ------------------------------------------------------------------------- + // Full BFS scan + // ------------------------------------------------------------------------- + + private List<FileStatus> findMetadataFileStatuses(Path directory) throws IOException { + List<FileStatus> metadataFiles = new ArrayList<>(); + List<Path> currentLevel = Collections.singletonList(directory); + boolean allFailed = true; + IOException err = null; + + while (!currentLevel.isEmpty()) { + List<Future<FileStatus[]>> futures = new ArrayList<>(currentLevel.size()); + for (Path dir : currentLevel) { + futures.add(listingExecutor.submit(() -> listDirectory(dir))); + } + + List<Path> nextLevel = new ArrayList<>(); + for (int i = 0; i < futures.size(); i++) { + FileStatus[] statuses = null; + try { + statuses = getResult(futures.get(i)); + } catch (IOException e) { + LOG.warn("Failed to list {}: {}", currentLevel.get(i), e.getMessage()); + err = e; + } + if (statuses == null) { + continue; + } + allFailed = false; + for (FileStatus status : statuses) { + if (status.isDir()) { + nextLevel.add(status.getPath()); + } else if (METADATA_FILE_NAME.equals(status.getPath().getName())) { + metadataFiles.add(status); + } + } + } + currentLevel = nextLevel; + } + + if (allFailed) { + throw new IOException("All directory listings failed under: " + directory, err); + } + return metadataFiles; + } + + private FileStatus[] listDirectory(Path dir) throws IOException { + FileStatus[] result = dir.getFileSystem().listStatus(dir); + if (result == null) { + throw new IOException( + "Cannot list directory (path does not exist or is not a directory): " + dir); + } + return result; + } + + private static FileStatus[] getResult(Future<FileStatus[]> future) throws IOException { + try { + return future.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + if (cause instanceof IOException) { + throw (IOException) cause; + } + throw new IOException("Directory listing failed", cause); + } + } + + // ------------------------------------------------------------------------- + // Database name derivation + // ------------------------------------------------------------------------- + + private String buildDatabaseName(String label, Path configuredDir, FileStatus metadataFile) { + Path snapshotDir = metadataFile.getPath().getParent(); + String configuredPath = configuredDir.toUri().getPath(); + String snapshotPath = snapshotDir.toUri().getPath(); + + String relative = snapshotPath.substring(configuredPath.length()); + if (relative.startsWith("/")) { + relative = relative.substring(1); + } + + StringBuilder dbName = new StringBuilder(label); + if (dbNameIncludeTs) { + Instant creationTs = Instant.ofEpochMilli(metadataFile.getModificationTime()); + dbName.append('/').append(CREATION_TS_FORMATTER.format(creationTs)); + } + if (!relative.isEmpty()) { + dbName.append('/').append(relative); + } + return dbName.toString(); + } + + // ------------------------------------------------------------------------- + // Construction-time helpers + // ------------------------------------------------------------------------- + + /** + * Converts the configured label → directory paths, rejecting empty configurations, directories + * assigned to more than one label, and directories nested inside one another (which would + * discover the same snapshots under multiple labels). + */ + private static Map<String, Path> validateAndConvert(Map<String, String> labelToDirPath) { + if (labelToDirPath.isEmpty()) { + throw new IllegalArgumentException( + "At least one directory must be configured via 'directory.{label}' options."); + } + + Map<String, Path> result = new LinkedHashMap<>(); + Set<String> normalizedDirs = new LinkedHashSet<>(); + for (Map.Entry<String, String> entry : labelToDirPath.entrySet()) { + Path dir = new Path(entry.getValue()); + if (!normalizedDirs.add(dir.toUri().getPath())) { + throw new IllegalArgumentException( + String.format( + "Directory '%s' is assigned to more than one label.", + entry.getValue())); + } + result.put(entry.getKey(), dir); + } + + for (String dir : normalizedDirs) { + String prefix = dir.endsWith("/") ? dir : dir + "/"; + for (String other : normalizedDirs) { + if (!other.equals(dir) && other.startsWith(prefix)) { + throw new IllegalArgumentException( + String.format( + "Directory '%s' is an ancestor of '%s'. Providing both would " + + "discover the same snapshots under multiple labels.", + dir, other)); + } + } + } + return result; + } + + private static ExecutorService createListingExecutor(int parallelism) { + return new ThreadPoolExecutor( + parallelism, + parallelism, + 0L, + TimeUnit.MILLISECONDS, + new LinkedBlockingQueue<>(parallelism), + r -> { + Thread t = new Thread(r, "state-catalog-listing"); + t.setDaemon(true); + return t; + }, + new ThreadPoolExecutor.CallerRunsPolicy()); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java new file mode 100644 index 00000000000..ccf3aa1234d --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java @@ -0,0 +1,493 @@ +/* + * 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.flink.state.catalog; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.AbstractCatalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogDatabase; +import org.apache.flink.table.catalog.CatalogDatabaseImpl; +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.CatalogPartition; +import org.apache.flink.table.catalog.CatalogPartitionSpec; +import org.apache.flink.table.catalog.CatalogView; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogTableStatistics; +import org.apache.flink.table.expressions.Expression; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** + * A read-only Flink SQL catalog that discovers checkpoints and savepoints from a configured set of + * directories and exposes their metadata as queryable SQL databases and views. + * + * <p>The catalog maps Flink's three-level hierarchy as follows: + * + * <ul> + * <li>Catalog: the name given at registration time (e.g. {@code "state"}) + * <li>Database: one entry per discovered snapshot (e.g. {@code "app1_savepoint-acce1cedsad"}) + * <li>Table: a single view named {@code "metadata"} per database, backed by the {@code + * savepoint_metadata} function from {@code StateModule} + * </ul> + * + * <p>Database names preserve hyphens from the original directory names. Backtick quoting is + * required in SQL for identifiers containing hyphens: + * + * <pre>{@code + * USE CATALOG state; + * USE `app1_savepoint-acce1cedsad`; + * SELECT * FROM metadata; + * }</pre> + * + * <p>{@code StateModule} must be loaded before querying any {@code metadata} view: + * + * <pre>{@code + * tableEnv.loadModule("state", StateModule.INSTANCE); + * }</pre> + * + * <p>Each catalog operation fetches state on demand. {@link #listDatabases()} performs a full + * directory scan; all other operations perform a single file check on the specific snapshot path + * reconstructed from the database name. There is no background polling and no shared cache. + * + * <p>All write operations throw {@link UnsupportedOperationException}. + */ +@PublicEvolving +public class StateCatalog extends AbstractCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(StateCatalog.class); + + public static final String METADATA_TABLE = "metadata"; + + private static final CatalogDatabase EMPTY_DATABASE = + new CatalogDatabaseImpl(Collections.emptyMap(), ""); + + private final SnapshotDiscovery discovery; + + public StateCatalog(String name, Map<String, String> labelsToDirs) { + this(name, labelsToDirs, StateCatalogOptions.LISTING_PARALLELISM.defaultValue()); + } + + public StateCatalog(String name, Map<String, String> labelsToDirs, int listingParallelism) { + this( + name, + labelsToDirs, + listingParallelism, + StateCatalogOptions.DB_NAME_INCLUDE_TS.defaultValue()); + } + + public StateCatalog( + String name, + Map<String, String> labelsToDirs, + int listingParallelism, + boolean dbNameIncludeTs) { + super(name, "default"); + this.discovery = new SnapshotDiscovery(labelsToDirs, listingParallelism, dbNameIncludeTs); + } + + @Override + @Nullable + public String getDefaultDatabase() { + return null; + } + + // ------------------------------------------------------------------------- + // Lifecycle + // ------------------------------------------------------------------------- + + @Override + public void open() throws CatalogException { + discovery.start(); + listDatabases(); + } + + @Override + public void close() throws CatalogException { + discovery.stop(); + } + + // ------------------------------------------------------------------------- + // Databases + // ------------------------------------------------------------------------- + + @Override + public List<String> listDatabases() throws CatalogException { + try { + return discovery.list(); + } catch (IOException e) { + LOG.warn("Failed to list databases in catalog '{}'", getName(), e); + return Collections.emptyList(); + } + } + + @Override + public CatalogDatabase getDatabase(String databaseName) + throws DatabaseNotExistException, CatalogException { + if (discovery.find(databaseName).isEmpty()) { + throw new DatabaseNotExistException(getName(), databaseName); + } + return EMPTY_DATABASE; + } + + @Override + public boolean databaseExists(String databaseName) throws CatalogException { + return discovery.find(databaseName).isPresent(); + } + + @Override + public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists) + throws DatabaseAlreadyExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade) + throws DatabaseNotExistException, DatabaseNotEmptyException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void alterDatabase(String name, CatalogDatabase newDatabase, boolean ignoreIfNotExists) + throws DatabaseNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + // ------------------------------------------------------------------------- + // Tables and views + // ------------------------------------------------------------------------- + + @Override + public List<String> listTables(String databaseName) + throws DatabaseNotExistException, CatalogException { + if (discovery.find(databaseName).isEmpty()) { + throw new DatabaseNotExistException(getName(), databaseName); + } + return Collections.emptyList(); + } + + @Override + public List<String> listViews(String databaseName) + throws DatabaseNotExistException, CatalogException { + if (discovery.find(databaseName).isEmpty()) { + throw new DatabaseNotExistException(getName(), databaseName); + } + return Collections.singletonList(METADATA_TABLE); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath) + throws TableNotExistException, CatalogException { + Optional<String> snapshotPath = discovery.find(tablePath.getDatabaseName()); + if (snapshotPath.isEmpty()) { + throw new TableNotExistException(getName(), tablePath); + } + String tableName = tablePath.getObjectName(); + if (METADATA_TABLE.equals(tableName)) { + return buildMetadataView(snapshotPath.get()); + } + throw new TableNotExistException(getName(), tablePath); + } + + @Override + public boolean tableExists(ObjectPath tablePath) throws CatalogException { + Optional<String> snapshotPath = discovery.find(tablePath.getDatabaseName()); + if (snapshotPath.isEmpty()) { + return false; + } + return METADATA_TABLE.equals(tablePath.getObjectName()); + } + + @Override + public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) + throws TableAlreadyExistException, DatabaseNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void alterTable( + ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void renameTable(ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists) + throws TableAlreadyExistException, TableNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + // ------------------------------------------------------------------------- + // Partitions (not supported) + // ------------------------------------------------------------------------- + + @Override + public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath) + throws TableNotExistException, TableNotPartitionedException, CatalogException { + if (!tableExists(tablePath)) { + throw new TableNotExistException(getName(), tablePath); + } + throw new TableNotPartitionedException(getName(), tablePath); + } + + @Override + public List<CatalogPartitionSpec> listPartitions( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws TableNotExistException, TableNotPartitionedException, CatalogException { + return listPartitions(tablePath); + } + + @Override + public List<CatalogPartitionSpec> listPartitionsByFilter( + ObjectPath tablePath, List<Expression> filters) + throws TableNotExistException, TableNotPartitionedException, CatalogException { + return listPartitions(tablePath); + } + + @Override + public CatalogPartition getPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws PartitionNotExistException, CatalogException { + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + @Override + public boolean partitionExists(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws CatalogException { + return false; + } + + @Override + public void createPartition( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogPartition partition, + boolean ignoreIfExists) + throws TableNotExistException, + TableNotPartitionedException, + PartitionSpecInvalidException, + PartitionAlreadyExistsException, + CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void dropPartition( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists) + throws PartitionNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void alterPartition( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogPartition newPartition, + boolean ignoreIfNotExists) + throws PartitionNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + // ------------------------------------------------------------------------- + // Functions (not supported) + // ------------------------------------------------------------------------- + + @Override + public List<String> listFunctions(String dbName) + throws DatabaseNotExistException, CatalogException { + return Collections.emptyList(); + } + + @Override + public CatalogFunction getFunction(ObjectPath functionPath) + throws FunctionNotExistException, CatalogException { + throw new FunctionNotExistException(getName(), functionPath); + } + + @Override + public boolean functionExists(ObjectPath functionPath) throws CatalogException { + return false; + } + + @Override + public void createFunction( + ObjectPath functionPath, CatalogFunction function, boolean ignoreIfExists) + throws FunctionAlreadyExistException, DatabaseNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void alterFunction( + ObjectPath functionPath, CatalogFunction newFunction, boolean ignoreIfNotExists) + throws FunctionNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void dropFunction(ObjectPath functionPath, boolean ignoreIfNotExists) + throws FunctionNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + // ------------------------------------------------------------------------- + // Statistics (read-only stubs) + // ------------------------------------------------------------------------- + + @Override + public CatalogTableStatistics getTableStatistics(ObjectPath tablePath) + throws TableNotExistException, CatalogException { + if (!tableExists(tablePath)) { + throw new TableNotExistException(getName(), tablePath); + } + return CatalogTableStatistics.UNKNOWN; + } + + @Override + public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath) + throws TableNotExistException, CatalogException { + if (!tableExists(tablePath)) { + throw new TableNotExistException(getName(), tablePath); + } + return CatalogColumnStatistics.UNKNOWN; + } + + @Override + public CatalogTableStatistics getPartitionStatistics( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws PartitionNotExistException, CatalogException { + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + @Override + public CatalogColumnStatistics getPartitionColumnStatistics( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws PartitionNotExistException, CatalogException { + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + @Override + public void alterTableStatistics( + ObjectPath tablePath, CatalogTableStatistics tableStatistics, boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void alterTableColumnStatistics( + ObjectPath tablePath, + CatalogColumnStatistics columnStatistics, + boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void alterPartitionStatistics( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogTableStatistics partitionStatistics, + boolean ignoreIfNotExists) + throws PartitionNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + @Override + public void alterPartitionColumnStatistics( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogColumnStatistics columnStatistics, + boolean ignoreIfNotExists) + throws PartitionNotExistException, CatalogException { + throw new UnsupportedOperationException("StateCatalog is read-only."); + } + + // ------------------------------------------------------------------------- + // View construction + // ------------------------------------------------------------------------- + + private static CatalogView buildMetadataView(String snapshotPath) { + String escapedPath = snapshotPath.replace("'", "''"); + String query = String.format("SELECT * FROM TABLE(savepoint_metadata('%s'))", escapedPath); + // Once the upstream StateCatalog PR is merged and OUTPUT_DATA_TYPE is available in + // SavepointMetadataTableFunction, replace the schema definition below with: + // Schema.newBuilder() + // .fromRowDataType(SavepointMetadataTableFunction.OUTPUT_DATA_TYPE) + // .build() + + Schema schema = + Schema.newBuilder() + .fromRowDataType( + DataTypes.ROW( + DataTypes.FIELD( + "checkpoint-id", DataTypes.BIGINT().notNull()), + DataTypes.FIELD("operator-name", DataTypes.STRING()), + DataTypes.FIELD("operator-uid", DataTypes.STRING()), + DataTypes.FIELD( + "operator-uid-hash", DataTypes.STRING().notNull()), + DataTypes.FIELD( + "operator-parallelism", DataTypes.INT().notNull()), + DataTypes.FIELD( + "operator-max-parallelism", + DataTypes.INT().notNull()), + DataTypes.FIELD( + "operator-subtask-state-count", + DataTypes.INT().notNull()), + DataTypes.FIELD( + "operator-coordinator-state-size-in-bytes", + DataTypes.BIGINT().notNull()), + DataTypes.FIELD( + "operator-total-size-in-bytes", + DataTypes.BIGINT().notNull()))) + .build(); + return CatalogView.of( + schema, + "Operator metadata for snapshot at " + snapshotPath, + query, + query, + Collections.emptyMap()); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalogFactory.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalogFactory.java new file mode 100644 index 00000000000..9f18a8ae9bc --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalogFactory.java @@ -0,0 +1,86 @@ +/* + * 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.flink.state.catalog; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.factories.CatalogFactory; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Set; + +/** + * Factory for creating {@link StateCatalog} instances via SQL DDL or programmatically. + * + * <p>Directories are configured with {@code directory.{label}} options: + * + * <pre>{@code + * CREATE CATALOG state WITH ( + * 'type' = 'state', + * 'directory.my-app' = '/checkpoints/app1', + * 'directory.staging' = '/savepoints/staging' + * ); + * }</pre> + */ +@PublicEvolving +public class StateCatalogFactory implements CatalogFactory { + + public static final String IDENTIFIER = "state"; + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return Collections.emptySet(); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return Set.of( + StateCatalogOptions.LISTING_PARALLELISM, StateCatalogOptions.DB_NAME_INCLUDE_TS); + } + + @Override + public Catalog createCatalog(Context context) { + Map<String, String> options = context.getOptions(); + + Map<String, String> labelsToDirs = new LinkedHashMap<>(); + for (Map.Entry<String, String> entry : options.entrySet()) { + if (entry.getKey().startsWith(StateCatalogOptions.DIRECTORY_PREFIX)) { + String label = + entry.getKey().substring(StateCatalogOptions.DIRECTORY_PREFIX.length()); + labelsToDirs.put(label, entry.getValue()); + } + } + + Configuration configuration = Configuration.fromMap(options); + return new StateCatalog( + context.getName(), + labelsToDirs, + configuration.get(StateCatalogOptions.LISTING_PARALLELISM), + configuration.get(StateCatalogOptions.DB_NAME_INCLUDE_TS)); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalogOptions.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalogOptions.java new file mode 100644 index 00000000000..6b0e7f10a67 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalogOptions.java @@ -0,0 +1,59 @@ +/* + * 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.flink.state.catalog; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; + +/** Configuration options for {@link StateCatalog}. */ +@PublicEvolving +public class StateCatalogOptions { + + /** + * Prefix for directory options. Each option of the form {@code directory.{label}} maps a + * human-readable label to a filesystem path. The label becomes the first segment of every + * database name derived from that directory (e.g. {@code my-app/savepoint-abc}). + */ + public static final String DIRECTORY_PREFIX = "directory."; + + public static final ConfigOption<Integer> LISTING_PARALLELISM = + ConfigOptions.key("listing-parallelism") + .intType() + .defaultValue(10) + .withDescription( + "Maximum number of concurrent directory listing requests issued " + + "during a scan. Directories at the same depth are listed " + + "in parallel. Increase for high-latency remote filesystems " + + "(e.g. S3); decrease to reduce load on the filesystem."); + + public static final ConfigOption<Boolean> DB_NAME_INCLUDE_TS = + ConfigOptions.key("db-name.include-ts") + .booleanType() + .defaultValue(true) + .withDescription( + "Whether derived database names include the snapshot's creation " + + "timestamp as a segment, i.e. label/creationTs/relativePath " + + "instead of label/relativePath. The timestamp is the " + + "modification time of the snapshot's _metadata file, " + + "formatted with yyyy-MM-dd'T'HH:mm:ssX (e.g. " + + "2026-07-22T10:30:45Z)."); + + private StateCatalogOptions() {} +} diff --git a/flink-libraries/flink-state-processing-api/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-libraries/flink-state-processing-api/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory index c5e2715f26d..c8bac30c71d 100644 --- a/flink-libraries/flink-state-processing-api/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory +++ b/flink-libraries/flink-state-processing-api/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory @@ -15,3 +15,4 @@ org.apache.flink.state.table.module.StateModuleFactory org.apache.flink.state.table.SavepointDynamicTableSourceFactory +org.apache.flink.state.catalog.StateCatalogFactory diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/SnapshotDiscoveryTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/SnapshotDiscoveryTest.java new file mode 100644 index 00000000000..fc0e3bd1589 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/SnapshotDiscoveryTest.java @@ -0,0 +1,248 @@ +/* + * 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.flink.state.catalog; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Unit tests for {@link SnapshotDiscovery}. */ +class SnapshotDiscoveryTest { + + // Every metadata file created by createMetadataFile() gets this exact modification time, so + // the creationTs segment in derived database names is deterministic across all tests. + private static final Instant FIXED_TS = Instant.parse("2024-03-15T10:30:45Z"); + private static final String TS = "2024-03-15T10:30:45Z"; + + private static final String MISSING_DIR = "/nonexistent-snapshot-discovery-test-path"; + + @TempDir Path tempDir; + + private final List<SnapshotDiscovery> started = new ArrayList<>(); + + private SnapshotDiscovery discovery; + + @BeforeEach + void setUp() { + discovery = start(Collections.singletonMap("app", tempDir.toString()), true); + } + + @AfterEach + void tearDown() { + started.forEach(SnapshotDiscovery::stop); + } + + // ------------------------------------------------------------------------- + // Construction-time validation + // ------------------------------------------------------------------------- + + @Test + void testConstructionValidation() { + assertThatThrownBy(() -> new SnapshotDiscovery(Collections.emptyMap(), 2, true)) + .as("no directory configured") + .isInstanceOf(IllegalArgumentException.class); + + assertThatThrownBy(() -> new SnapshotDiscovery(dirs("/state/app", "/state/app"), 2, true)) + .as("same directory under two labels") + .isInstanceOf(IllegalArgumentException.class); + + assertThatThrownBy(() -> new SnapshotDiscovery(dirs("/state", "/state/app"), 2, true)) + .as("one directory nested inside the other") + .isInstanceOf(IllegalArgumentException.class); + } + + // ------------------------------------------------------------------------- + // find() + // ------------------------------------------------------------------------- + + @Test + void testFindRejectsInvalidDatabaseNames() { + assertThat(discovery.find(null)).isEmpty(); + assertThat(discovery.find("")).isEmpty(); + assertThat(discovery.find(" ")).isEmpty(); + assertThat(discovery.find("/savepoint-abc")).isEmpty(); + assertThat(discovery.find("unknown/" + TS + "/savepoint-abc")).isEmpty(); + + // With db-name.include-ts enabled (the default), a name with no '/' has no room for the + // mandatory creationTs segment, so it can never match. + assertThat(discovery.find("app")).isEmpty(); + assertThat(discovery.find("savepoint-abc")).isEmpty(); + } + + @Test + void testFindResolvesRelativePathVerbatim() throws IOException { + createMetadataFile(tempDir.resolve("savepoint-abc")); + createMetadataFile(tempDir.resolve("jobId").resolve("chk-3")); + createMetadataFile(tempDir.resolve("a").resolve("b").resolve("c")); + + assertThat(discovery.find("app/" + TS + "/savepoint-abc")) + .hasValue(tempDir.resolve("savepoint-abc").toString()); + assertThat(discovery.find("app/" + TS + "/jobId/chk-3")) + .hasValue(tempDir.resolve("jobId").resolve("chk-3").toString()); + assertThat(discovery.find("app/" + TS + "/a/b/c")) + .hasValue(tempDir.resolve("a").resolve("b").resolve("c").toString()); + + assertThat(discovery.find("app/" + TS + "/savepoint-nonexistent")).isEmpty(); + } + + @Test + void testFindSnapshotWithTrailingSlash() throws IOException { + createMetadataFile(tempDir.resolve("savepoint-abc")); + + // trailing slash after the creationTs → relativePath is empty, same as a ts-only path + assertThat(discovery.find("app/" + TS + "/")).isEmpty(); + } + + @Test + void testFindTsOnlyPathMatchesSnapshotDirectlyUnderConfiguredDir() throws IOException { + createMetadataFile(tempDir); + + assertThat(discovery.find("app/" + TS)).hasValue(tempDir.toString()); + } + + @Test + void testFindWithTsDisabled() throws IOException { + SnapshotDiscovery noTs = start(Collections.singletonMap("app", tempDir.toString()), false); + createMetadataFile(tempDir.resolve("savepoint-abc")); + + assertThat(noTs.find("app/savepoint-abc")) + .hasValue(tempDir.resolve("savepoint-abc").toString()); + // With db-name.include-ts disabled, everything after the label is taken verbatim as the + // relative path — no segment is skipped as a timestamp. + assertThat(noTs.find("app/extra-segment/savepoint-abc")).isEmpty(); + } + + // ------------------------------------------------------------------------- + // list() + // ------------------------------------------------------------------------- + + @Test + void testListReflectsFilesystemChangesWithoutCaching() throws IOException { + assertThat(discovery.list()).isEmpty(); + + createMetadataFile(tempDir.resolve("savepoint-new")); + assertThat(discovery.list()).containsExactly("app/" + TS + "/savepoint-new"); + + Files.delete(tempDir.resolve("savepoint-new").resolve("_metadata")); + assertThat(discovery.list()).isEmpty(); + } + + @Test + void testListMultipleSnapshots() throws IOException { + createMetadataFile(tempDir.resolve("savepoint-a")); + createMetadataFile(tempDir.resolve("savepoint-b")); + createMetadataFile(tempDir.resolve("jobId").resolve("chk-1")); + + assertThat(discovery.list()) + .containsExactlyInAnyOrder( + "app/" + TS + "/savepoint-a", + "app/" + TS + "/savepoint-b", + "app/" + TS + "/jobId/chk-1"); + } + + @Test + void testListNonMetadataFilesIgnored() throws IOException { + Files.createDirectories(tempDir.resolve("savepoint-a")); + Files.createFile(tempDir.resolve("savepoint-a").resolve("other.file")); + + assertThat(discovery.list()).isEmpty(); + } + + @Test + void testListReturnsSnapshotsFromHealthyDirectoryWhenOtherFails() throws IOException { + createMetadataFile(tempDir.resolve("savepoint-ok")); + + Map<String, String> labelToDir = new LinkedHashMap<>(); + labelToDir.put("good", tempDir.toString()); + labelToDir.put("bad", MISSING_DIR); + + assertThat(start(labelToDir, true).list()).containsExactly("good/" + TS + "/savepoint-ok"); + } + + @Test + void testListThrowsWhenAllDirectoriesFail() { + SnapshotDiscovery allBad = start(Collections.singletonMap("bad", MISSING_DIR), true); + + assertThatThrownBy(allBad::list) + .isInstanceOf(IOException.class) + .hasMessageContaining("All configured directories failed") + .cause() + .isInstanceOf(IOException.class) + .hasMessageContaining("All directory listings failed"); + } + + @Test + void testListWithTsDisabled() throws IOException { + SnapshotDiscovery noTs = start(Collections.singletonMap("app", tempDir.toString()), false); + createMetadataFile(tempDir.resolve("savepoint-a")); + createMetadataFile(tempDir.resolve("jobId").resolve("chk-1")); + + assertThat(noTs.list()).containsExactlyInAnyOrder("app/savepoint-a", "app/jobId/chk-1"); + } + + @Test + void testDbNameCreationTsMatchesExpectedFormat() throws IOException { + // Uses the real (unset) modification time to verify the formatter itself, rather than the + // fixed FIXED_TS used elsewhere in this file. + Files.createDirectories(tempDir.resolve("savepoint-live")); + Files.createFile(tempDir.resolve("savepoint-live").resolve("_metadata")); + + assertThat(discovery.list().get(0)) + .matches("app/\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z/savepoint-live"); + } + + // ------------------------------------------------------------------------- + // Helpers + // ------------------------------------------------------------------------- + + private SnapshotDiscovery start(Map<String, String> labelToDir, boolean dbNameIncludeTs) { + SnapshotDiscovery snapshotDiscovery = new SnapshotDiscovery(labelToDir, 2, dbNameIncludeTs); + snapshotDiscovery.start(); + started.add(snapshotDiscovery); + return snapshotDiscovery; + } + + private static Map<String, String> dirs(String firstDir, String secondDir) { + Map<String, String> labelToDir = new LinkedHashMap<>(); + labelToDir.put("a", firstDir); + labelToDir.put("b", secondDir); + return labelToDir; + } + + private static void createMetadataFile(Path snapshotDir) throws IOException { + Files.createDirectories(snapshotDir); + Path file = Files.createFile(snapshotDir.resolve("_metadata")); + Files.setLastModifiedTime(file, FileTime.from(FIXED_TS)); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/StateCatalogDiscoveryITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/StateCatalogDiscoveryITCase.java new file mode 100644 index 00000000000..92d0bc256dd --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/StateCatalogDiscoveryITCase.java @@ -0,0 +1,208 @@ +/* + * 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.flink.state.catalog; + +import org.apache.flink.runtime.checkpoint.Checkpoints; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.state.table.module.StateModule; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.catalog.CatalogView; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Integration tests for basic {@link StateCatalog} functionality driven through {@code CREATE + * CATALOG} DDL and SQL: multi-label discovery and the {@code metadata} view. Checkpoint metadata is + * written directly via {@link Checkpoints#storeCheckpointMetadata} — no minicluster or real state + * backend is involved, so these tests are backend-agnostic by construction. + * + * <p>For reads of real (generated) keyed-state savepoints, see {@code + * StateCatalogGeneratedSavepointITCase} (HashMap-only, checked-in fixtures) and {@code + * KeyedStateReadingITCase} (parameterized across backends, savepoints taken at runtime). + */ +class StateCatalogDiscoveryITCase { + + @Test + void testMetadataQueryReturnsOperators(@TempDir Path tempDir) throws Exception { + OperatorID opId1 = new OperatorID(1, 2); + OperatorState op1 = new OperatorState("source", "source-uid", opId1, 2, 128); + OperatorID opId2 = new OperatorID(3, 4); + OperatorState op2 = new OperatorState("sink", null, opId2, 1, 128); + + Path savepointDir = Files.createDirectories(tempDir.resolve("savepoint-test")); + writeMetadata(savepointDir, 42L, Arrays.asList(op1, op2)); + + TableEnvironment tableEnv = newTableEnv(); + createCatalog(tableEnv, "state", directoryOption("app", tempDir)); + tableEnv.executeSql("USE CATALOG state"); + + StateCatalog catalog = getCatalog(tableEnv, "state"); + String dbName = catalog.listDatabases().get(0); + tableEnv.executeSql("USE `" + dbName + "`"); + + List<Row> rows = collectWithSql(tableEnv, "SELECT * FROM metadata"); + + assertThat(rows).hasSize(2); + rows.forEach(row -> assertThat(row.getField("checkpoint-id")).isEqualTo(42L)); + assertThat(rows.stream().map(r -> r.getField("operator-name")).collect(Collectors.toList())) + .containsExactlyInAnyOrder("source", "sink"); + + catalog.close(); + } + + @Test + void testMultipleLabelsDiscovered(@TempDir Path tempDir) throws Exception { + Path checkpointsDir = Files.createDirectories(tempDir.resolve("checkpoints")); + Path savepointsDir = Files.createDirectories(tempDir.resolve("savepoints")); + touchMetadata(checkpointsDir.resolve("savepoint-a")); + touchMetadata(savepointsDir.resolve("savepoint-b")); + + String directoryOptions = + directoryOption("ckpts", checkpointsDir) + + ", " + + directoryOption("svpts", savepointsDir); + + TableEnvironment tableEnv = newTableEnv(); + createCatalog(tableEnv, "with_ts", directoryOptions); + createCatalog( + tableEnv, "without_ts", directoryOptions + ", 'db-name.include-ts' = 'false'"); + + StateCatalog withTs = getCatalog(tableEnv, "with_ts"); + assertThat(withTs.listDatabases()) + .hasSize(2) + .allSatisfy( + dbName -> + assertThat(dbName) + .matches( + "(ckpts|svpts)/\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z/savepoint-[ab]")); + withTs.close(); + + StateCatalog withoutTs = getCatalog(tableEnv, "without_ts"); + assertThat(withoutTs.listDatabases()) + .containsExactlyInAnyOrder("ckpts/savepoint-a", "svpts/savepoint-b"); + withoutTs.close(); + } + + @Test + void testCatalogOperations(@TempDir Path tempDir) throws Exception { + Path savepointDir = Files.createDirectories(tempDir.resolve("savepoint-abc")); + writeMetadata(savepointDir, 7L, Collections.emptyList()); + + TableEnvironment tableEnv = newTableEnv(); + createCatalog(tableEnv, "state", directoryOption("app", tempDir)); + + StateCatalog catalog = getCatalog(tableEnv, "state"); + List<String> dbs = catalog.listDatabases(); + assertThat(dbs).hasSize(1); + String dbName = dbs.get(0); + assertThat(dbName).matches("app/\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z/savepoint-abc"); + + assertThat(catalog.databaseExists(dbName)).isTrue(); + assertThat(catalog.databaseExists("app/nonexistent")).isFalse(); + + assertThat(catalog.listTables(dbName)).isEmpty(); + assertThat(catalog.listViews(dbName)).containsExactly(StateCatalog.METADATA_TABLE); + + assertThat(catalog.tableExists(new ObjectPath(dbName, StateCatalog.METADATA_TABLE))) + .isTrue(); + assertThat(catalog.tableExists(new ObjectPath(dbName, "other"))).isFalse(); + + CatalogView view = + (CatalogView) catalog.getTable(new ObjectPath(dbName, StateCatalog.METADATA_TABLE)); + assertThat(view.getOriginalQuery()) + .contains("savepoint_metadata") + .contains(savepointDir.toAbsolutePath().toString()); + + // Verify querying the metadata view via SQL works and returns expected rows + tableEnv.executeSql("USE CATALOG state"); + tableEnv.executeSql("USE `" + dbName + "`"); + assertThat(collectWithSql(tableEnv, "SELECT * FROM metadata")).isEmpty(); + + catalog.close(); + } + + // ------------------------------------------------------------------------- + // Helpers + // ------------------------------------------------------------------------- + + private static TableEnvironment newTableEnv() { + TableEnvironment env = TableEnvironment.create(EnvironmentSettings.inBatchMode()); + env.loadModule("state", StateModule.INSTANCE); + return env; + } + + private static void createCatalog( + TableEnvironment tableEnv, String catalogName, String withOptions) { + tableEnv.executeSql( + String.format( + "CREATE CATALOG %s WITH ('type' = '%s', %s)", + catalogName, StateCatalogFactory.IDENTIFIER, withOptions)); + } + + private static String directoryOption(String label, Path dir) { + return String.format( + "'directory.%s' = '%s'", label, dir.toAbsolutePath().toString().replace("'", "''")); + } + + private static StateCatalog getCatalog(TableEnvironment tableEnv, String name) { + return (StateCatalog) tableEnv.getCatalog(name).get(); + } + + private static void touchMetadata(Path snapshotDir) throws Exception { + Files.createDirectories(snapshotDir); + Files.createFile(snapshotDir.resolve("_metadata")); + } + + private static void writeMetadata( + Path snapshotDir, long checkpointId, List<OperatorState> operators) throws Exception { + CheckpointMetadata metadata = + new CheckpointMetadata(checkpointId, operators, Collections.emptyList()); + try (OutputStream out = Files.newOutputStream(snapshotDir.resolve("_metadata"))) { + Checkpoints.storeCheckpointMetadata(metadata, out); + } + } + + private static List<Row> collectWithSql(TableEnvironment tEnv, String sql) throws Exception { + List<Row> rows = new ArrayList<>(); + TableResult result = tEnv.executeSql(sql); + try (CloseableIterator<Row> it = result.collect()) { + it.forEachRemaining(rows::add); + } + return rows; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/StateCatalogTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/StateCatalogTest.java new file mode 100644 index 00000000000..44e8ab8264f --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/StateCatalogTest.java @@ -0,0 +1,143 @@ +/* + * 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.flink.state.catalog; + +import org.apache.flink.table.catalog.CatalogDatabaseImpl; +import org.apache.flink.table.catalog.CatalogView; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.time.Instant; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Unit and functional tests for {@link StateCatalog} that are specific to the catalog layer + * (CatalogView/CatalogTable semantics, unsupported write operations). Directory scanning, db-name + * derivation, and dynamic re-discovery are {@link StateCatalog}'s delegation to {@link + * SnapshotDiscovery} and are covered exhaustively by {@link SnapshotDiscoveryTest} instead of being + * re-verified here. + */ +class StateCatalogTest { + + // Every metadata file created by createMetadataFile() gets this exact modification time, so + // the creationTs segment in derived database names is deterministic across all tests. + private static final Instant FIXED_TS = Instant.parse("2024-03-15T10:30:45Z"); + private static final String TS = "2024-03-15T10:30:45Z"; + private static final String METADATA_TABLE = StateCatalog.METADATA_TABLE; + + @TempDir Path tempDir; + + @Test + void testCatalogOperations() throws Exception { + createMetadataFile(tempDir.resolve("savepoint-abc")); + StateCatalog catalog = openCatalog("app1", tempDir); + String dbName = "app1/" + TS + "/savepoint-abc"; + + // databaseExists + assertThat(catalog.databaseExists(dbName)).isTrue(); + assertThat(catalog.databaseExists("app1/" + TS + "/savepoint-nonexistent")).isFalse(); + assertThat(catalog.databaseExists("unknown/" + TS + "/savepoint-abc")).isFalse(); + + // tableExists + assertThat(catalog.tableExists(new ObjectPath(dbName, METADATA_TABLE))).isTrue(); + assertThat(catalog.tableExists(new ObjectPath(dbName, "nonexistent"))).isFalse(); + assertThat( + catalog.tableExists( + new ObjectPath("app1/" + TS + "/nonexistent", METADATA_TABLE))) + .isFalse(); + + // getTable returns CatalogView with correct query + Path savepointDir = tempDir.resolve("savepoint-abc"); + CatalogView view = (CatalogView) catalog.getTable(new ObjectPath(dbName, METADATA_TABLE)); + assertThat(view.getOriginalQuery()) + .contains("savepoint_metadata") + .contains(savepointDir.toAbsolutePath().toString()); + + // getDatabase throws for unknown + assertThatThrownBy(() -> catalog.getDatabase("app1/" + TS + "/savepoint-nonexistent")) + .isInstanceOf(DatabaseNotExistException.class); + + // getTable throws for unknown snapshot + assertThatThrownBy( + () -> + catalog.getTable( + new ObjectPath( + "app1/" + TS + "/savepoint-nonexistent", + METADATA_TABLE))) + .isInstanceOf(TableNotExistException.class); + + catalog.close(); + } + + @Test + void testListFunctionsAlwaysReturnsEmpty() throws Exception { + StateCatalog catalog = openCatalog("app1", tempDir); + assertThat(catalog.listFunctions("nonexistent")).isEmpty(); + catalog.close(); + } + + @Test + void testWriteOperationsThrow() throws Exception { + StateCatalog catalog = openCatalog("app1", tempDir); + + assertThatThrownBy( + () -> + catalog.createDatabase( + "db", + new CatalogDatabaseImpl(Collections.emptyMap(), ""), + false)) + .isInstanceOf(UnsupportedOperationException.class); + assertThatThrownBy(() -> catalog.dropDatabase("db", true, false)) + .isInstanceOf(UnsupportedOperationException.class); + assertThatThrownBy(() -> catalog.createTable(new ObjectPath("db", "t"), null, false)) + .isInstanceOf(UnsupportedOperationException.class); + + catalog.close(); + } + + // ------------------------------------------------------------------------- + // Helpers + // ------------------------------------------------------------------------- + + private static void createMetadataFile(Path snapshotDir) throws IOException { + Files.createDirectories(snapshotDir); + Path file = Files.createFile(snapshotDir.resolve("_metadata")); + Files.setLastModifiedTime(file, FileTime.from(FIXED_TS)); + } + + private static StateCatalog openCatalog(String label, Path directory) throws Exception { + StateCatalog catalog = + new StateCatalog( + "state", + Collections.singletonMap(label, directory.toAbsolutePath().toString())); + catalog.open(); + return catalog; + } +}
