B-Prod-P6: ForSt-RS keyed-state backend wires disaggregated remote storage

flink-state-backends/flink-statebackend-forst-rs/src/main/java/org/apache/flink/state/forstrs/ForStRsOptions.java

@@ -44,7 +44,57 @@ public final class ForStRsOptions {
                                     + " 'per-state': each state name gets its own CF (lazy"
                                     + " creation, soft limit 256 CFs).");
 
+    // ----- B-Prod-P6: disaggregated remote storage -----
+
+    /** OpenDAL URI for the remote storage backend (e.g. {@code s3://bucket/}). */
+    public static final ConfigOption<String> STORAGE_URI =
+            ConfigOptions.key("state.backend.forst-rs.storage.uri")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "If set, the keyed-state backend opens the engine via OpenDAL on this"
+                                    + " URI (memory://, file:///abs/path, or s3://bucket/). When"
+                                    + " unset, the backend falls back to a local-FS engine at the"
+                                    + " usual data directory.");
+
+    /**
+     * Flat JSON object of OpenDAL backend-specific config (e.g. {@code
+     * {"region":"us-east-1","endpoint":"..."}}).
+     */
+    public static final ConfigOption<String> OPENDAL_CONFIG =
+            ConfigOptions.key("state.backend.forst-rs.storage.opendal-config")
+                    .stringType()
+                    .defaultValue("{}")
+                    .withDescription(
+                            "Flat JSON object holding OpenDAL backend-specific configuration."
+                                    + " Keys must match the underlying service builder field names"
+                                    + " (e.g. region, endpoint, access_key_id, secret_access_key for"
+                                    + " s3). Ignored for memory:// and file:// URIs.");
+
+    /** Local directory used for the SST LRU cache when {@link #STORAGE_URI} is set. */
+    public static final ConfigOption<String> CACHE_DIR =
+            ConfigOptions.key("state.backend.forst-rs.storage.cache-dir")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Local directory used by the SST LRU cache that fronts remote storage."
+                                    + " Required when storage.uri is set.");
+
+    /** Total LRU cache budget on local disk, in MiB. */
+    public static final ConfigOption<Long> CACHE_CAPACITY_MB =
+            ConfigOptions.key("state.backend.forst-rs.storage.cache-capacity-mb")
+                    .longType()
+                    .defaultValue(1024L)
+                    .withDescription(
+                            "Total bytes (MiB) the SST LRU cache may occupy on local disk."
+                                    + " Default 1 GiB. A value of 0 disables the cache (every read"
+                                    + " hits the remote backend).");
+
     private CfMode cfMode = CfMode.SINGLE;
+    private String storageUri;
+    private String opendalConfigJson = "{}";
+    private String cacheDir;
+    private long cacheCapacityMb = 1024L;
 
     public ForStRsOptions() {}
 
@@ -57,6 +107,54 @@ public ForStRsOptions cfMode(CfMode m) {
         return this;
     }
 
+    /** OpenDAL URI for remote storage; {@code null} means use the local data directory. */
+    public String storageUri() {
+        return storageUri;
+    }
+
+    public ForStRsOptions storageUri(String uri) {
+        this.storageUri = uri;
+        return this;
+    }
+
+    /** Flat JSON object of OpenDAL config; never null (defaults to {@code "{}"}). */
+    public String opendalConfigJson() {
+        return opendalConfigJson;
+    }
+
+    public ForStRsOptions opendalConfigJson(String json) {
+        this.opendalConfigJson = json == null || json.isEmpty() ? "{}" : json;
+        return this;
+    }
+
+    /** Local cache directory (must be set when {@link #storageUri()} is non-null). */
+    public String cacheDir() {
+        return cacheDir;
+    }
+
+    public ForStRsOptions cacheDir(String dir) {
+        this.cacheDir = dir;
+        return this;
+    }
+
+    /** LRU cache capacity in MiB; defaults to 1 GiB. */
+    public long cacheCapacityMb() {
+        return cacheCapacityMb;
+    }
+
+    public ForStRsOptions cacheCapacityMb(long mb) {
+        if (mb < 0) {
+            throw new IllegalArgumentException("cacheCapacityMb must be >= 0, got " + mb);
+        }
+        this.cacheCapacityMb = mb;
+        return this;
+    }
+
+    /** Convenience: cache capacity converted to bytes for the FFI call. */
+    public long cacheCapacityBytes() {
+        return cacheCapacityMb * 1024L * 1024L;
+    }
+
     /** Column-family routing mode. */
     public enum CfMode {
         SINGLE("single"),

flink-state-backends/flink-statebackend-forst-rs/src/main/java/org/apache/flink/state/forstrs/ForStRsStateBackend.java

@@ -38,9 +38,9 @@
  * <ul>
  *   <li>{@link #createKeyedStateBackend(KeyedStateBackendParameters)} — currently throws because
  *       the simpler {@link ForStRsKeyedStateBackend} stepping-stone does <i>not</i> implement
- *       {@link CheckpointableKeyedStateBackend} (no snapshot / key-group / savepoint plumbing
- *       yet). Use {@link #createBasicKeyedBackend(TypeSerializer)} for proof-of-concept and unit
- *       tests until the L5 sync-v1 / L6 rescaling work lands.
+ *       {@link CheckpointableKeyedStateBackend} (no snapshot / key-group / savepoint plumbing yet).
+ *       Use {@link #createBasicKeyedBackend(TypeSerializer)} for proof-of-concept and unit tests
+ *       until the L5 sync-v1 / L6 rescaling work lands.
  *   <li>{@link #createOperatorStateBackend(OperatorStateBackendParameters)} — delegates to Flink's
  *       {@link DefaultOperatorStateBackendBuilder}, which is the standard pattern for backends
  *       whose operator state is just a serialized bytestream rather than a KV store.
@@ -87,15 +87,16 @@ public OperatorStateBackend createOperatorStateBackend(
     }
 
     /**
-     * Phase-D L5 stepping-stone factory: opens an in-memory ForSt-RS engine and returns a
-     * {@link ForStRsKeyedStateBackend} bound to it. The returned backend owns the underlying
-     * {@link Arena}, {@link ForStRsLinker}, {@link FrsDb} and default {@link FrsCfHandle}; closing
-     * it releases all of them.
+     * Phase-D L5 stepping-stone factory: opens an in-memory ForSt-RS engine and returns a {@link
+     * ForStRsKeyedStateBackend} bound to it. The returned backend owns the underlying {@link
+     * Arena}, {@link ForStRsLinker}, {@link FrsDb} and default {@link FrsCfHandle}; closing it
+     * releases all of them.
      *
-     * <p>This entry-point is provided because {@link #createKeyedStateBackend(KeyedStateBackendParameters)}
-     * cannot yet return a {@link ForStRsKeyedStateBackend} — that class does not implement the
-     * {@link CheckpointableKeyedStateBackend} contract. Once snapshot / key-group / savepoint
-     * plumbing is wired in Phase-D L5/L6 the two factory methods will collapse into one.
+     * <p>This entry-point is provided because {@link
+     * #createKeyedStateBackend(KeyedStateBackendParameters)} cannot yet return a {@link
+     * ForStRsKeyedStateBackend} — that class does not implement the {@link
+     * CheckpointableKeyedStateBackend} contract. Once snapshot / key-group / savepoint plumbing is
+     * wired in Phase-D L5/L6 the two factory methods will collapse into one.
      */
     public <K> ForStRsKeyedStateBackend<K> createBasicKeyedBackend(
             TypeSerializer<K> keySerializer) {
@@ -116,5 +117,4 @@ public <K> ForStRsKeyedStateBackend<K> createBasicKeyedBackend(
             throw e;
         }
     }
-
 }

flink-state-backends/flink-statebackend-forst-rs/src/main/java/org/apache/flink/state/forstrs/ffm/ForStRsLinker.java

@@ -75,6 +75,7 @@ public final class ForStRsLinker {
     private final MethodHandle frsDbOpenMemory;
     private final MethodHandle frsDbOpenMemoryTuned;
     private final MethodHandle frsDbOpenFromCheckpoint;
+    private final MethodHandle frsDbOpenRemote;
     private final MethodHandle frsDbClose;
 
     // --- 2. CF management ---
@@ -179,6 +180,20 @@ public ForStRsLinker(Arena arena) {
                                 ValueLayout.ADDRESS, // target_dir (c_char*)
                                 ValueLayout.ADDRESS)); // out_handle
 
+        // B-Prod-P6: open the engine on top of an OpenDAL URI with a local
+        // LRU cache. JSON config is a flat string-to-string object; pass
+        // "{}" or NULL when the URI scheme needs no extra config.
+        this.frsDbOpenRemote =
+                bind(
+                        "frs_db_open_remote",
+                        FunctionDescriptor.of(
+                                ValueLayout.JAVA_INT,
+                                ValueLayout.ADDRESS, // uri (c_char*)
+                                ValueLayout.ADDRESS, // opendal_config_json (c_char*)
+                                ValueLayout.ADDRESS, // cache_dir (c_char*)
+                                ValueLayout.JAVA_LONG, // cache_capacity_bytes (u64)
+                                ValueLayout.ADDRESS)); // out_handle
+
         this.frsDbClose =
                 bind(
                         "frs_db_close",
@@ -582,6 +597,45 @@ public FrsDb dbOpen(Arena arena, String path) {
         return new FrsDb(this, handle);
     }
 
+    /**
+     * Opens a remote-storage-backed engine with a local LRU SST cache (B-Prod-P6).
+     *
+     * <p>{@code uri} is an OpenDAL URI such as {@code memory://}, {@code file:///abs/path}, or
+     * {@code s3://bucket/}. {@code opendalConfigJson} carries any scheme-specific knobs (e.g.
+     * {@code {"region":"us-east-1","endpoint":"https://minio.example.com"}}); pass {@code "{}"} or
+     * {@code null} when none are needed. {@code cacheDir} is the local directory used for the LRU
+     * SST cache; {@code cacheCapacityBytes} caps its on-disk footprint.
+     *
+     * <p>Caller closes via {@link FrsDb#close()}.
+     */
+    public FrsDb dbOpenRemote(
+            Arena arena,
+            String uri,
+            String opendalConfigJson,
+            String cacheDir,
+            long cacheCapacityBytes) {
+        MemorySegment uriSeg = allocateCString(arena, uri);
+        MemorySegment cfgSeg =
+                opendalConfigJson == null
+                        ? MemorySegment.NULL
+                        : allocateCString(arena, opendalConfigJson);
+        MemorySegment cacheDirSeg = allocateCString(arena, cacheDir);
+        MemorySegment outHandle = arena.allocate(ValueLayout.ADDRESS);
+        int rc;
+        try {
+            rc =
+                    (int)
+                            frsDbOpenRemote.invokeExact(
+                                    uriSeg, cfgSeg, cacheDirSeg, cacheCapacityBytes, outHandle);
+        } catch (Throwable t) {
+            throw new FrsBackendException(
+                    FrsStatus.PANIC, "frs_db_open_remote threw: " + t.getMessage());
+        }
+        check(rc, "frs_db_open_remote");
+        MemorySegment handle = outHandle.get(ValueLayout.ADDRESS, 0);
+        return new FrsDb(this, handle);
+    }
+
     /**
      * Opens an engine by restoring its state from a checkpoint directory previously produced by
      * {@link #createCheckpoint(FrsDb, String)}. The checkpoint directory must contain {@code
@@ -1185,8 +1239,7 @@ public void setCompactionFilterTtl(
                                     db.handle(), cf.handle(), ttlMs, stateType, timestampOffset);
         } catch (Throwable t) {
             throw new FrsBackendException(
-                    FrsStatus.PANIC,
-                    "frs_cf_set_compaction_filter_ttl threw: " + t.getMessage());
+                    FrsStatus.PANIC, "frs_cf_set_compaction_filter_ttl threw: " + t.getMessage());
         }
         check(rc, "frs_cf_set_compaction_filter_ttl");
     }
@@ -1289,9 +1342,9 @@ public byte[] getAt(FrsDb db, FrsCfHandle cf, FrsSnapshot snapshot, byte[] key)
     }
 
     /**
-     * Opens a forward iterator that yields the latest version of each user-key with {@code seq &lt;=
-     * snapshot.seq}. Drive it with {@link #iteratorNext(FrsIterator)} and release with
-     * {@link FrsIterator#close()} (uses the standard non-prefix close path).
+     * Opens a forward iterator that yields the latest version of each user-key with {@code seq
+     * &lt;= snapshot.seq}. Drive it with {@link #iteratorNext(FrsIterator)} and release with {@link
+     * FrsIterator#close()} (uses the standard non-prefix close path).
      */
     public FrsIterator iteratorOpenAt(FrsDb db, FrsCfHandle cf, FrsSnapshot snapshot, Arena arena) {
         MemorySegment outIter = arena.allocate(ValueLayout.ADDRESS);

flink-state-backends/flink-statebackend-forst-rs/src/main/java/org/apache/flink/state/forstrs/ffm/FrsSnapshot.java

@@ -70,8 +70,8 @@ public boolean isClosed() {
 
     /**
      * Releases the snapshot back to the engine. Idempotent: calling more than once is a no-op (does
-     * not double-release). Per spec §10.0 the underlying release call cannot fail with
-     * {@code INVALID_ARGUMENT} as long as the snapshot was originally obtained from {@code db}.
+     * not double-release). Per spec §10.0 the underlying release call cannot fail with {@code
+     * INVALID_ARGUMENT} as long as the snapshot was originally obtained from {@code db}.
      */
     @Override
     public void close() {

flink-state-backends/flink-statebackend-forst-rs/src/main/java/org/apache/flink/state/forstrs/keyed/ForStRsAbstractKeyedStateBackend.java

@@ -33,15 +33,14 @@
 import org.apache.flink.runtime.state.InternalKeyContextImpl;
 import org.apache.flink.runtime.state.KeyGroupRange;
 import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
-import org.apache.flink.runtime.state.KeyedStateHandle;
 import org.apache.flink.runtime.state.Keyed;
+import org.apache.flink.runtime.state.KeyedStateHandle;
 import org.apache.flink.runtime.state.PriorityComparable;
 import org.apache.flink.runtime.state.SavepointResources;
 import org.apache.flink.runtime.state.SnapshotResult;
 import org.apache.flink.runtime.state.SnapshotStrategy;
 import org.apache.flink.runtime.state.StateHandleID;
 import org.apache.flink.runtime.state.StateSnapshotTransformer.StateSnapshotTransformFactory;
-import org.apache.flink.runtime.state.StreamCompressionDecorator;
 import org.apache.flink.runtime.state.UncompressedStreamCompressionDecorator;
 import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
 import org.apache.flink.runtime.state.metrics.LatencyTrackingStateConfig;
@@ -59,17 +58,17 @@
  * Spec §4 skeleton: a {@link AbstractKeyedStateBackend} subclass that wires the ForSt-RS engine
  * into Flink's keyed-state SPI. The constructor satisfies Flink 2.2.0's {@code
  * AbstractKeyedStateBackend} ctor (10 args) and the abstract methods are implemented as
- * "implemented in P3/P4" stubs throwing {@link UnsupportedOperationException}; the inner
- * round-trip primitives still live on {@link ForStRsKeyedStateBackend} (which this skeleton
- * delegates to once full snapshot/restore wiring lands).
+ * "implemented in P3/P4" stubs throwing {@link UnsupportedOperationException}; the inner round-trip
+ * primitives still live on {@link ForStRsKeyedStateBackend} (which this skeleton delegates to once
+ * full snapshot/restore wiring lands).
  *
  * <p><b>Why a separate class.</b> The existing {@link ForStRsKeyedStateBackend} (Phase-D L5) is a
- * standalone {@code Closeable} consumed by a wide test surface that would not survive switching
- * its parent class today (e.g., {@link ForStRsKeyedStateBackend#setCurrentKey} returns void with
- * the byte-prefix invalidation policy that cleanly works only when the class isn't already
- * inheriting key-context plumbing from {@code AbstractKeyedStateBackend}). Per the plan, this
- * skeleton lands now to "match what Flink's keyed-state SPI registries expect"; the L5 class will
- * be folded into this one in P3/P4 once snapshot/restore + key-group iteration are wired.
+ * standalone {@code Closeable} consumed by a wide test surface that would not survive switching its
+ * parent class today (e.g., {@link ForStRsKeyedStateBackend#setCurrentKey} returns void with the
+ * byte-prefix invalidation policy that cleanly works only when the class isn't already inheriting
+ * key-context plumbing from {@code AbstractKeyedStateBackend}). Per the plan, this skeleton lands
+ * now to "match what Flink's keyed-state SPI registries expect"; the L5 class will be folded into
+ * this one in P3/P4 once snapshot/restore + key-group iteration are wired.
  *
  * @param <K> key type
  */
@@ -80,16 +79,16 @@ public class ForStRsAbstractKeyedStateBackend<K> extends AbstractKeyedStateBacke
     private final ForStRsKeyedStateBackend<K> delegate;
 
     /**
-     * The snapshot strategy that drives incremental checkpoints (B-Prod-P3). Set lazily by
-     * {@link #setSnapshotStrategy(ForStRsSnapshotStrategy)} once the keyed-backend builder has the
-     * KGR + UUID + cfMap to construct it.
+     * The snapshot strategy that drives incremental checkpoints (B-Prod-P3). Set lazily by {@link
+     * #setSnapshotStrategy(ForStRsSnapshotStrategy)} once the keyed-backend builder has the KGR +
+     * UUID + cfMap to construct it.
      */
     private ForStRsSnapshotStrategy snapshotStrategy;
 
     /**
-     * The SST registry shared between this backend's snapshot strategy and the
-     * {@code notifyCheckpointComplete}/{@code notifyCheckpointAborted} hooks. Optional — only
-     * required once snapshot wiring is connected.
+     * The SST registry shared between this backend's snapshot strategy and the {@code
+     * notifyCheckpointComplete}/{@code notifyCheckpointAborted} hooks. Optional — only required
+     * once snapshot wiring is connected.
      */
     private ForStRsSstRegistry sstRegistry;
 
@@ -251,12 +250,13 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
     }
 
     /**
-     * Override of {@link org.apache.flink.api.common.state.CheckpointListener#notifyCheckpointAborted(long)}.
+     * Override of {@link
+     * org.apache.flink.api.common.state.CheckpointListener#notifyCheckpointAborted(long)}.
      *
      * <p>For an aborted checkpoint we must roll back the registry's ref-count bumps so the aborted
-     * checkpoint's "newly-uploaded" SSTs can drop to zero ref (eligible for discard) — the
-     * baseline shared with previously completed checkpoints is preserved because the registry's
-     * ref-counts are independent per checkpoint contribution.
+     * checkpoint's "newly-uploaded" SSTs can drop to zero ref (eligible for discard) — the baseline
+     * shared with previously completed checkpoints is preserved because the registry's ref-counts
+     * are independent per checkpoint contribution.
      */
     @Override
     public void notifyCheckpointAborted(long checkpointId) throws Exception {

flink-state-backends/flink-statebackend-forst-rs/src/main/java/org/apache/flink/state/forstrs/keyed/ForStRsCheckpointRestoreException.java

@@ -25,12 +25,12 @@
 /**
  * Module-internal "checkpoint restore failed" exception (B-Prod-P4 Task 4.2).
  *
- * <p>Flink's runtime does not currently ship a public {@code CheckpointRestoreException} —
- * the module-internal {@link org.apache.flink.runtime.state.BackendBuildingException} is the
- * closest public type, but it does not carry the failed-path / checkpoint-id pair we want for
- * strict-restore diagnostics. We therefore expose this dedicated subclass of {@link IOException}
- * so a missing SST surfaces with both fields and so callers can pattern-match without leaking
- * a runtime-internal type.
+ * <p>Flink's runtime does not currently ship a public {@code CheckpointRestoreException} — the
+ * module-internal {@link org.apache.flink.runtime.state.BackendBuildingException} is the closest
+ * public type, but it does not carry the failed-path / checkpoint-id pair we want for
+ * strict-restore diagnostics. We therefore expose this dedicated subclass of {@link IOException} so
+ * a missing SST surfaces with both fields and so callers can pattern-match without leaking a
+ * runtime-internal type.
  *
  * <p>Used by {@link ForStRsRestoreOperation} when a downloaded checkpoint is missing one or more
  * SST files referenced by its manifest, or when manifest parsing itself fails.
@@ -46,7 +46,8 @@ public class ForStRsCheckpointRestoreException extends IOException {
     /** Checkpoint id the restore was operating on. */
     private final long checkpointId;
 
-    public ForStRsCheckpointRestoreException(String missingPath, long checkpointId, String message) {
+    public ForStRsCheckpointRestoreException(
+            String missingPath, long checkpointId, String message) {
         super(message);
         this.missingPath = missingPath;
         this.checkpointId = checkpointId;