SPI foundation: replace java.sql.Connection with DeploymentContext

CONTRIBUTING.md

@@ -39,6 +39,11 @@ For bugs, please include:
    or any user-facing SQL behavior; running `make test` regenerates the
    diff. The `QuidemTestBase` class in `hoptimator-jdbc` is the shared
    harness.
+
+   Behavior that isn't SQL — e.g. the direct `TableService.create(...)`
+   API — is covered by plain JUnit integration tests tagged
+   `@Tag("integration")` (see the `*TableServiceIntegrationTest` classes),
+   which drive the new APIs directly instead of routing through Quidem.
 3. **Cover your changes.** New code needs tests; don't ship behavior that
    isn't exercised by either a unit or an integration test. Generate a
    coverage report with:

deploy/docker/venice/docker-compose-single-dc-setup.yaml

@@ -32,6 +32,14 @@ services:
     image: venicedb/venice-controller:0.4.858
     container_name: venice-controller
     hostname: venice-controller
+    # Local-only override: disable Venice's default 6h store-recreation cooldown
+    # (controller.store.recreation.after.deletion.time.window.seconds, default 21600) so
+    # integration tests that create and delete Venice stores are re-runnable without waiting.
+    # Appends the override to the image's baked-in controller.properties, then launches normally.
+    command:
+      - sh
+      - -c
+      - echo "controller.store.recreation.after.deletion.time.window.seconds=0" >> configs/single-dc/controller.properties && java -jar bin/venice-controller-all.jar configs/single-dc/cluster.properties configs/single-dc/controller.properties
     depends_on:
       kafka:
         condition: service_healthy

docs/extending/config-providers.md

@@ -30,17 +30,17 @@ don't need a custom provider.
 
 ```java
 public interface ConfigProvider {
-  Properties loadConfig(Connection connection) throws Exception;
+  Properties loadConfig(DeploymentContext context) throws Exception;
 }
 ```
 
 One method. Return a `Properties` populated however you like; throw if
 loading fails — Hoptimator will log it and continue with whatever the
 other providers produced.
 
-The `Connection` argument is the active Hoptimator JDBC connection.
-`HoptimatorConnection.connectionProperties()` and the bundled
-`K8sContext.create(connection)` are useful when you want to scope your
+The `DeploymentContext` argument is a Calcite-free handle onto the caller's
+environment. `context.properties()` and the bundled
+`K8sContext.create(context)` are useful when you want to scope your
 loading by namespace, kubeconfig, or anything else the caller already
 configured.
 
@@ -64,9 +64,8 @@ order.
 public class VaultConfigProvider implements ConfigProvider {
 
   @Override
-  public Properties loadConfig(Connection connection) throws Exception {
-    HoptimatorConnection conn = (HoptimatorConnection) connection;
-    String namespace = K8sContext.create(conn).namespace();
+  public Properties loadConfig(DeploymentContext context) throws Exception {
+    String namespace = K8sContext.create(context).namespace();
 
     // Pull namespace-scoped values from Vault. Whatever you return becomes
     // available as {{key}} in any template — and as a connection property

docs/extending/deployers.md

@@ -72,7 +72,7 @@ A `Deployer` doesn't get loaded directly. Instead, you ship a
 
 ```java
 public interface DeployerProvider {
-  <T extends Deployable> Collection<Deployer> deployers(T obj, Connection connection);
+  <T extends Deployable> Collection<Deployer> deployers(T obj, DeploymentContext context);
   int priority();
 }
 ```
@@ -82,6 +82,17 @@ provider returns the deployers that apply to it. Return an empty
 collection when the deployable isn't yours — the runtime will skip you and
 move on to the next provider.
 
+The `DeploymentContext` is a Calcite-free handle onto everything a deployer
+needs: connection-level `properties()` (namespace, hints, cluster config) and
+per-`Database` connection config via `databaseProperties(catalog, schema,
+urlPrefix)`. The SQL path supplies a Calcite-backed implementation
+(`CalciteDeploymentContext`); the direct table-create API supplies a
+`DirectDeploymentContext` that carries the caller's Avro schema. A deployer that
+needs the row type calls `HoptimatorDriver.rowType(source, context)`, which
+resolves it from the Calcite catalog (SQL path) or derives it from the carried
+Avro schema (direct path) — so a deployer never touches `java.sql.Connection` or
+a Calcite `SchemaPlus` directly.
+
 `priority()` controls ordering: providers with **lower priority numbers
 run first**. If two providers can both deploy the same object, the lower-
 priority one wins. The default Kubernetes provider is priority `2`; ship
@@ -103,9 +114,9 @@ com.example.hoptimator.mysystem.MySystemDeployerProvider
 The bundled Kafka path is a good shape to copy:
 
 - [`KafkaDeployerProvider`](https://github.com/linkedin/Hoptimator/blob/main/hoptimator-kafka/src/main/java/com/linkedin/hoptimator/kafka/KafkaDeployerProvider.java)
-  — type-checks the `Deployable`, extracts per-schema connection
-  properties from the Calcite schema (i.e. the JDBC URL the `Database`
-  CRD points at), and constructs the deployer.
+  — type-checks the `Deployable`, resolves the `Database`'s connection
+  config via `context.databaseProperties(catalog, schema, "jdbc:kafka://")`
+  (the JDBC URL the `Database` CRD points at), and constructs the deployer.
 - [`KafkaDeployer`](https://github.com/linkedin/Hoptimator/blob/main/hoptimator-kafka/src/main/java/com/linkedin/hoptimator/kafka/KafkaDeployer.java)
   — `create()` calls Kafka's AdminClient API to create the topic;
   `restore()` walks back and deletes any topic the current operation
@@ -115,7 +126,7 @@ The shape any provider should follow:
 
 - Type-check the `Deployable` and return an empty collection if it's not
   what you handle.
-- Extract any per-schema configuration from the connection.
+- Extract any per-schema configuration from the context.
 - Construct one or more deployer instances and return them.
 
 ## Validation
@@ -126,7 +137,7 @@ Deployers can opt into pre-deploy validation by also implementing
 ```java
 public class MyDeployer implements Deployer, Validated {
   @Override
-  public void validate(Validator.Issues issues) {
+  public void validate(Validator.Issues issues, DeploymentContext context) {
     // emit warnings or errors before any side effects
   }
 }
@@ -143,7 +154,7 @@ The validate-then-specify path is the main test surface:
 ```java
 Source source = new Source("my-database", List.of("MYSYS", "foo"), Map.of());
 DeployerProvider provider = new MyDeployerProvider();
-Collection<Deployer> deployers = provider.deployers(source, mockConnection);
+Collection<Deployer> deployers = provider.deployers(source, context);
 
 assertThat(deployers).hasSize(1);
 List<String> specs = deployers.iterator().next().specify();

docs/extending/validators.md

@@ -26,7 +26,7 @@ the operator's status field for CRD-driven changes).
 
 ```java
 public interface Validated {
-  void validate(Validator.Issues issues);
+  void validate(Validator.Issues issues, DeploymentContext context);
 }
 
 public interface Validator extends Validated {
@@ -35,10 +35,16 @@ public interface Validator extends Validated {
 }
 
 public interface ValidatorProvider {
-  <T> Collection<Validator> validators(T obj);
+  <T> Collection<Validator> validators(T obj, DeploymentContext context);
 }
 ```
 
+The `DeploymentContext` (Calcite-free) exposes connection-level
+`properties()` and per-`Database` config via `databaseProperties(...)` —
+everything a validator needs to run lookups against external systems without
+touching `java.sql.Connection`. A validator that needs the deployable's row
+type calls `HoptimatorDriver.rowType(source, context)`.
+
 Two ways to participate:
 
 - **Make your own type `Validated`.** If you've authored a new
@@ -58,7 +64,7 @@ nested contexts let the final error message read like a structured report.
 ```java
 public class NamingPolicyValidator implements Validator {
   @Override
-  public void validate(Validator.Issues issues) {
+  public void validate(Validator.Issues issues, DeploymentContext context) {
     if (tableName.contains("_")) {
       issues.error("Table names must not contain underscores");
     }
@@ -112,7 +118,7 @@ input:
 ```java
 public class MyPolicyValidatorProvider implements ValidatorProvider {
   @Override
-  public <T> Collection<Validator> validators(T obj) {
+  public <T> Collection<Validator> validators(T obj, DeploymentContext context) {
     if (obj instanceof Source) {
       return List.of(new NamingPolicyValidator((Source) obj));
     }
@@ -182,7 +188,7 @@ recognize.
 
 ```java
 Issues issues = new Issues("test");
-new MyValidator(testSource).validate(issues);
+new MyValidator(testSource).validate(issues, context);
 issues.close();
 assertThat(issues.valid()).isTrue();   // or assertThat(issues.toString()).contains("ERROR: ...")
 ```

docs/getting-started/architecture.md

@@ -138,6 +138,37 @@ Triggers (`TableTrigger`, `CronJob`) plug in here too — they let upstream
 events or schedules drive downstream side-effects without modeling them inside
 the pipeline.
 
+## Creating a table without SQL
+
+A single table (a `Source`) doesn't need a query to describe it — just a row
+type. For that case there's a SQL-free, **connection-free** entrypoint,
+`TableService.create(...)` (and `TableService.delete(...)`) in
+`hoptimator-jdbc`, which takes connection-level properties plus a table path and
+an **Avro schema** and runs the *same* validation and deployment as
+`CREATE TABLE` — the row type is derived from the Avro schema (via
+`AvroConverter`) instead of from SQL column declarations. Both paths converge on
+`HoptimatorDdlUtils.deployTableInternal`, so validators, deployers, and rollback
+behave identically regardless of where the schema came from.
+
+This is what lets a caller (e.g. a gRPC service) create a table by handing over
+a name and a schema, with no SQL parsing, no planning, and no JDBC connection. A
+`dryRun` flag mirrors `!specify` / the MCP `plan` tool: it validates and renders
+the specs without deploying anything.
+
+This works because the deploy/validate SPI is decoupled from Calcite. Providers
+(`Deployer`, `Validator`, `Connector`, `Config`) receive a neutral
+`DeploymentContext` — not a `java.sql.Connection` — that exposes only what they
+need: connection-level properties and per-`Database` config. The row type is
+resolved on demand via `HoptimatorDriver.rowType(source, context)`. The SQL path
+supplies a Calcite-backed `DeploymentContext` (row type read from the catalog);
+the direct path supplies a `DirectDeploymentContext` that carries the caller's
+Avro schema (deriving the row type on demand) and resolves `Database` config
+registry-natively via a `DatabaseConfigResolver`
+(the K8s implementation reads `Database` CRDs directly — see
+`DatabaseConfigResolvers`). The direct path opens no connection and touches no
+Calcite catalog; only the SQL engine's read/plan path still uses the JDBC
+driver layer.
+
 ## Module map
 
 The repo is split into focused modules. The ones you'll touch most often:

hoptimator-api/build.gradle

@@ -1,5 +1,5 @@
 plugins {
-  id 'java'
+  id 'java-library'
   id 'maven-publish'
 }
 

hoptimator-api/src/main/java/com/linkedin/hoptimator/ConfigProvider.java

@@ -1,9 +1,8 @@
 package com.linkedin.hoptimator;
 
-import java.sql.Connection;
 import java.util.Properties;
 
 public interface ConfigProvider {
 
-  Properties loadConfig(Connection connection) throws Exception;
+  Properties loadConfig(DeploymentContext context) throws Exception;
 }

hoptimator-api/src/main/java/com/linkedin/hoptimator/ConnectorProvider.java

@@ -1,11 +1,10 @@
 package com.linkedin.hoptimator;
 
-import java.sql.Connection;
 import java.util.Collection;
 
 
 public interface ConnectorProvider {
 
   /** Find connectors capable of configuring data plane connectors for the obj. */
-  <T> Collection<Connector> connectors(T obj, Connection connection);
+  <T> Collection<Connector> connectors(T obj, DeploymentContext context);
 }

hoptimator-api/src/main/java/com/linkedin/hoptimator/Deployer.java

@@ -13,6 +13,17 @@ public interface Deployer {
 
   void update() throws SQLException;
 
+  /**
+   * Whether the backing resource this deployer manages already exists. Used by the connection-free
+   * direct path to enforce {@code CREATE} (not {@code OR REPLACE}) semantics — the SQL/DDL path
+   * enforces the same thing via the Calcite catalog before deployers run, so this is not consulted
+   * there. Defaults to {@code false} (unknown/assume-absent) so deployers that do not implement it
+   * keep their existing behavior.
+   */
+  default boolean exists() throws SQLException {
+    return false;
+  }
+
   /** Render a list of specs, usually YAML. */
   List<String> specify() throws SQLException;
 

hoptimator-api/src/main/java/com/linkedin/hoptimator/DeployerProvider.java

@@ -1,13 +1,12 @@
 package com.linkedin.hoptimator;
 
-import java.sql.Connection;
 import java.util.Collection;
 
 
 public interface DeployerProvider {
 
   /** Find deployers capable of deploying the obj. */
-  <T extends Deployable> Collection<Deployer> deployers(T obj, Connection connection);
+  <T extends Deployable> Collection<Deployer> deployers(T obj, DeploymentContext context);
 
   /** A DeployerProvider with lower priority will execute first */
   int priority();

hoptimator-api/src/main/java/com/linkedin/hoptimator/DeploymentContext.java

@@ -0,0 +1,48 @@
+package com.linkedin.hoptimator;
+
+import java.util.Properties;
+
+import javax.annotation.Nullable;
+
+
+/**
+ * The ambient information a {@link Deployer}, {@link Connector}, or {@link Validator} needs in
+ * order to act on a {@link Deployable}, independent of how it was produced.
+ *
+ * <p>This replaces passing a raw {@code java.sql.Connection} through the SPI. The SQL path
+ * supplies a Calcite-backed implementation; a direct (non-SQL) caller supplies an
+ * implementation backed by its own control plane. Neither the deploy nor validate machinery
+ * needs to know which producer it came from.
+ *
+ * <p>A context exposes only two things:
+ * <ul>
+ *   <li>connection-level {@link #properties()} (namespace, hints, cluster config);
+ *   <li>per-{@code Database} connection config via {@link #databaseProperties}.
+ * </ul>
+ *
+ * <p>A table's row type is <em>not</em> exposed here (that would couple this API to a schema
+ * representation). It is resolved by the producer-specific machinery: the SQL path reads it from
+ * the Calcite catalog; the direct path carries it on its own context implementation.
+ */
+public interface DeploymentContext {
+
+  /** Connection-level properties and hints (e.g. namespace, {@code k8s.*}, hints, mode). */
+  Properties properties();
+
+  /**
+   * Returns the parsed connection properties for a database, extracted from its connection URL
+   * after stripping {@code connectionPrefix}, or {@code null} when the database is unknown or its
+   * URL does not start with {@code connectionPrefix}.
+   *
+   * <p>The {@code Database} it identifies is keyed by a catalog and/or a schema: catalog-style
+   * databases (e.g. MySQL) are matched by catalog, while schema-style databases (e.g. Kafka,
+   * Venice) are matched by schema. Both are individually optional — mirroring the {@code Database}
+   * CRD, where {@code catalog} and {@code schema} are optional — but at least one must be provided.
+   *
+   * @param catalog          the catalog name, or {@code null}
+   * @param schema           the schema name, or {@code null}
+   * @param connectionPrefix the expected URL scheme prefix (e.g. {@code "jdbc:kafka://"})
+   */
+  @Nullable Properties databaseProperties(@Nullable String catalog, @Nullable String schema,
+      String connectionPrefix);
+}

hoptimator-api/src/main/java/com/linkedin/hoptimator/Validated.java

@@ -1,14 +1,12 @@
 package com.linkedin.hoptimator;
 
-import java.sql.Connection;
-
 
 public interface Validated {
 
   /**
-   * Validates {@code this}, recording any problems in {@code issues}. The connection is always
+   * Validates {@code this}, recording any problems in {@code issues}. The context is always
    * supplied so validators can run lookups against external systems (e.g. pre-delete dependency
    * checks).
    */
-  void validate(Validator.Issues issues, Connection connection);
+  void validate(Validator.Issues issues, DeploymentContext context);
 }

hoptimator-api/src/main/java/com/linkedin/hoptimator/Validator.java

@@ -1,9 +1,8 @@
 package com.linkedin.hoptimator;
 
-import java.sql.Connection;
+import java.util.Collections;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -53,8 +52,8 @@ public DefaultValidator(T t) {
     }
 
     @Override
-    public void validate(Issues issues, Connection connection) {
-      t.validate(issues.child(t.getClass().getSimpleName()), connection);
+    public void validate(Issues issues, DeploymentContext context) {
+      t.validate(issues.child(t.getClass().getSimpleName()), context);
     }
   }
 

hoptimator-api/src/main/java/com/linkedin/hoptimator/ValidatorProvider.java

@@ -1,6 +1,5 @@
 package com.linkedin.hoptimator;
 
-import java.sql.Connection;
 import java.util.Collection;
 
 
@@ -9,5 +8,5 @@ public interface ValidatorProvider {
   /**
    * Returns validators that should be applied to {@code obj}.
    */
-  <T> Collection<Validator> validators(T obj, Connection connection);
+  <T> Collection<Validator> validators(T obj, DeploymentContext context);
 }