SQL-free direct table API (TableService) + lossless Avro preservation

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

@@ -0,0 +1,50 @@
+package com.linkedin.hoptimator.jdbc;
+
+import java.sql.SQLException;
+import java.util.List;
+import java.util.Properties;
+
+import javax.annotation.Nullable;
+
+
+/**
+ * Resolves per-{@code Database} connection config, decoupled from how the {@code Database} registry
+ * is stored. This is the seam that lets a {@link com.linkedin.hoptimator.DeploymentContext} answer
+ * {@code databaseProperties(...)} — and the direct table API resolve a database identifier —
+ * without holding a Calcite {@link HoptimatorConnection}. A registry-native module (e.g.
+ * {@code hoptimator-k8s}) supplies an implementation that reads {@code Database} CRDs (or any other
+ * registry) directly; see {@link DatabaseConfigResolverProvider}.
+ *
+ * <p>{@link #databaseProperties} deliberately mirrors {@link
+ * com.linkedin.hoptimator.DeploymentContext#databaseProperties}: on the direct path {@link
+ * DirectDeploymentContext} answers that SPI method by forwarding here, while {@link
+ * CalciteDeploymentContext} answers the same method from the Calcite catalog. They are separate
+ * interfaces (rather than one) because {@code DeploymentContext} is the engine-neutral SPI in
+ * {@code hoptimator-api} and must not depend on a registry-native module, whereas this resolver is
+ * the {@code ServiceLoader} backend that supplies the direct path's answer. This interface also
+ * carries {@link #databaseName}, which {@code DeploymentContext} does not — so the two are not
+ * interchangeable.
+ */
+public interface DatabaseConfigResolver {
+
+  /**
+   * Returns the parsed connection properties for a database, or {@code null} when the database is
+   * unknown or its URL does not start with {@code connectionPrefix}. The {@code Database} is keyed
+   * by a catalog and/or a schema; both are individually 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);
+
+  /**
+   * Resolves the {@code database} identifier for a table at {@code tablePath} — the value exposed as
+   * {@link com.linkedin.hoptimator.Source#database()} and used to name/derive deployed resources and
+   * to match template {@code databases} filters. This is the registered {@code Database} name for
+   * both schema-style and catalog-style databases; the store-level schema (e.g. the target MySQL
+   * database) is carried separately by {@link com.linkedin.hoptimator.Source#schema()}.
+   */
+  String databaseName(List<String> tablePath) throws SQLException;
+}

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

@@ -0,0 +1,27 @@
+package com.linkedin.hoptimator.jdbc;
+
+import java.util.Properties;
+
+
+/**
+ * Service-loaded factory for a {@link DatabaseConfigResolver}. This lets a registry-native module
+ * (e.g. {@code hoptimator-k8s}) supply a resolver that reads {@code Database} config directly from
+ * its source — inverting the dependency so both the SQL path and the connection-free direct path in
+ * {@code hoptimator-jdbc} can resolve config without a Calcite catalog.
+ *
+ * <p>The highest-{@link #priority()} provider wins. At least one provider must be registered; there
+ * is no built-in fallback, so a registry-native module must always be on the classpath.
+ */
+public interface DatabaseConfigResolverProvider {
+
+  /**
+   * Builds a resolver from connection-level properties (e.g. {@code k8s.*} config needed to reach
+   * the registry).
+   */
+  DatabaseConfigResolver resolver(Properties connectionProperties);
+
+  /** Higher wins. Default {@code 0}. */
+  default int priority() {
+    return 0;
+  }
+}

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

@@ -0,0 +1,29 @@
+package com.linkedin.hoptimator.jdbc;
+
+import java.util.Comparator;
+import java.util.Properties;
+import java.util.ServiceLoader;
+
+
+/** Discovers the registry-native {@link DatabaseConfigResolver} to use. */
+public final class DatabaseConfigResolvers {
+
+  private DatabaseConfigResolvers() {
+  }
+
+  /**
+   * Returns the highest-priority service-loaded {@link DatabaseConfigResolver}, built from raw
+   * connection properties. A registry-native provider — e.g. the {@code hoptimator-k8s} one — must
+   * be on the classpath; there is no Calcite-based fallback, so config resolution is identical on
+   * the SQL and connection-free direct paths.
+   */
+  public static DatabaseConfigResolver forProperties(Properties connectionProperties) {
+    return ServiceLoader.load(DatabaseConfigResolverProvider.class).stream()
+        .map(ServiceLoader.Provider::get)
+        .max(Comparator.comparingInt(DatabaseConfigResolverProvider::priority))
+        .map(provider -> provider.resolver(connectionProperties))
+        .orElseThrow(() -> new IllegalStateException(
+            "No DatabaseConfigResolverProvider registered; a registry-native resolver "
+                + "(e.g. hoptimator-k8s) must be on the classpath."));
+  }
+}

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

@@ -0,0 +1,85 @@
+package com.linkedin.hoptimator.jdbc;
+
+import com.linkedin.hoptimator.DeploymentContext;
+import com.linkedin.hoptimator.avro.AvroConverter;
+import org.apache.avro.Schema;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeSystem;
+import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
+
+import javax.annotation.Nullable;
+import java.util.Properties;
+
+
+/**
+ * A {@link DeploymentContext} for the direct (non-SQL) API path. Unlike
+ * {@link CalciteDeploymentContext}, it holds no Calcite {@link HoptimatorConnection}:
+ *
+ * <ul>
+ *   <li>the caller's original Avro schema is <em>carried</em> (rather than a table being looked up
+ *       from a Calcite {@code Table}); deployers that speak Avro (e.g. Venice) use
+ *       it verbatim via {@link #avroSchema()}, and {@link #rowType()} is derived from it on demand —
+ *       the schema is the single source of truth, so the two can't drift;
+ *   <li>connection-level {@link #properties()} are a plain bag;
+ *   <li>per-{@code Database} config is resolved through an injected {@link DatabaseConfigResolver},
+ *       so this context does not depend on how the {@code Database} registry is stored.
+ * </ul>
+ *
+ * <p>Because it is not a {@link CalciteDeploymentContext}, deployers cannot reach a
+ * {@code java.sql.Connection} through it -- the direct path stays decoupled from Calcite.
+ */
+public final class DirectDeploymentContext implements DeploymentContext {
+
+  private final Properties properties;
+  private final DatabaseConfigResolver databaseConfigResolver;
+  private final @Nullable Schema avroSchema;
+  private @Nullable RelDataType rowType;  // lazily derived from avroSchema
+
+  /** For schema-free operations such as delete, where no row type or Avro schema is needed. */
+  public DirectDeploymentContext(Properties properties, DatabaseConfigResolver databaseConfigResolver) {
+    this(properties, databaseConfigResolver, null);
+  }
+
+  public DirectDeploymentContext(Properties properties, DatabaseConfigResolver databaseConfigResolver,
+      @Nullable Schema avroSchema) {
+    this.properties = properties;
+    this.databaseConfigResolver = databaseConfigResolver;
+    this.avroSchema = avroSchema;
+  }
+
+  /**
+   * The row type for the table being deployed, derived on first use from the carried
+   * {@link #avroSchema()}. Throws for schema-free operations (e.g. a delete) that carry no schema.
+   */
+  public RelDataType rowType() {
+    if (avroSchema == null) {
+      throw new IllegalStateException("No Avro schema is carried by this context (e.g. a delete). "
+          + "A deployer requested a row type on a schema-free operation.");
+    }
+    if (rowType == null) {
+      rowType = AvroConverter.rel(avroSchema, new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT));
+    }
+    return rowType;
+  }
+
+  /**
+   * The caller's original (merged key+value) Avro schema for the table being deployed, or
+   * {@code null} when the caller supplied none (e.g. a delete). Deployers should prefer this over
+   * re-synthesizing Avro from {@link #rowType()}, since the row type cannot represent Avro
+   * namespaces, nested record identities, unions, or defaults.
+   */
+  public @Nullable Schema avroSchema() {
+    return avroSchema;
+  }
+
+  @Override
+  public Properties properties() {
+    return properties;
+  }
+
+  @Override
+  public @Nullable Properties databaseProperties(@Nullable String catalog, @Nullable String schema,
+      String connectionPrefix) {
+    return databaseConfigResolver.databaseProperties(catalog, schema, connectionPrefix);
+  }
+}

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

@@ -3,7 +3,9 @@
 import com.linkedin.hoptimator.Catalog;
 import com.linkedin.hoptimator.DeploymentContext;
 import com.linkedin.hoptimator.Source;
+import com.linkedin.hoptimator.avro.AvroConverter;
 import com.linkedin.hoptimator.avro.AvroSchemaSource;
+import com.linkedin.hoptimator.avro.AvroSchemas;
 import org.apache.avro.Schema;
 import org.apache.calcite.avatica.ConnectStringParser;
 import org.apache.calcite.jdbc.CalciteConnection;
@@ -23,6 +25,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
 import java.io.IOException;
 import java.sql.Connection;
 import java.sql.Driver;
@@ -166,6 +169,9 @@ public Connection connect(String url, Properties props) throws SQLException {
 
   public static RelDataType rowType(Source source, DeploymentContext context)
       throws SQLException {
+    if (context instanceof DirectDeploymentContext) {
+      return ((DirectDeploymentContext) context).rowType();
+    }
     if (context instanceof CalciteDeploymentContext) {
       HoptimatorConnection connection = ((CalciteDeploymentContext) context).connection();
       SchemaPlus schema = Objects.requireNonNull(connection.calciteConnection().getRootSchema());
@@ -184,13 +190,27 @@ public static RelDataType rowType(Source source, DeploymentContext context)
   }
 
   /**
-   * Returns the native value (payload) Avro schema for a table when it is backed by an
-   * {@link AvroSchemaSource} resolved through the Calcite catalog (e.g. a Venice store), or
-   * {@code null} otherwise (a SQL source with no native Avro, such as a MySQL table or a computed
-   * view). Callers rendering {@code {{avroValueSchema}}} then synthesize a value schema from the row
-   * type.
+   * Returns the value (payload) Avro schema for a table — the data record without any
+   * {@code KEY_}-prefixed key scaffolding — on either path, or {@code null} when none is available:
+   *
+   * <ul>
+   *   <li><b>Direct path</b> ({@link DirectDeploymentContext}): the value portion of the caller's
+   *       carried Avro schema, split off losslessly via {@link AvroConverter#valueSchemaOf}. The
+   *       caller handed us the exact schema, so we preserve its namespaces, nested record identities,
+   *       unions, and defaults instead of re-synthesizing from the flat row type.
+   *   <li><b>SQL path</b> ({@link CalciteDeploymentContext}): the native value schema of a
+   *       pre-existing source table that implements {@link AvroSchemaSource} (e.g. a Venice store
+   *       resolved through the Calcite catalog).
+   *   <li>Otherwise {@code null} — e.g. a delete (no schema carried), or a SQL source with no native
+   *       Avro (a MySQL table, a computed view). Callers rendering {@code {{avroValueSchema}}} then
+   *       synthesize a value schema from the row type.
+   * </ul>
    */
   public static Schema valueSchema(Source source, DeploymentContext context) {
+    if (context instanceof DirectDeploymentContext) {
+      Schema avroSchema = ((DirectDeploymentContext) context).avroSchema();
+      return avroSchema == null ? null : AvroConverter.valueSchemaOf(avroSchema, AvroSchemas.KEY_PREFIX);
+    }
     if (!(context instanceof CalciteDeploymentContext)) {
       return null;
     }
@@ -209,6 +229,22 @@ public static Schema valueSchema(Source source, DeploymentContext context) {
     return table instanceof AvroSchemaSource ? ((AvroSchemaSource) table).valueSchema() : null;
   }
 
+  /**
+   * Returns the caller-provided (merged key+value) Avro schema on the direct path, or {@code null}
+   * otherwise. On the direct API path the caller hands us the exact Avro schema they want deployed;
+   * carrying it verbatim lets deployers that speak Avro (e.g. Venice) avoid the
+   * lossy Avro->RelDataType->Avro round-trip, preserving namespaces,
+   * nested record names, unions, and defaults. Returns {@code null} on the SQL path (where the
+   * native schema, if any, comes from {@link #valueSchema}) and whenever no Avro was supplied
+   * (e.g. a delete).
+   */
+  public static @Nullable Schema providedAvroSchema(DeploymentContext context) {
+    if (context instanceof DirectDeploymentContext) {
+      return ((DirectDeploymentContext) context).avroSchema();
+    }
+    return null;
+  }
+
   private static final class ConnectionHolder {
     HoptimatorConnection connection;
   }

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

@@ -0,0 +1,120 @@
+package com.linkedin.hoptimator.jdbc;
+
+import com.linkedin.hoptimator.Deployer;
+import com.linkedin.hoptimator.DeploymentContext;
+import com.linkedin.hoptimator.PendingDelete;
+import com.linkedin.hoptimator.Source;
+import com.linkedin.hoptimator.util.DeploymentService;
+import org.apache.avro.Schema;
+
+import java.sql.SQLException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.Consumer;
+
+
+/**
+ * SQL-free entry point for creating and deleting a table (a {@link com.linkedin.hoptimator.Source})
+ * from a table path plus an Avro schema, without going through Calcite SQL parsing or DDL.
+ *
+ * <p>This is the programmatic counterpart to {@code CREATE TABLE} / {@code DROP TABLE}: the row
+ * type is derived from the supplied Avro schema (rather than SQL column declarations), but the
+ * table is validated and deployed through exactly the same {@code Validator} / {@code Deployer}
+ * SPI as the DDL path.
+ *
+ * <p>These entry points are <em>connection-free</em>: they take connection-level
+ * {@link Properties} (and, for create, log hooks) and resolve the {@code Database} registry and
+ * per-database config registry-natively (see {@link DatabaseConfigResolvers#forProperties}) — the
+ * direct API opens no JDBC {@link java.sql.Connection}.
+ */
+public final class TableService {
+
+  private TableService() {
+  }
+
+  /**
+   * Creates (or dry-run specifies) a table from an Avro schema, without a JDBC connection.
+   *
+   * @param connectionProperties connection-level properties (namespace, {@code k8s.*}, hints, mode)
+   * @param logHooks             sinks for human-readable deploy log lines; may be empty
+   * @param path                 the fully-qualified table path (e.g. {@code [DATABASE, TABLE]} or
+   *                             {@code [CATALOG, DATABASE, TABLE]})
+   * @param avroSchema           the Avro schema describing the table's row type
+   * @param options              table options (equivalent to DDL {@code WITH (...)}); may be empty
+   * @param updateIfExists       when {@code false}, creating a table that already
+   *                             exists fails. When {@code true}, an existing table is updated in
+   *                             place (schema evolution / config change), and a table that does not
+   *                             yet exist is still created. This flag is authoritative on the direct
+   *                             path and is not overridden by the connection's {@code mode}.
+   * @param dryRun               when {@code true}, validate and render specs without mutating
+   *                             anything (like {@code !specify} / the {@code Plan} RPC)
+   * @return the specs (populated only for dry-run), the resolved row type, and the table path
+   * @throws SQLException on validation or deployment errors
+   */
+  public static HoptimatorDdlUtils.SpecifyResult create(Properties connectionProperties,
+      List<Consumer<String>> logHooks, List<String> path, Schema avroSchema, Map<String, String> options,
+      boolean updateIfExists, boolean dryRun) throws SQLException {
+    if (path == null || path.size() < 2) {
+      throw new SQLException("A table path must include at least a database and a table name.");
+    }
+    if (avroSchema == null) {
+      throw new SQLException("An Avro schema is required to create a table.");
+    }
+    if (avroSchema.getType() != Schema.Type.RECORD) {
+      throw new SQLException("The Avro schema must be a record; got " + avroSchema.getType() + ".");
+    }
+    DatabaseConfigResolver resolver = DatabaseConfigResolvers.forProperties(connectionProperties);
+
+    // updateIfExists is authoritative for the direct path: it maps straight to CREATE (fail if the
+    // table already exists, enforced store-natively by the deployers) or UPDATE (create-or-update),
+    // independent of the connection's mode. Dry-run always resolves to SPECIFY.
+    HoptimatorDdlUtils.DdlMode mode = dryRun
+        ? HoptimatorDdlUtils.DdlMode.SPECIFY
+        : (updateIfExists ? HoptimatorDdlUtils.DdlMode.UPDATE : HoptimatorDdlUtils.DdlMode.CREATE);
+
+    // Resolve the target database identifier registry-natively and carry the caller's Avro schema on
+    // the context (the direct path touches no Calcite catalog); deployers derive the row type from it
+    // on demand, so we don't resolve it here.
+    String database = resolver.databaseName(path);
+    String tableName = path.get(path.size() - 1);
+    DirectDeploymentContext context = new DirectDeploymentContext(connectionProperties, resolver, avroSchema);
+
+    return HoptimatorDdlUtils.deployTableInternal(logHooks, context, null, path,
+        database, tableName, options, false, updateIfExists, mode);
+  }
+
+  /**
+   * Deletes a table, mirroring {@code DROP TABLE}, without a JDBC connection. Unlike {@link #create},
+   * this needs no schema — it resolves the {@link Source} from the path and runs the same pre-delete
+   * dependency guard and deployer teardown as the DDL path.
+   *
+   * @param connectionProperties connection-level properties (namespace, {@code k8s.*}, hints)
+   * @param path                 the fully-qualified table path (e.g. {@code [DATABASE, TABLE]})
+   * @throws SQLException on validation or teardown errors
+   */
+  public static void delete(Properties connectionProperties, List<String> path) throws SQLException {
+    if (path == null || path.size() < 2) {
+      throw new SQLException("A table path must include at least a database and a table name.");
+    }
+
+    DatabaseConfigResolver resolver = DatabaseConfigResolvers.forProperties(connectionProperties);
+    String database = resolver.databaseName(path);
+    Source source = new Source(database, path, Map.of());
+    DeploymentContext context = new DirectDeploymentContext(connectionProperties, resolver);
+
+    Collection<Deployer> deployers = null;
+    try {
+      // Pre-delete dependency guard (mirrors the DDL DROP path).
+      ValidationService.validateOrThrow(new PendingDelete<>(source), context);
+      deployers = DeploymentService.deployers(source, context);
+      DeploymentService.delete(deployers);
+    } catch (SQLException | RuntimeException e) {
+      if (deployers != null) {
+        DeploymentService.restore(deployers);
+      }
+      throw e;
+    }
+  }
+}