Add REFRESH DDL command

deploy/samples/logicaldb.yaml

@@ -23,3 +23,17 @@ spec:
   url: jdbc:logical://nearline=ads-database;offline=ads-catalog-database
   schema: LOGICAL-OFFLINE
   dialect: Calcite
+
+---
+
+# Reverse-ETL logical table: offline (demo) -> online (demo). Its implicit offline-tier
+# trigger writes to the ONLINE tier, which is the real physical table ADS.<name>. Used to
+# exercise `REFRESH <physical table>` (backfill the tier a consumer reads).
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: Database
+metadata:
+  name: logical-retl
+spec:
+  url: jdbc:logical://offline=ads-catalog-database;online=ads-database
+  schema: LOGICAL-RETL
+  dialect: Calcite

docs/user-guide/ddl-reference.md

@@ -253,6 +253,46 @@ Its lifecycle belongs to the Kubernetes Job controller: retries via the template
 its own inspectable record (a `Failed` Job labelled `backfill=true`). Re-issue the
 `FIRE` to launch a fresh one.
 
+## REFRESH
+
+```
+REFRESH <table>
+  [ FROM <bound> TO <bound> ]
+```
+
+Backfills a **physical table** by firing the trigger(s) that **produce** it. It is
+the table-level counterpart to `FIRE TRIGGER`: rather than naming a trigger, you
+name the table you want refreshed, and Hoptimator finds and fires whatever writes
+to it. The optional `FROM … TO …` window behaves exactly as it does for
+`FIRE TRIGGER` (same bounds, same one-off backfill semantics — see above).
+
+```sql
+-- fire the trigger that produces this table ("run now")
+REFRESH "ADS"."MEMBERS";
+
+-- backfill it over a fixed historical window
+REFRESH "ADS"."MEMBERS" FROM '2026-05-01' TO '2026-05-08';
+```
+
+REFRESH targets a **physical** table on purpose. Hoptimator doesn't really
+distinguish logical from physical — a logical table is just a physical table with
+extra moving parts (tiers, an inter-tier pipeline, a trigger) — and a consumer
+always reads a *specific* physical table (tier). So "refresh the tier I read" is
+unambiguous, whereas "refresh the logical table" isn't (which tier's data do you
+want fresh?). Refreshing a **logical table** by name is therefore rejected with a
+hint to refresh a specific physical tier instead. In practice a physical table has
+zero or one producing trigger, so there's no fan-out.
+
+A `REFRESH` errors when the table doesn't exist, or when **nothing produces it** (no
+trigger writes to it) — a refresh that silently does nothing is a footgun, not a
+no-op.
+
+How the producing trigger is discovered is a backend concern. Discovery reuses the
+pipeline dependency **graph** (`GraphService` / the `GraphProvider` SPI): the table
+identifier is resolved via the Calcite schema, and the one-hop graph around it
+exposes its producing triggers as `trigger → table` edges. `hoptimator-jdbc` never
+touches Kubernetes directly.
+
 ## CREATE TABLE
 
 ```
@@ -341,11 +381,9 @@ the parse alone.
 
 **Parses but not yet executed:**
 
-- `REFRESH MATERIALIZED VIEW <name>` — intended to re-run a batch-style
-  materialization on demand.
 - `FIRE TABLE | VIEW | MATERIALIZED VIEW <name>` — intended to
   manually fire a side effect (e.g. for testing without waiting for a
-  schedule). (`FIRE TRIGGER` is fully implemented — see above.)
+  schedule). (`FIRE TRIGGER` and `REFRESH` are fully implemented — see above.)
 - `PAUSE MATERIALIZED VIEW <name>` / `RESUME MATERIALIZED VIEW <name>` —
   parser support exists; executor does not. (`PAUSE TRIGGER` /
   `RESUME TRIGGER` above are fully implemented.)

hoptimator-jdbc/src/main/codegen/config.fmpp

@@ -48,7 +48,7 @@ data: {
       "com.linkedin.hoptimator.jdbc.ddl.SqlResumeMaterializedView"
       "com.linkedin.hoptimator.jdbc.ddl.SqlResumeTrigger"
       "com.linkedin.hoptimator.jdbc.ddl.SqlRefresh"
-      "com.linkedin.hoptimator.jdbc.ddl.SqlRefreshMaterializedView"
+      "com.linkedin.hoptimator.jdbc.ddl.SqlRefreshObject"
     ]
 
     # List of new keywords. Example: "DATABASES", "TABLES". If the keyword is

hoptimator-jdbc/src/main/codegen/includes/parserImpls.ftl

@@ -428,26 +428,16 @@ SqlDrop SqlDropFunction(Span s, boolean replace) :
 SqlRefresh SqlRefresh() :
 {
     final Span s;
-    final SqlRefresh refresh;
-}
-{
-    <REFRESH> { s = span(); }
-    (
-        refresh = SqlRefreshMaterializedView(s)
-    )
-    {
-        return refresh;
-    }
-}
-
-SqlRefresh SqlRefreshMaterializedView(Span s) :
-{
     final SqlIdentifier id;
+    SqlNode from = null;
+    SqlNode to = null;
 }
 {
-    <MATERIALIZED> <VIEW> id = CompoundIdentifier()
+    <REFRESH> { s = span(); }
+    id = CompoundIdentifier()
+    [ <FROM> from = FireBound() <TO> to = FireBound() ]
     {
-        return new SqlRefreshMaterializedView(s.end(this), id);
+        return new SqlRefreshObject(s.end(this), id, from, to);
     }
 }
 

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/GraphService.java

@@ -42,7 +42,21 @@ public static PipelineGraph buildGraph(String identifier, int depth, HoptimatorC
     if (depth < 0) {
       throw new SQLException("depth must be non-negative; got: " + depth);
     }
-    GraphTarget target = resolve(identifier, connection);
+    return buildGraph(resolve(identifier, connection), depth, connection);
+  }
+
+  /**
+   * Build a {@link PipelineGraph} for an already-resolved {@link GraphTarget}. Lets callers that
+   * already hold a target (e.g. after {@link #resolve}) build a subgraph without round-tripping
+   * through a string identifier and the schema resolver again.
+   *
+   * @throws SQLException if depth is negative, no provider supports the target, or the provider throws.
+   */
+  public static PipelineGraph buildGraph(GraphTarget target, int depth, HoptimatorConnection connection)
+      throws SQLException {
+    if (depth < 0) {
+      throw new SQLException("depth must be non-negative; got: " + depth);
+    }
     for (GraphProvider provider : providers()) {
       if (provider.supports(target)) {
         return provider.forTarget(target, depth, connection);

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlExecutor.java

@@ -33,6 +33,7 @@
 import com.linkedin.hoptimator.jdbc.ddl.SqlDropTrigger;
 import com.linkedin.hoptimator.jdbc.ddl.SqlFireTrigger;
 import com.linkedin.hoptimator.jdbc.ddl.SqlPauseTrigger;
+import com.linkedin.hoptimator.jdbc.ddl.SqlRefreshObject;
 import com.linkedin.hoptimator.jdbc.ddl.SqlResumeTrigger;
 import com.linkedin.hoptimator.util.DeploymentService;
 import com.linkedin.hoptimator.util.planner.HoptimatorJdbcSchema;
@@ -338,28 +339,74 @@ public void execute(SqlFireTrigger fire, CalcitePrepare.Context context) {
 
     // FIRE carries no user options — it is a pure "run now" action. A windowed fire resolves its
     // bounds to absolute instants here; a plain fire has neither. Nothing touches the spec.
-    Trigger.Fire fireRequest;
-    if (fire.from != null) {
-      fireRequest = new Trigger.Fire(
-          resolveFireBound(((SqlLiteral) fire.from).getValueAs(String.class), fire),
-          resolveFireBound(((SqlLiteral) fire.to).getValueAs(String.class), fire));
-    } else {
-      fireRequest = new Trigger.Fire(null, null);
+    Trigger.Fire fireRequest = buildFireRequest(fire.from, fire.to, fire);
+    fireTriggerByName(name, fireRequest, fire);
+    logger.info("FIRE TRIGGER {} completed", name);
+  }
+
+  /** Builds a {@link Trigger.Fire} from the optional {@code FROM ... TO ...} bounds shared by
+   *  {@code FIRE TRIGGER} and {@code REFRESH}. A windowed fire resolves both bounds to absolute
+   *  instants; a plain fire (no window) has neither. */
+  private Trigger.Fire buildFireRequest(SqlNode from, SqlNode to, SqlNode node) {
+    if (from != null) {
+      return new Trigger.Fire(
+          resolveFireBound(((SqlLiteral) from).getValueAs(String.class), node),
+          resolveFireBound(((SqlLiteral) to).getValueAs(String.class), node));
     }
-    Trigger trigger = new Trigger(name, null, null, new HashMap<>(), null, null, fireRequest);
+    return new Trigger.Fire(null, null);
+  }
 
+  /** Fires a single trigger by name, reusing the {@code FIRE TRIGGER} deploy path: build a
+   *  name-only {@link Trigger} carrying the {@link Trigger.Fire} intent and update it. Restores the
+   *  deployers and rethrows as a {@link DdlException} on any failure. */
+  private void fireTriggerByName(String name, Trigger.Fire fireRequest, SqlNode node) {
+    Trigger trigger = new Trigger(name, null, null, new HashMap<>(), null, null, fireRequest);
     Collection<Deployer> deployers = null;
     try {
       logger.info("Firing trigger {}", name);
       deployers = DeploymentService.deployers(trigger, connection);
       DeploymentService.update(deployers);
-      logger.info("FIRE TRIGGER {} completed", name);
     } catch (Exception e) {
       if (deployers != null) {
         DeploymentService.restore(deployers);
       }
-      throw new DdlException(fire, e.getMessage(), e);
+      throw new DdlException(node, e.getMessage(), e);
+    }
+  }
+
+  /** Executes a {@code REFRESH} command: a windowed backfill of a physical table.
+   *  REFRESH backfills a physical table by firing the trigger(s) that produce it, reusing the
+   *  {@code FIRE TRIGGER} machinery. Discovering those triggers is delegated to
+   *  {@link RefreshService} (via the dependency graph), so the DDL layer stays decoupled from the
+   *  backend. Errors when the table is unknown, is a logical table (refresh a tier instead), or has
+   *  no trigger producing it. */
+  public void execute(SqlRefreshObject refresh, CalcitePrepare.Context context) {
+    logger.info("Validating statement: {}", refresh);
+    try {
+      ValidationService.validateOrThrow(refresh, connection);
+    } catch (SQLException e) {
+      throw new DdlException(refresh, e.getMessage(), e);
+    }
+
+    String objectName = String.join(".", refresh.name.names);
+    List<String> triggers;
+    try {
+      triggers = RefreshService.producingTriggers(refresh.name.names, connection);
+    } catch (SQLException e) {
+      throw new DdlException(refresh, e.getMessage(), e);
+    }
+
+    // A REFRESH that fires nothing is a footgun — fail loudly instead of silently doing nothing.
+    if (triggers.isEmpty()) {
+      throw new DdlException(refresh, "Cannot REFRESH " + objectName
+          + ": no trigger produces it.");
+    }
+
+    Trigger.Fire fireRequest = buildFireRequest(refresh.from, refresh.to, refresh);
+    for (String triggerName : triggers) {
+      fireTriggerByName(triggerName, fireRequest, refresh);
     }
+    logger.info("REFRESH {} completed ({} trigger(s) fired)", objectName, triggers.size());
   }
 
   private static final Pattern FIRE_RELATIVE = Pattern.compile("^-(\\d+)([smhd])$");

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/RefreshService.java

@@ -0,0 +1,60 @@
+package com.linkedin.hoptimator.jdbc;
+
+import com.linkedin.hoptimator.graph.GraphEdge;
+import com.linkedin.hoptimator.graph.GraphNode;
+import com.linkedin.hoptimator.graph.GraphTarget;
+import com.linkedin.hoptimator.graph.PipelineGraph;
+
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+
+/**
+ * Resolves the trigger(s) a {@code REFRESH} should fire — the trigger(s) that <em>produce</em>
+ * the named physical table.
+ *
+ * <p>Hoptimator doesn't distinguish logical from physical tables, and a consumer always reads a
+ * specific physical table, so REFRESH targets a physical table and fires whatever writes to it.
+ * Discovery reuses the pipeline dependency graph: {@link GraphService#resolve} classifies the
+ * identifier via the Calcite schema, and the one-hop graph around it exposes the producing triggers
+ * as {@code trigger -> table} ({@link GraphEdge.Type#TRIGGERS}) edges. In practice a physical table
+ * has zero or one producing trigger. The DDL layer never touches Kubernetes — the graph is built by
+ * a pluggable {@code GraphProvider}.
+ */
+final class RefreshService {
+
+  private RefreshService() {
+  }
+
+  /**
+   * Returns the names of the triggers that produce {@code path} (usually zero or one). Throws when
+   * the identifier doesn't resolve to a table, or resolves to a logical table (which has no single
+   * physical output to refresh — a caller should refresh a specific tier instead).
+   */
+  static List<String> producingTriggers(List<String> path, HoptimatorConnection connection)
+      throws SQLException {
+    String identifier = String.join(".", path);
+    GraphTarget target = GraphService.resolve(identifier, connection);
+    if (target instanceof GraphTarget.LogicalTable) {
+      throw new SQLException(identifier + " is a logical table; REFRESH a specific physical table "
+          + "(tier) instead.");
+    }
+    PipelineGraph graph = GraphService.buildGraph(target, 1, connection);
+    return producingTriggers(graph);
+  }
+
+  /** Names of the triggers that produce the graph's root table — {@code trigger -> root}
+   *  ({@link GraphEdge.Type#TRIGGERS}) edges. Consumer edges ({@code root -> trigger}) are ignored. */
+  static List<String> producingTriggers(PipelineGraph graph) {
+    List<String> names = new ArrayList<>();
+    for (GraphEdge edge : graph.edges()) {
+      if (edge.type() == GraphEdge.Type.TRIGGERS
+          && edge.to().equals(graph.root())
+          && edge.from() instanceof GraphNode.Trigger) {
+        names.add(((GraphNode.Trigger) edge.from()).name());
+      }
+    }
+    return names;
+  }
+}